| # Licensed to the Apache Software Foundation (ASF) under one or more | |
| # contributor license agreements. See the NOTICE file distributed with | |
| # this work for additional information regarding copyright ownership. | |
| # The ASF licenses this file to You under the Apache License, Version 2.0 | |
| # (the "License"); you may not use this file except in compliance with | |
| # the License. You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| import google_crc32c | |
| import logging | |
| import time | |
| from datetime import datetime, timezone, timedelta | |
| from google.cloud import secretmanager | |
| from typing import List, Union, Tuple, Dict | |
| # What the "created_by" label is set to for secrets created by this service. | |
| SECRET_MANAGER_LABEL = "beam-infra-secret-manager" | |
| class SecretManagerLoggerAdapter(logging.LoggerAdapter): | |
| """Logger adapter that adds a prefix to all log messages.""" | |
| def process(self, msg, kwargs): | |
| return f"[SecretManager] {msg}", kwargs | |
| class SecretManager: | |
| """Service to manage GCP API keys rotation.""" | |
| project_id: str # The GCP project ID where secrets are managed | |
| rotation_interval: int # The interval (in days) at which to rotate secrets | |
| grace_period: int # The grace period (in days) before a secret is considered for rotation | |
| max_retries: int # The maximum number of retries for API calls | |
| client: secretmanager.SecretManagerServiceClient # GCP Secret Manager client | |
| logger: Union[logging.Logger, logging.LoggerAdapter] # Logger for logging messages | |
| def __init__(self, project_id: str, logger: logging.Logger, rotation_interval: int = 30, grace_period: int = 7, max_retries: int = 3) -> None: | |
| self.project_id = project_id | |
| self.rotation_interval = rotation_interval | |
| self.grace_period = grace_period | |
| self.max_retries = max_retries | |
| self.client = secretmanager.SecretManagerServiceClient() | |
| self.logger = SecretManagerLoggerAdapter(logger, {}) | |
| self.logger.info(f"Initialized SecretManager for project '{self.project_id}'") | |
| def _get_secret_ids(self) -> List[str]: | |
| """ | |
| Retrieves the list of secrets from the Secret Manager and populates the `secrets_ids` list. | |
| This method filters secrets based on a specific label indicating they were created by this service. | |
| Returns: | |
| List[str]: A list of secret IDs that were created by this service. | |
| """ | |
| self.logger.debug(f"Retrieving secrets with the label from project '{self.project_id}'") | |
| secret_ids = [] | |
| try: | |
| for secret in self.client.list_secrets(request={"parent": f"projects/{self.project_id}"}): | |
| secret_id = secret.name.split("/")[-1] | |
| if "created_by" in secret.labels and secret.labels["created_by"] == SECRET_MANAGER_LABEL: | |
| secret_ids.append(secret_id) | |
| except Exception as e: | |
| self.logger.error(f"Error retrieving secrets: {e}") | |
| self.logger.debug(f"Found {len(secret_ids)} secrets created by {SECRET_MANAGER_LABEL} in project '{self.project_id}'") | |
| return secret_ids | |
| def _secret_exists(self, secret_id: str) -> bool: | |
| """ | |
| Checks if a secret with the given ID exists. | |
| Args: | |
| secret_id (str): The ID of the secret to check. | |
| Returns: | |
| bool: True if the secret exists, False otherwise. | |
| """ | |
| self.logger.debug(f"Checking if secret '{secret_id}' exists") | |
| try: | |
| name = self.client.secret_path(self.project_id, secret_id) | |
| self.client.get_secret(request={"name": name}) | |
| self.logger.debug(f"Secret '{secret_id}' exists") | |
| return True | |
| except Exception as e: | |
| self.logger.debug(f"Secret '{secret_id}' does not exist: {e}") | |
| return False | |
| def _secret_is_managed(self, secret_id: str) -> bool: | |
| """ | |
| Checks if a secret with the given ID exists and is managed by this service. | |
| Args: | |
| secret_id (str): The ID of the secret to check. | |
| Returns: | |
| bool: True if the secret is managed by this service, False otherwise. | |
| """ | |
| self.logger.debug(f"Checking if secret '{secret_id}' exists and is managed by {SECRET_MANAGER_LABEL}") | |
| if not self._secret_exists(secret_id): | |
| self.logger.debug(f"Secret '{secret_id}' does not exist, cannot be managed") | |
| return False | |
| name = self.client.secret_path(self.project_id, secret_id) | |
| secret = self.client.get_secret(request={"name": name}) | |
| is_managed = "created_by" in secret.labels and secret.labels["created_by"] == SECRET_MANAGER_LABEL | |
| self.logger.debug(f"Secret '{secret_id}' is managed by {SECRET_MANAGER_LABEL}: {is_managed}") | |
| return is_managed | |
| def create_secret(self, secret_id: str) -> str: | |
| """ | |
| Create a new secret with the given name. A secret is a logical wrapper | |
| around a collection of secret versions. Secret versions hold the actual | |
| secret material. This method creates a new secret with automatic replication | |
| and labels for tracking. | |
| Args: | |
| secret_id (str): The ID to assign to the new secret. This ID must be unique within the project. | |
| Returns: | |
| str: The secret path of the newly created secret. | |
| """ | |
| if self._secret_is_managed(secret_id): | |
| self.logger.debug(f"Secret '{secret_id}' already exists, returning existing secret path") | |
| name = self.client.secret_path(self.project_id, secret_id) | |
| return name | |
| self.logger.info(f"Creating new secret '{secret_id}' with rotation interval of {self.rotation_interval} days") | |
| response = self.client.create_secret( | |
| request={ | |
| "parent": f"projects/{self.project_id}", | |
| "secret_id": f"{secret_id}", | |
| "secret": { | |
| "replication": { | |
| "automatic": {} | |
| }, | |
| "labels": { | |
| "created_by": SECRET_MANAGER_LABEL, | |
| "created_at": datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S"), | |
| "rotation_interval_days": str(self.rotation_interval), | |
| "grace_period_days": str(self.grace_period), | |
| "last_version_created_at": datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S"), | |
| } | |
| } | |
| } | |
| ) | |
| # created_by : This label is used to identify secrets created by this service. | |
| # created_at : This label stores the timestamp when the secret was created. | |
| # rotation_interval_days : This label specifies the rotation interval for the secret. | |
| # grace_period_days : This label specifies the grace period for the secret. | |
| # last_version_created_at : This label stores the timestamp when the last version of the secret was created, this | |
| # helps with the rotation and grace period calculations. | |
| # Wait for the secret to be created | |
| self.logger.debug(f"Waiting for secret '{secret_id}' to be created") | |
| delay = 1 | |
| for _ in range(self.max_retries): | |
| if self._secret_is_managed(secret_id): | |
| self.logger.debug(f"Secret '{secret_id}' is now available") | |
| break | |
| self.logger.debug(f"Secret '{secret_id}' not found, retrying in {delay} seconds") | |
| time.sleep(delay) | |
| delay *= 2 | |
| else: | |
| error_msg = f"Could not verify creation of secret '{secret_id}' after {self.max_retries} retries." | |
| self.logger.error(error_msg) | |
| raise RuntimeError(error_msg) | |
| self.logger.info(f"Successfully created secret '{secret_id}'") | |
| return response.name | |
| def get_secret(self, secret_id: str) -> secretmanager.Secret: | |
| """ | |
| Retrieves the specified secret by its ID. | |
| Args: | |
| secret_id (str): The ID of the secret to retrieve. | |
| Returns: | |
| secretmanager.Secret: The requested secret. | |
| """ | |
| self.logger.info(f"Retrieving secret '{secret_id}'") | |
| if not self._secret_exists(secret_id): | |
| error_msg = f"Secret {secret_id} does not exist. Please create it first." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| if not self._secret_is_managed(secret_id): | |
| error_msg = f"Secret {secret_id} is not managed by this service." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| name = self.client.secret_path(self.project_id, secret_id) | |
| return self.client.get_secret(request={"name": name}) | |
| def delete_secret(self, secret_id: str) -> None: | |
| """ | |
| Deletes the specified secret and all its versions. | |
| Args: | |
| secret_id (str): The ID of the secret to delete. | |
| """ | |
| if not self._secret_is_managed(secret_id): | |
| self.logger.debug(f"Secret '{secret_id}' is not managed by this service, cannot delete") | |
| return | |
| self.logger.info(f"Deleting secret '{secret_id}' and all its versions") | |
| name = self.client.secret_path(self.project_id, secret_id) | |
| self.client.delete_secret(request={"name": name}) | |
| # Wait for the secret to be deleted | |
| self.logger.debug(f"Waiting for secret '{secret_id}' to be deleted") | |
| delay = 1 | |
| for _ in range(self.max_retries): | |
| if not self._secret_exists(secret_id): | |
| self.logger.debug(f"Secret '{secret_id}' is now deleted") | |
| break | |
| self.logger.debug(f"Secret '{secret_id}' still exists, retrying in {delay} seconds") | |
| time.sleep(delay) | |
| delay *= 2 | |
| else: | |
| error_msg = f"Could not verify deletion of secret '{secret_id}' after {self.max_retries} retries." | |
| self.logger.error(error_msg) | |
| raise RuntimeError(error_msg) | |
| self.logger.info(f"Successfully deleted secret '{secret_id}'") | |
| def is_different_user_access(self, secret_id: str, allowed_users: List[str]) -> bool: | |
| """ | |
| Checks if the current access policy of a secret allows only the specified users to read it. | |
| This is used to determine if an update is needed. | |
| Args: | |
| secret_id (str): The ID of the secret to check access for. | |
| allowed_users (List[str]): A list of user emails to check against the current access policy. | |
| Returns: | |
| bool: True if the current access policy is different from the specified users, False otherwise. | |
| """ | |
| self.logger.debug(f"Checking if access for secret '{secret_id}' differs from allowed users: {allowed_users}") | |
| if not self._secret_is_managed(secret_id): | |
| self.logger.debug(f"Secret '{secret_id}' is not managed by this service, cannot check access") | |
| return True | |
| accessor_role = "roles/secretmanager.secretAccessor" | |
| resource_name = self.client.secret_path(self.project_id, secret_id) | |
| try: | |
| policy = self.client.get_iam_policy(request={"resource": resource_name}) | |
| except Exception as e: | |
| self.logger.error(f"Failed to get IAM policy for secret '{secret_id}': {e}") | |
| return True | |
| current_members = set() | |
| for binding in policy.bindings: | |
| if binding.role == accessor_role: | |
| current_members.update(binding.members) | |
| allowed_members = {f"user:{user_email}" for user_email in allowed_users} | |
| is_different = current_members != allowed_members | |
| self.logger.debug(f"Current members: {current_members}") | |
| self.logger.debug(f"Allowed members: {allowed_members}") | |
| self.logger.debug(f"Access for secret '{secret_id}' differs: {is_different}") | |
| return is_different | |
| def update_secret_access(self, secret_id: str, allowed_users: List[str]) -> None: | |
| """ | |
| Updates the access policy of a secret to allow only the specified users to read it. | |
| Any existing users will be removed and replaced with the new list. | |
| Args: | |
| secret_id (str): The ID of the secret to update access for. | |
| allowed_users (List[str]): A list of user emails to grant read access to. | |
| """ | |
| self.logger.debug(f"Updating access for secret '{secret_id}' to allow users: {allowed_users}") | |
| if not self._secret_is_managed(secret_id): | |
| error_msg = f"Secret {secret_id} is not managed by this service, cannot update access." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| accessor_role = "roles/secretmanager.secretAccessor" | |
| resource_name = self.client.secret_path(self.project_id, secret_id) | |
| policy = self.client.get_iam_policy(request={"resource": resource_name}) | |
| members = [f"user:{user_email}" for user_email in allowed_users] | |
| binding_found = False | |
| for binding in policy.bindings: | |
| if binding.role == accessor_role: | |
| binding.members[:] = members | |
| self.logger.debug(f"Replaced members for role '{accessor_role}' in secret '{secret_id}' with: {allowed_users}") | |
| binding_found = True | |
| break | |
| if not binding_found: | |
| policy.bindings.add( | |
| role=accessor_role, | |
| members=members | |
| ) | |
| self.logger.debug(f"Created new binding for role '{accessor_role}' in secret '{secret_id}'") | |
| self.client.set_iam_policy( | |
| request={ | |
| "resource": resource_name, | |
| "policy": policy | |
| } | |
| ) | |
| self.logger.info(f"Successfully updated access for secret '{secret_id}' to allow users: {allowed_users}") | |
| def _get_secret_versions(self, secret_id: str) -> List[secretmanager.SecretVersion]: | |
| """ | |
| Retrieves all versions of a secret. | |
| Args: | |
| secret_id (str): The ID of the secret to list versions for. | |
| Returns: | |
| List[secretmanager.SecretVersion]: A list of secret versions. | |
| """ | |
| self.logger.debug(f"Retrieving versions for secret '{secret_id}'") | |
| if not self._secret_is_managed(secret_id): | |
| self.logger.debug(f"Secret '{secret_id}' is not managed by this service, cannot retrieve versions") | |
| return [] | |
| parent = self.client.secret_path(self.project_id, secret_id) | |
| versions = list(self.client.list_secret_versions(request={"parent": parent})) | |
| self.logger.debug(f"Found {len(versions)} versions for secret '{secret_id}'") | |
| return versions | |
| def _secret_version_exists(self, secret_id: str, version_id: str) -> bool: | |
| """ | |
| Checks if a specific version of a secret exists. | |
| Args: | |
| secret_id (str): The ID of the secret to check. | |
| version_id (str): The ID of the version to check. | |
| Returns: | |
| bool: True if the version exists, False otherwise. | |
| """ | |
| self.logger.debug(f"Checking if version '{version_id}' exists for secret '{secret_id}'") | |
| if not self._secret_is_managed(secret_id): | |
| self.logger.debug(f"Secret '{secret_id}' is not managed by this service, cannot check version existence") | |
| return False | |
| versions = self._get_secret_versions(secret_id) | |
| exists = any(version.name.split("/")[-1] == version_id for version in versions) | |
| self.logger.debug(f"Version '{version_id}' exists: {exists}") | |
| return exists | |
| def _secret_version_is_enabled(self, secret_id: str, version_id: str) -> bool: | |
| """ | |
| Checks if a specific version of a secret is enabled. | |
| Args: | |
| secret_id (str): The ID of the secret to check. | |
| version_id (str): The ID of the version to check. | |
| Returns: | |
| bool: True if the version is enabled, False otherwise. | |
| """ | |
| self.logger.debug(f"Checking if version '{version_id}' of secret '{secret_id}' is enabled") | |
| if not self._secret_is_managed(secret_id): | |
| self.logger.debug(f"Secret '{secret_id}' is not managed by this service, version cannot be enabled") | |
| return False | |
| versions = self._get_secret_versions(secret_id) | |
| for version in versions: | |
| if version.name.split("/")[-1] == version_id: | |
| is_enabled = version.state == secretmanager.SecretVersion.State.ENABLED | |
| self.logger.debug(f"Version '{version_id}' is enabled: {is_enabled}") | |
| return is_enabled | |
| self.logger.debug(f"Version '{version_id}' does not exist for secret '{secret_id}'") | |
| return False | |
| def _secret_version_is_destroyed(self, secret_id: str, version_id: str) -> bool: | |
| """ | |
| Checks if a specific version of a secret is destroyed. | |
| Args: | |
| secret_id (str): The ID of the secret to check. | |
| version_id (str): The ID of the version to check. | |
| Returns: | |
| bool: True if the version is destroyed, False otherwise. | |
| """ | |
| self.logger.debug(f"Checking if version '{version_id}' of secret '{secret_id}' is destroyed") | |
| if not self._secret_is_managed(secret_id): | |
| self.logger.debug(f"Secret '{secret_id}' is not managed by this service, version cannot be destroyed") | |
| return False | |
| versions = self._get_secret_versions(secret_id) | |
| for version in versions: | |
| if version.name.split("/")[-1] == version_id: | |
| is_destroyed = version.state == secretmanager.SecretVersion.State.DESTROYED | |
| self.logger.debug(f"Version '{version_id}' is destroyed: {is_destroyed}") | |
| return is_destroyed | |
| self.logger.debug(f"Version '{version_id}' does not exist for secret '{secret_id}'") | |
| return False | |
| def _get_latest_secret_version_id(self, secret_id: str) -> str: | |
| """ | |
| Retrieves the latest enabled version of a secret. | |
| Args: | |
| secret_id (str): The ID of the secret to retrieve the latest version for. | |
| Returns: | |
| str: The name of the latest secret version. | |
| """ | |
| self.logger.debug(f"Retrieving latest enabled version of secret '{secret_id}'") | |
| if not self._secret_is_managed(secret_id): | |
| error_msg = f"Secret {secret_id} does not exist or is not managed by this service, cannot retrieve latest version." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| for version in self._get_secret_versions(secret_id): | |
| if version.state == secretmanager.SecretVersion.State.ENABLED: | |
| version_id = version.name.split("/")[-1] | |
| self.logger.debug(f"Found latest enabled version '{version_id}' for secret '{secret_id}'") | |
| return version_id | |
| error_msg = f"No enabled versions found for secret {secret_id}." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| def _is_key_rotation_due(self, secret_id: str) -> bool: | |
| """ | |
| Checks if the key rotation is due based on the last version created timestamp. | |
| Args: | |
| secret_id (str): The ID of the secret to check. | |
| Returns: | |
| bool: True if the key rotation is due, False otherwise. | |
| """ | |
| self.logger.debug(f"Checking if key rotation is due for secret '{secret_id}'") | |
| if not self._secret_is_managed(secret_id): | |
| error_msg = f"Secret {secret_id} does not exist or is not managed by this service, cannot check rotation." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| secret = self.get_secret(secret_id) | |
| last_version_created_at = secret.labels["last_version_created_at"] | |
| last_version_date = datetime.strptime(last_version_created_at, "%Y%m%d_%H%M%S").replace(tzinfo=timezone.utc) | |
| due_date = last_version_date + timedelta(days=self.rotation_interval) | |
| is_due = datetime.now(timezone.utc) >= due_date | |
| self.logger.debug(f"Key rotation due for secret '{secret_id}': {is_due}") | |
| return is_due | |
| def add_secret_version(self, secret_id: str, data_id: str, payload: Union[bytes, str]) -> str: | |
| """ | |
| Adds a new version to the specified secret with the given data ID and payload. | |
| If the secret does not exist, it will be created first. All previous versions will be disabled. | |
| Args: | |
| secret_id (str): The ID of the secret to which the version will be added. | |
| data_id (str): The ID of the data to be stored in the new version. | |
| payload (bytes): The secret data to be stored in the new version. | |
| Returns: | |
| str: The name of the newly created secret version. | |
| """ | |
| self.logger.info(f"Adding new version to secret '{secret_id}'") | |
| secret_path = self.create_secret(secret_id) | |
| if not isinstance(payload, (bytes, str)): | |
| error_msg = "Payload must be a bytes object or a string that can be encoded to bytes." | |
| self.logger.error(error_msg) | |
| raise TypeError(error_msg) | |
| # Join data_id and payload to form the payload | |
| if isinstance(payload, str): | |
| payload = f"{data_id}:{payload}" | |
| else: | |
| payload = f"{data_id}:{payload.decode('utf-8')}" if isinstance(payload, bytes) else payload | |
| # Ensure payload is bytes | |
| payload_bytes = payload.encode('utf-8') if isinstance(payload, str) else payload | |
| crc32c = google_crc32c.Checksum() | |
| crc32c.update(payload_bytes) | |
| self.logger.debug(f"Creating secret version with CRC32C checksum") | |
| response = self.client.add_secret_version( | |
| request={ | |
| "parent": secret_path, | |
| "payload": { | |
| "data": payload_bytes, | |
| "data_crc32c": int(crc32c.hexdigest(), 16), | |
| } | |
| } | |
| ) | |
| version_id = response.name.split("/")[-1] | |
| # Update the last version created timestamp | |
| self.logger.debug(f"Updating last version created timestamp for secret '{secret_id}'") | |
| secret_obj = self.get_secret(secret_id) | |
| labels = dict(secret_obj.labels) | |
| labels["last_version_created_at"] = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") | |
| secret = {"name": secret_obj.name, "labels": labels} | |
| update_mask = {"paths": ["labels"]} | |
| self.client.update_secret(request={"secret": secret, "update_mask": update_mask}) | |
| # Wait for the new version to be available | |
| self.logger.debug(f"Waiting for new version '{version_id}' of secret '{secret_id}' to be available") | |
| delay = 1 | |
| for _ in range(self.max_retries): | |
| if self._secret_version_exists(secret_id, version_id): | |
| self.logger.debug(f"Version '{version_id}' of secret '{secret_id}' is now available") | |
| break | |
| self.logger.debug(f"Version '{version_id}' of secret '{secret_id}' not found, retrying in {delay} seconds") | |
| time.sleep(delay) | |
| delay *= 2 | |
| else: | |
| error_msg = f"Could not verify creation of secret version '{version_id}' after {self.max_retries} retries." | |
| self.logger.error(error_msg) | |
| raise RuntimeError(error_msg) | |
| # Disable all the previous versions except the newly created one | |
| for ver in self._get_secret_versions(secret_id): | |
| if ver.name != response.name and ver.state == secretmanager.SecretVersion.State.ENABLED: | |
| self.logger.debug(f"Disabling previous version '{ver.name}' of secret '{secret_id}'") | |
| self.disable_secret_version(secret_id, ver.name.split("/")[-1]) | |
| self.logger.info(f"Successfully added version '{version_id}' to secret '{secret_id}'") | |
| return response.name | |
| def get_latest_secret_version(self, secret_id: str) -> Tuple[str, bytes]: | |
| """ | |
| Retrieves the latest enabled version of a secret. | |
| Args: | |
| secret_id (str): The ID of the secret from which to retrieve the version. | |
| Returns: | |
| Tuple[str, bytes]: A tuple containing the data ID and the payload of the latest secret version. | |
| """ | |
| self.logger.info(f"Retrieving latest version of secret '{secret_id}'") | |
| if not self._secret_is_managed(secret_id): | |
| error_msg = f"Secret {secret_id} does not exist or is not managed by this service, cannot retrieve latest version." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| version_id = self._get_latest_secret_version_id(secret_id) | |
| name = f"projects/{self.project_id}/secrets/{secret_id}/versions/{version_id}" | |
| self.logger.debug(f"Accessing secret version '{version_id}' of secret '{secret_id}'") | |
| response = self.client.access_secret_version(request={"name": name}) | |
| crc32c = google_crc32c.Checksum() | |
| crc32c.update(response.payload.data) | |
| if int(crc32c.hexdigest(), 16) != response.payload.data_crc32c: | |
| error_msg = "CRC32C checksum mismatch. The data may be corrupted." | |
| self.logger.error(f"{error_msg} for secret '{secret_id}' version '{version_id}'") | |
| raise ValueError(error_msg) | |
| self.logger.info(f"Successfully retrieved version '{version_id}' of secret '{secret_id}'") | |
| data_str = response.payload.data.decode('utf-8') | |
| data_id, payload = data_str.split(":", 1) | |
| return data_id, payload.encode('utf-8') | |
| def enable_secret_version(self, secret_id: str, version_id: str) -> None: | |
| """ | |
| Enables a specific version of a secret. | |
| Args: | |
| secret_id (str): The ID of the secret from which to enable the version. | |
| version_id (str): The version ID to enable. | |
| """ | |
| self.logger.info(f"Enabling version '{version_id}' of secret '{secret_id}'") | |
| if not self._secret_is_managed(secret_id): | |
| error_msg = f"Secret {secret_id} does not exist or is not managed by this service, cannot enable version." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| self.logger.debug(f"Verifying version '{version_id}' exists for secret '{secret_id}'") | |
| version_exists = any( | |
| version.name.split("/")[-1] == version_id and version.state == secretmanager.SecretVersion.State.DISABLED | |
| for version in self._get_secret_versions(secret_id) | |
| ) | |
| if not version_exists: | |
| error_msg = f"Version {version_id} does not exist or is not disabled for secret {secret_id}." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| name = f"projects/{self.project_id}/secrets/{secret_id}/versions/{version_id}" | |
| self.logger.debug(f"Enabling version '{version_id}' of secret '{secret_id}'") | |
| response = self.client.enable_secret_version(request={"name": name}) | |
| if response.name.split("/")[-1] != version_id or response.state != secretmanager.SecretVersion.State.ENABLED: | |
| error_msg = f"Failed to enable secret version {version_id} for secret {secret_id}." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| # Wait for the version to be enabled | |
| self.logger.debug(f"Waiting for version '{version_id}' of secret '{secret_id}' to be enabled") | |
| delay = 1 | |
| for _ in range(self.max_retries): | |
| if self._secret_version_is_enabled(secret_id, version_id): | |
| self.logger.debug(f"Version '{version_id}' of secret '{secret_id}' is now enabled") | |
| break | |
| self.logger.debug(f"Version '{version_id}' of secret '{secret_id}' still disabled, retrying in {delay} seconds") | |
| time.sleep(delay) | |
| delay *= 2 | |
| else: | |
| error_msg = f"Could not verify enabling of version '{version_id}' of secret '{secret_id}' after {self.max_retries} retries." | |
| self.logger.error(error_msg) | |
| raise RuntimeError(error_msg) | |
| self.logger.info(f"Successfully enabled version '{version_id}' of secret '{secret_id}'") | |
| def disable_secret_version(self, secret_id: str, version_id: str) -> None: | |
| """ | |
| Disables a specific version of a secret. | |
| Args: | |
| secret_id (str): The ID of the secret from which to delete the version. | |
| version_id (str): The version ID to delete. | |
| """ | |
| self.logger.info(f"Disabling version '{version_id}' of secret '{secret_id}'") | |
| if not self._secret_is_managed(secret_id): | |
| error_msg = f"Secret {secret_id} does not exist or is not managed by this service, cannot disable version." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| self.logger.debug(f"Verifying version '{version_id}' exists for secret '{secret_id}'") | |
| version_exists = any( | |
| version.name.split("/")[-1] == version_id | |
| for version in self._get_secret_versions(secret_id) | |
| ) | |
| if not version_exists: | |
| error_msg = f"Version {version_id} does not exist for secret {secret_id}." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| name = f"projects/{self.project_id}/secrets/{secret_id}/versions/{version_id}" | |
| self.logger.debug(f"Disabling version '{version_id}' of secret '{secret_id}'") | |
| response = self.client.disable_secret_version(request={"name": name}) | |
| if response.name.split("/")[-1] != version_id or response.state != secretmanager.SecretVersion.State.DISABLED: | |
| error_msg = f"Failed to disable secret version {version_id} for secret {secret_id}." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| # Wait for the version to be disabled | |
| self.logger.debug(f"Waiting for version '{version_id}' of secret '{secret_id}' to be disabled") | |
| delay = 1 | |
| for _ in range(self.max_retries): | |
| if not self._secret_version_is_enabled(secret_id, version_id): | |
| self.logger.debug(f"Version '{version_id}' of secret '{secret_id}' is now disabled") | |
| break | |
| self.logger.debug(f"Version '{version_id}' of secret '{secret_id}' still enabled, retrying in {delay} seconds") | |
| time.sleep(delay) | |
| delay *= 2 | |
| else: | |
| error_msg = f"Could not verify disabling of version '{version_id}' of secret '{secret_id}' after {self.max_retries} retries." | |
| self.logger.error(error_msg) | |
| raise RuntimeError(error_msg) | |
| self.logger.info(f"Successfully disabled version '{version_id}' of secret '{secret_id}'") | |
| def destroy_secret_version(self, secret_id: str, version_id: str) -> str: | |
| """ | |
| Destroys a specific version of a secret. | |
| Args: | |
| secret_id (str): The ID of the secret from which to delete the version. | |
| version_id (str): The version ID to delete. | |
| Returns: | |
| str: The data ID of the destroyed version. | |
| """ | |
| self.logger.info(f"Destroying version '{version_id}' of secret '{secret_id}'") | |
| if not self._secret_is_managed(secret_id): | |
| error_msg = f"Secret {secret_id} does not exist or is not managed by this service, cannot destroy version." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| self.logger.debug(f"Verifying version '{version_id}' exists for secret '{secret_id}'") | |
| version_exists = any( | |
| version.name.split("/")[-1] == version_id | |
| for version in self._get_secret_versions(secret_id) | |
| ) | |
| if not version_exists: | |
| error_msg = f"Version {version_id} does not exist for secret {secret_id}." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| # Enable the version before destroying it to get the data ID | |
| if not self._secret_version_is_enabled(secret_id, version_id): | |
| self.logger.debug(f"Version '{version_id}' of secret '{secret_id}' is not enabled, enabling it before destruction") | |
| self.enable_secret_version(secret_id, version_id) | |
| # Get the data ID from the specific version we're about to destroy | |
| name = f"projects/{self.project_id}/secrets/{secret_id}/versions/{version_id}" | |
| response = self.client.access_secret_version(request={"name": name}) | |
| data_str = response.payload.data.decode('utf-8') | |
| data_id, _ = data_str.split(":", 1) | |
| self.logger.debug(f"Data ID for version '{version_id}' of secret '{secret_id}': {data_id}") | |
| # Now destroy the version | |
| self.logger.debug(f"Destroying version '{version_id}' of secret '{secret_id}'") | |
| response = self.client.destroy_secret_version(request={"name": name}) | |
| if response.name.split("/")[-1] != version_id or response.state != secretmanager.SecretVersion.State.DESTROYED: | |
| error_msg = f"Failed to destroy secret version {version_id} for secret {secret_id}." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| # Wait for the version to be destroyed | |
| self.logger.debug(f"Waiting for version '{version_id}' of secret '{secret_id}' to be destroyed") | |
| delay = 1 | |
| for _ in range(self.max_retries): | |
| if self._secret_version_is_destroyed(secret_id, version_id): | |
| self.logger.debug(f"Version '{version_id}' of secret '{secret_id}' is now destroyed") | |
| break | |
| self.logger.debug(f"Version '{version_id}' of secret '{secret_id}' still not destroyed, retrying in {delay} seconds") | |
| time.sleep(delay) | |
| delay *= 2 | |
| else: | |
| error_msg = f"Could not verify destruction of version '{version_id}' of secret '{secret_id}' after {self.max_retries} retries." | |
| self.logger.error(error_msg) | |
| raise RuntimeError(error_msg) | |
| self.logger.info(f"Successfully destroyed version '{version_id}' of secret '{secret_id}'") | |
| return data_id | |
| def purge_disabled_secret_versions(self, secret_id: str) -> List[str]: | |
| """ | |
| Purges (destroys) all disabled versions of a secret that are older than the grace period. | |
| To determine if a version is older than the grace period, it checks the creation time of each version, | |
| if the latest version was created more than the grace period ago, it will purge the disabled versions. | |
| Args: | |
| secret_id (str): The ID of the secret for which to purge disabled versions. | |
| Returns: | |
| List[str]: A list of data IDs of the destroyed versions. | |
| """ | |
| self.logger.info(f"Purging disabled versions of secret '{secret_id}'") | |
| if not self._secret_is_managed(secret_id): | |
| error_msg = f"Secret {secret_id} does not exist or is not managed by this service, cannot purge versions." | |
| self.logger.error(error_msg) | |
| raise ValueError(error_msg) | |
| data_ids = [] | |
| for version in self._get_secret_versions(secret_id): | |
| if version.state == secretmanager.SecretVersion.State.DISABLED: | |
| version_id = version.name.split("/")[-1] | |
| create_time = datetime.fromtimestamp(version.create_time.timestamp(), tz=timezone.utc) # type: ignore | |
| if create_time < datetime.now(timezone.utc) - timedelta(days=self.grace_period): | |
| self.logger.debug(f"Destroying disabled version '{version_id}' of secret '{secret_id}'") | |
| data_ids.append(self.destroy_secret_version(secret_id, version_id)) | |
| else: | |
| self.logger.debug(f"Skipping version '{version_id}' of secret '{secret_id}' as it is within the grace period") | |
| return data_ids | |
| def cron(self) -> Dict[str, List[str]]: | |
| """ | |
| Performs periodic maintenance tasks: | |
| - Purges disabled secret versions that are older than the grace period. | |
| Returns: | |
| Dict[str, List[str]]: A dictionary with secret IDs as keys and lists of destroyed data IDs as values. | |
| """ | |
| self.logger.info("Starting periodic maintenance tasks (cron)") | |
| destroyed_secret_ids = {} | |
| for secret_id in self._get_secret_ids(): | |
| self.logger.debug(f"Processing secret '{secret_id}' for maintenance") | |
| purged_data_ids = self.purge_disabled_secret_versions(secret_id) | |
| if purged_data_ids: | |
| self.logger.info(f"Purged disabled versions of secret '{secret_id}': {purged_data_ids}") | |
| destroyed_secret_ids[secret_id] = purged_data_ids | |
| return destroyed_secret_ids |