feat: Add support for aws secret manager (#2222)

Co-authored-by: Tal <[email protected]>

docs/deployment/secret-manager.mdx

@@ -16,15 +16,15 @@ The `SECRET_MANAGER_TYPE` environment variable plays a crucial role in the Secre
 **Functionality**:
 
 **Default Secret Manager**: If the `SECRET_MANAGER_TYPE` environment variable is set, its value dictates the default type of secret manager that the factory will create.
-The value of this variable should correspond to one of the types defined in SecretManagerTypes enum (`FILE`, `GCP`, `K8S`, `VAULT`).
+The value of this variable should correspond to one of the types defined in SecretManagerTypes enum (`FILE`, `AWS`, `GCP`, `K8S`, `VAULT`).
 
 **Example Configuration**:
 
 Setting `SECRET_MANAGER_TYPE=GCP` in the environment will make the factory create instances of GcpSecretManager by default.
 If `SECRET_MANAGER_TYPE` is not set or is set to `FILE`, the factory defaults to creating instances of FileSecretManager.
 This environment variable provides flexibility and ease of configuration, allowing different secret managers to be used in different environments or scenarios without code changes.
 
-## File Secert Manager
+## File Secret Manager
 
 The `FileSecretManager` is a concrete implementation of the BaseSecretManager for managing secrets stored in the file system. It uses a specified directory (defaulting to ./) to read, write, and delete secret files.
 
@@ -39,6 +39,24 @@ Usage:
 - Writing a secret creates or updates a file with the given content.
 - Deleting a secret removes the corresponding file.
 
+## AWS Secret Manager
+
+The `AwsSecretManager` integrates with Amazon Web Services' Secrets Manager service for secure secret management. It provides a robust solution for storing and managing secrets in AWS environments.
+
+Configuration:
+
+Required environment variables:
+- `AWS_REGION`: The AWS region where your secrets are stored
+- For local development:
+  - `AWS_ACCESS_KEY_ID`: Your AWS access key
+  - `AWS_SECRET_ACCESS_KEY`: Your AWS secret access key
+
+
+Usage:
+
+- Manages secrets using AWS Secrets Manager service
+- Supports creating, updating, reading, and deleting secrets
+
 ## Kubernetes Secret Manager
 
 The `KubernetesSecretManager` interfaces with Kubernetes' native secrets system. It manages secrets within a specified Kubernetes namespace and is designed to operate within a Kubernetes cluster.

keep/secretmanager/awssecretmanager.py

@@ -0,0 +1,184 @@
+import json
+import os
+
+import boto3
+import opentelemetry.trace as trace
+from botocore.exceptions import ClientError
+from keep.secretmanager.secretmanager import BaseSecretManager
+
+tracer = trace.get_tracer(__name__)
+
+
+class AwsSecretManager(BaseSecretManager):
+    def __init__(self, context_manager, **kwargs):
+        super().__init__(context_manager)
+        try:
+            session = boto3.session.Session()
+            self.client = session.client(
+                service_name="secretsmanager", region_name=os.environ.get("AWS_REGION")
+            )
+        except Exception as e:
+            self.logger.error(
+                "Failed to initialize AWS Secrets Manager client",
+                extra={"error": str(e)},
+            )
+            raise
+
+    def write_secret(self, secret_name: str, secret_value: str) -> None:
+        """
+        Writes a secret to AWS Secrets Manager.
+        Args:
+            secret_name (str): The name of the secret.
+            secret_value (str): The value of the secret.
+        Raises:
+            ClientError: If an AWS-specific error occurs while writing the secret.
+            Exception: If any other unexpected error occurs.
+        """
+        with tracer.start_as_current_span("write_secret"):
+            self.logger.info("Writing secret", extra={"secret_name": secret_name})
+
+            try:
+                # Check if secret exists by trying to describe it
+                self.client.describe_secret(SecretId=secret_name)
+
+                # If secret exists, update it with new value
+                self.client.put_secret_value(
+                    SecretId=secret_name, SecretString=secret_value
+                )
+                self.logger.info(
+                    "Secret updated successfully", extra={"secret_name": secret_name}
+                )
+            except ClientError as e:
+                if e.response["Error"]["Code"] == "ResourceNotFoundException":
+                    try:
+                        # Create new secret if it doesn't exist
+                        self.client.create_secret(
+                            Name=secret_name, SecretString=secret_value
+                        )
+                        self.logger.info(
+                            "Secret created successfully",
+                            extra={"secret_name": secret_name},
+                        )
+                    except Exception as e:
+                        self.logger.error(
+                            "Unexpected error while creating secret",
+                            extra={
+                                "secret_name": secret_name,
+                                "error": str(e),
+                                "error_type": type(e).__name__,
+                            },
+                        )
+                        raise
+                else:
+                    self.logger.error(
+                        "AWS error while writing secret",
+                        extra={
+                            "secret_name": secret_name,
+                            "error": str(e),
+                            "error_code": e.response["Error"]["Code"],
+                        },
+                    )
+                    raise
+            except Exception as e:
+                self.logger.error(
+                    "Unexpected error while writing secret",
+                    extra={
+                        "secret_name": secret_name,
+                        "error": str(e),
+                        "error_type": type(e).__name__,
+                    },
+                )
+                raise
+
+    def read_secret(self, secret_name: str, is_json: bool = False) -> str | dict:
+        """
+        Reads a secret from AWS Secrets Manager.
+        Args:
+            secret_name (str): The name of the secret.
+            is_json (bool): Whether to parse the secret as JSON. Defaults to False.
+        Returns:
+            str | dict: The secret value as a string, or as a dict if is_json=True.
+        Raises:
+            ClientError: If an AWS-specific error occurs while reading the secret.
+            Exception: If any other unexpected error occurs.
+        """
+        with tracer.start_as_current_span("read_secret"):
+            self.logger.debug("Getting secret", extra={"secret_name": secret_name})
+
+            try:
+                response = self.client.get_secret_value(SecretId=secret_name)
+                secret_value = response["SecretString"]
+
+                if is_json:
+                    try:
+                        secret_value = json.loads(secret_value)
+                    except json.JSONDecodeError as e:
+                        self.logger.error(
+                            "Failed to parse secret as JSON",
+                            extra={"secret_name": secret_name, "error": str(e)},
+                        )
+                        raise
+
+                self.logger.debug(
+                    "Got secret successfully", extra={"secret_name": secret_name}
+                )
+                return secret_value
+
+            except ClientError as e:
+                self.logger.error(
+                    "AWS error while reading secret",
+                    extra={
+                        "secret_name": secret_name,
+                        "error": str(e),
+                        "error_code": e.response["Error"]["Code"],
+                    },
+                )
+                raise
+            except Exception as e:
+                self.logger.error(
+                    "Unexpected error while reading secret",
+                    extra={
+                        "secret_name": secret_name,
+                        "error": str(e),
+                        "error_type": type(e).__name__,
+                    },
+                )
+                raise
+
+    def delete_secret(self, secret_name: str) -> None:
+        """
+        Deletes a secret from AWS Secrets Manager.
+        Args:
+            secret_name (str): The name of the secret.
+        Raises:
+            ClientError: If an AWS-specific error occurs while deleting the secret.
+            Exception: If any other unexpected error occurs.
+        """
+        with tracer.start_as_current_span("delete_secret"):
+            try:
+                self.client.delete_secret(
+                    SecretId=secret_name, ForceDeleteWithoutRecovery=True
+                )
+                self.logger.info(
+                    "Secret deleted successfully", extra={"secret_name": secret_name}
+                )
+            except ClientError as e:
+                self.logger.error(
+                    "AWS error while deleting secret",
+                    extra={
+                        "secret_name": secret_name,
+                        "error": str(e),
+                        "error_code": e.response["Error"]["Code"],
+                    },
+                )
+                raise
+            except Exception as e:
+                self.logger.error(
+                    "Unexpected error while deleting secret",
+                    extra={
+                        "secret_name": secret_name,
+                        "error": str(e),
+                        "error_type": type(e).__name__,
+                    },
+                )
+                raise

keep/secretmanager/secretmanagerfactory.py

@@ -10,6 +10,7 @@ class SecretManagerTypes(enum.Enum):
     GCP = "gcp"
     K8S = "k8s"
     VAULT = "vault"
+    AWS = "aws"
 
 
 class SecretManagerFactory:
@@ -41,6 +42,10 @@ def get_secret_manager(
             from keep.secretmanager.vaultsecretmanager import VaultSecretManager
 
             return VaultSecretManager(context_manager, **kwargs)
+        elif secret_manager_type == SecretManagerTypes.AWS:
+            from keep.secretmanager.awssecretmanager import AwsSecretManager
+
+            return AwsSecretManager(context_manager, **kwargs)
 
         raise NotImplementedError(
             f"Secret manager type {str(secret_manager_type)} not implemented"