feat(integrations): support aws secret manager (#8)

* feat(factory): add aws secret manager

* refactor(integrations): abstract pagination logic

* style(integrations): remove space

* feat(integrations): add secret manager support

* docs(readme): add integrations section

* test(integrations): verify secret manager fetch secrets

* chore(makefile): add additional runs for each integration

Makefile

@@ -18,5 +18,10 @@ lint:
 format:
 	poetry run ruff check --fix .
 
-run:
-	poetry run python -m fuzzy_secret_stdout
+run: run_ssm
+
+run_ssm:
+	poetry run python -m fuzzy_secret_stdout -i AWS_SSM
+
+run_secretmanager:
+	poetry run python -m fuzzy_secret_stdout -i AWS_SECRET_MAN | jq .

README.md

@@ -16,6 +16,7 @@ Dependencies:
 
 * Python 3.9+
 * [`fzf`](https://github.com/junegunn/fzf?tab=readme-ov-file#installation)
+* Valid [AWS Credentials](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html) available your terminal context
 
 ## Usage
 
@@ -27,5 +28,19 @@ fuzzy-secret-stdout
 fss
 
 # fuzzy search and explicitly specify the secret store to search
-fss -i AWS_SSM
+fss -i AWS_SECRET_MAN
+
+# fuzzy search aws secret manager and pipe into jq
+fss -i AWS_SECRET_MAN | jq .
 ```
+
+## Integrations
+
+`fuzzy-secret-stdout` supports the following secret stores:
+
+| Secret Store                                                                                                                             | Command Line Argument  |
+| -------------                                                                                                                            | ---------------------- |
+| [AWS Systems Manager Parameter Store](https://docs.aws.amazon.com/systems-manager/latest/userguide/systems-manager-parameter-store.html) | `AWS_SSM`              |
+| [AWS Secrets Manager](https://docs.aws.amazon.com/secretsmanager/latest/userguide/intro.html)                                            | `AWS_SECRET_MAN`       |
+
+The *Command Line Argument* above is passed as the `-i` flag. `AWS_SSM` is the default.

fuzzy_secret_stdout/integrations/__init__.py

@@ -1,8 +1,11 @@
 from typing import Optional
 from abc import ABC, abstractmethod
+import logging
 
 from fuzzy_secret_stdout.models import SecretStoreItem
 
+logger = logging.getLogger(__name__)
+
 class SecretIntegration(ABC):
 
     @abstractmethod
@@ -12,3 +15,42 @@ def fetch_all(self, max_batch_results: Optional[int] = 3) -> list[SecretStoreIte
     @abstractmethod
     def fetch_secrets(self, item_names: list[str]) -> list[SecretStoreItem]: # pragma: nocover
         pass
+
+    def _paginate_results(self, func, max_batch_results: int, outer_response_key: str, inner_response_key: str, label: str) -> list[SecretStoreItem]:
+        """
+        Paginates over a function to collect all the results
+
+        Args:
+            func (): the function repeat call e.g call to aws boto describe_parameters
+            max_batch_results: the maximum number of results allowed per internal batch
+            outer_response_key: the key in the response to use to get internal data
+                e.g in an AWS describe_parameters, data is slotted into the "Parameters" payload
+            inner_response_key: the key in the innner response to get secret names
+                e.g in the AWS describe_parameters, data is slotted into the "Name" field
+            label: name used for logging
+
+        Returns:
+            List of SecretStoreItem
+        """
+        logging.info("fetching all %s keys with batch results %s", label, max_batch_results)
+
+        raw_result: dict = func(self._boto_client, MaxResults=max_batch_results)
+
+        if outer_response_key not in raw_result or not raw_result[outer_response_key]:
+            logging.debug("could not find any %s keys", label)
+            return []
+
+        results: list[SecretStoreItem] = []
+        for parameter in raw_result[outer_response_key]:
+            results.append(SecretStoreItem(parameter[inner_response_key]))
+
+        while 'NextToken' in raw_result:
+            logging.info("found %s %s keys and a NextToken, fetching next batch", label, len(raw_result[outer_response_key]))
+
+            raw_result = func(self._boto_client, NextToken=raw_result['NextToken'], MaxResults=max_batch_results)
+
+            for parameter in raw_result[outer_response_key]:
+                results.append(SecretStoreItem(parameter['Name']))
+
+        logging.info("found %s total %s keys", label, len(results))
+        return results

fuzzy_secret_stdout/integrations/aws_secret_manager.py

@@ -0,0 +1,24 @@
+import logging
+from typing import Optional
+
+from fuzzy_secret_stdout.models import SecretStoreItem
+from fuzzy_secret_stdout.integrations import SecretIntegration
+
+logger = logging.getLogger(__name__)
+
+class AWSSecretManager(SecretIntegration):
+
+    def __init__(self, boto_client) -> None:
+        self._boto_client = boto_client
+
+    def fetch_all(self, max_batch_results: Optional[int] = 3) -> list[SecretStoreItem]:
+
+        def inner(boto_client, **kwargs):
+            return boto_client.list_secrets(**kwargs)
+
+        return self._paginate_results(inner, max_batch_results, 'SecretList', 'Name', 'secretmanager')
+
+    def fetch_secrets(self, item_names: list[str]) -> list[SecretStoreItem]:
+        result = self._boto_client.batch_get_secret_value(SecretIdList=item_names)
+        result = [SecretStoreItem(x['Name'], x['SecretString']) for x in result['SecretValues']]
+        return result

fuzzy_secret_stdout/integrations/aws_ssm.py

@@ -12,27 +12,11 @@ def __init__(self, boto_client) -> None:
         self._boto_client = boto_client
 
     def fetch_all(self, max_batch_results: Optional[int] = 3) -> list[SecretStoreItem]:
-        logging.info("fetching all ssm keys with batch results %s", max_batch_results)
 
-        raw_result: dict = self._boto_client.describe_parameters(MaxResults=max_batch_results)
+        def inner(boto_client, **kwargs):
+            return boto_client.describe_parameters(**kwargs)
 
-        if 'Parameters' not in raw_result or not raw_result['Parameters']:
-            logging.debug("could not find any ssm keys")
-            return []
-
-        results: list[SecretStoreItem] = []
-        for parameter in raw_result['Parameters']:
-            results.append(SecretStoreItem(parameter['Name']))
-
-        while 'NextToken' in raw_result:
-            logging.info("found %s ssm keys and a NextToken, fetching next batch", len(raw_result['Parameters']))
-
-            raw_result = self._boto_client.describe_parameters(NextToken=raw_result['NextToken'], MaxResults=max_batch_results)
-            for parameter in raw_result['Parameters']:
-                results.append(SecretStoreItem(parameter['Name']))
-
-        logging.info("found %s total ssm keys", len(results))
-        return results
+        return self._paginate_results(inner, max_batch_results, 'Parameters', 'Name', 'ssm')
 
     def fetch_secrets(self, item_names: list[str]) -> list[SecretStoreItem]:
         result = self._boto_client.get_parameters(Names=item_names, WithDecryption=True)

fuzzy_secret_stdout/integrations/factory.py

@@ -3,10 +3,12 @@
 import boto3
 
 from fuzzy_secret_stdout.integrations.aws_ssm import AWSParameterStore
+from fuzzy_secret_stdout.integrations.aws_secret_manager import AWSSecretManager
 from fuzzy_secret_stdout.integrations import SecretIntegration
 
 class Integration(str, Enum):
     AWS_SSM = "AWS_SSM"
+    AWS_SECRET_MAN = "AWS_SECRET_MAN"
 
     @staticmethod
     def list_options() -> list[str]:
@@ -16,5 +18,7 @@ def list_options() -> list[str]:
 def create_integration(integration: Integration) -> SecretIntegration:
     if integration == Integration.AWS_SSM:
         return AWSParameterStore(boto3.client('ssm'))
+    elif integration == Integration.AWS_SECRET_MAN:
+        return AWSSecretManager(boto3.client('secretsmanager'))
     else:
         raise NotImplementedError(f'integration {integration} not implemented')

tests/integrations/test_aws_secret_manager.py

@@ -0,0 +1,106 @@
+from unittest.mock import Mock, call
+from fuzzy_secret_stdout.integrations.aws_secret_manager import AWSSecretManager
+from fuzzy_secret_stdout.models import SecretStoreItem
+
+import pytest
+
+@pytest.mark.parametrize("list_secrets_return", [
+    pytest.param({}, id='empty_response'),
+    pytest.param({'SecretList': []}, id='empty_parameters')
+])
+def test_fetch_all_no_parameters(list_secrets_return: dict):
+
+    mock_secret_man: Mock = Mock()
+    mock_secret_man.list_secrets.return_value = list_secrets_return
+
+    integration = AWSSecretManager(mock_secret_man)
+    result = integration.fetch_all()
+
+    assert result == []
+    assert mock_secret_man.list_secrets.call_args_list == [call(MaxResults=3)]
+
+
+@pytest.mark.parametrize("max_results", [
+    3,
+    5,
+    10
+])
+def test_fetch_all_max_results_override(max_results: int):
+    mock_secret_man: Mock = Mock()
+    mock_secret_man.list_secrets.return_value = []
+
+    integration = AWSSecretManager(mock_secret_man)
+    integration.fetch_all(max_batch_results=max_results)
+
+    assert mock_secret_man.list_secrets.call_args_list == [call(MaxResults=max_results)]
+
+def test_fetch_all__keys_no_pagination():
+
+    mock_secret_man: Mock = Mock()
+    mock_secret_man.list_secrets.return_value = {
+        'SecretList': [
+            {'Name': 'param1'},
+            {'Name': 'param2'},
+            {'Name': 'param3'},
+        ]
+    }
+
+    integration = AWSSecretManager(mock_secret_man)
+    result = integration.fetch_all()
+
+    assert mock_secret_man.list_secrets.call_args_list == [call(MaxResults=3)]
+    assert result == [
+        SecretStoreItem(key='param1'),
+        SecretStoreItem(key='param2'),
+        SecretStoreItem(key='param3')
+    ]
+
+def test_fetch_all_keys_pagination():
+
+    mock_secret_man: Mock = Mock()
+    mock_secret_man.list_secrets.side_effect = [
+        # initial call
+        {
+            'SecretList': [ {'Name': 'param1'}],
+            'NextToken': 'token1'
+        },
+        # second call
+        {
+            'SecretList': [ {'Name': 'param2'}, {'Name': 'param3'} ],
+            'NextToken': 'token2'
+        },
+        # final call
+        {
+            'SecretList': [
+                {'Name': 'param4'}
+            ]
+        },
+    ]
+
+    integration = AWSSecretManager(mock_secret_man)
+    result = integration.fetch_all()
+
+    assert result == [
+        SecretStoreItem(key='param1'),
+        SecretStoreItem(key='param2'),
+        SecretStoreItem(key='param3'),
+        SecretStoreItem(key='param4')
+    ]
+
+    assert mock_secret_man.list_secrets.call_args_list == [call(MaxResults=3), call(NextToken='token1', MaxResults=3), call(NextToken='token2', MaxResults=3)]
+
+def test_fetch_secrets():
+    input = ['param1']
+
+    mock_secret_man: Mock = Mock()
+    mock_secret_man.batch_get_secret_value.return_value = {
+        'SecretValues': [
+            {'Name': 'param1', 'SecretString': 'value1'}
+        ]
+    }
+
+    integration = AWSSecretManager(mock_secret_man)
+    result = integration.fetch_secrets(input)
+
+    assert result == [SecretStoreItem(key='param1', value='value1')]
+    assert mock_secret_man.batch_get_secret_value.call_args_list == [call(SecretIdList=['param1'])]

tests/integrations/test_aws_ssm.py

@@ -2,7 +2,6 @@
 from fuzzy_secret_stdout.integrations.aws_ssm import AWSParameterStore
 from fuzzy_secret_stdout.models import SecretStoreItem
 
-
 import pytest
 
 

tests/integrations/test_factory.py

@@ -4,6 +4,7 @@
 
 from fuzzy_secret_stdout.integrations.factory import create_integration, Integration
 from fuzzy_secret_stdout.integrations.aws_ssm import AWSParameterStore
+from fuzzy_secret_stdout.integrations.aws_secret_manager import AWSSecretManager
 
 @patch('fuzzy_secret_stdout.integrations.factory.boto3')
 def test_create_integration_ssm(mock_boto: Mock):
@@ -13,10 +14,17 @@ def test_create_integration_ssm(mock_boto: Mock):
     assert result._boto_client == mock_boto.client.return_value
     assert mock_boto.client.call_args_list == [call('ssm')]
 
+@patch('fuzzy_secret_stdout.integrations.factory.boto3')
+def test_create_integration_secretmanager(mock_boto: Mock):
+    result = create_integration(Integration.AWS_SECRET_MAN)
+
+    assert isinstance(result, AWSSecretManager)
+    assert result._boto_client == mock_boto.client.return_value
+    assert mock_boto.client.call_args_list == [call('secretsmanager')]
 
 def test_create_integration_unimplemented():
     with pytest.raises(NotImplementedError, match='integration DUMMY not implemented'):
         create_integration("DUMMY")
 
 def test_integration_list_options():
-    assert Integration.list_options() == ['AWS_SSM']
+    assert Integration.list_options() == ['AWS_SSM', 'AWS_SECRET_MAN']