Enable the use of secrets-manager as a backend, like lastpass

db_facts/aws_secrets_manager.py

@@ -0,0 +1,38 @@
+import json
+
+import boto3
+
+from .db_facts_types import AWSSecret, AWSSecretUsernamePassword
+from .db_type import canonicalize_db_type, db_protocol
+
+
+def pull_aws_secrets_manager_secret(sm_entry_name: str) -> AWSSecret:
+    response = boto3.client("secretsmanager").get_secret_value(SecretId=sm_entry_name)
+    return json.loads(response["SecretString"])
+
+
+def pull_aws_secrets_manager_username_password(
+    sm_entry_name: str,
+) -> AWSSecretUsernamePassword:
+    secret = pull_aws_secrets_manager_secret(sm_entry_name)
+    return {"user": secret["username"], "password": secret["password"]}
+
+
+def db_info_from_secrets_manager(sm_entry_name: str):
+    response = pull_aws_secrets_manager_secret(sm_entry_name)
+
+    result = {key.lower(): value for key, value in response.items()}
+
+    result["host"] = result.pop("hostname")
+    result["user"] = result.pop("username")
+
+    # mypy has issues with `result.get('type')`
+    # https://stackoverflow.com/questions/70955906/how-to-deal-with-incompatible-type-optionalstr-expected-str
+    if "type" in result:
+        result["type"] = canonicalize_db_type(result["type"])
+        result["protocol"] = db_protocol(result["type"])
+    else:
+        result["type"] = ""
+        result["protocol"] = ""
+
+    return result

db_facts/db_facts_types.py

@@ -65,6 +65,8 @@ class DBFacts(TypedDict, total=False):
     DBFacts = Dict[str, Any]
 DBName = List[str]
 LastPassUsernamePassword = Dict[str, str]
+AWSSecret = Any
+AWSSecretUsernamePassword = Dict[str, str]
 DBConfig = Dict[str, Any]
 DBCLIConfig = Any
 JinjaContext = Dict[str, Any]

db_facts/db_info.py

@@ -5,6 +5,10 @@
 from .errors import fail_on_invalid_db_name
 from .config import load_config
 from .lpass import pull_lastpass_username_password, db_info_from_lpass
+from .aws_secrets_manager import (
+    db_info_from_secrets_manager,
+    pull_aws_secrets_manager_username_password
+)
 from .db_facts_types import DBConfig, DBCLIConfig, DBFacts, DBName
 from .db_config import db_config
 
@@ -78,6 +82,24 @@ def db(db_name: DBName, dbcli_config: DBCLIConfig = None) -> DBFacts:
             additional_attributes = \
                 pull_lastpass_username_password(lastpass_entry_name)
             db_info['exports'].update(additional_attributes)
+        elif 'pull_secrets_manager_from' in dbcli_config['exports_from'][method]:
+            template_for_secrets_manager_entry_name =\
+                dbcli_config['exports_from'][method]['pull_secrets_manager_from']
+            secrets_manager_entry_name = template(template_for_secrets_manager_entry_name,
+                                                  (db_info, {}))
+            additional_attributes = \
+                db_info_from_secrets_manager(secrets_manager_entry_name)
+            db_info['exports'].update(additional_attributes)
+        elif 'pull_secrets_manager_username_password_from' in \
+             dbcli_config['exports_from'][method]:
+            method = dbcli_config['exports_from'][method]
+            template_for_secrets_manager_entry_name =\
+                method['pull_secrets_manager_username_password_from']
+            secrets_manager_entry_name = template(template_for_secrets_manager_entry_name,
+                                           (db_info, {}))
+            additional_attributes = \
+                pull_aws_secrets_manager_username_password(secrets_manager_entry_name)
+            db_info['exports'].update(additional_attributes)
         else:
             raise SyntaxError(f'Did not understand exports_from {method}')
 

metrics/bigfiles_high_water_mark

@@ -1 +1 @@
-369
+383

metrics/flake8_high_water_mark

@@ -1 +1 @@
-1
+2

requirements.txt

@@ -8,6 +8,7 @@ mypy
 lxml # needed for mypy coverage report
 types-pyyaml # needed for mypy coverage report
 types-pkg-resources # needed for mypy coverage report
+boto3-stubs[essential] # needed for mypy coverage report
 wheel # needed to publish to PyPI in CircleCI
 twine # needed to publish to PyPI in CircleCI
 sphinx>=3 # used to generate and upload docs - sphinx-autodoc-typehints requires 4 or better per https://github.com/agronholm/sphinx-autodoc-typehints/pull/138

setup.py

@@ -113,7 +113,7 @@ def initialize_options(self) -> None:
       author_email='[email protected]',
       packages=find_packages(),
       package_data={"db_facts": ["py.typed"]},
-      install_requires=['jinja2', 'pyyaml'],
+      install_requires=['jinja2', 'pyyaml', 'boto3'],
       entry_points={
         'console_scripts': [
             'db_facts = db_facts.__main__:main'

tests/mock_dbcli_config.py

@@ -13,6 +13,12 @@
         },
         'invalid-method': {
         },
+        'secrets_manager': {
+            'pull_secrets_manager_from': "{{ secrets_manager_entry }}"
+        },
+        'secrets_manager_user_and_pass_only': {
+            'pull_secrets_manager_username_password_from': "{{ secrets_manager_entry }}"
+        },
     },
     'dbs': {
         'baz': {
@@ -29,6 +35,10 @@
             'exports_from': 'lpass',
             'lastpass_entry': 'different lpass entry name'
         },
+        'fromage': {
+            'exports_from': 'secrets_manager',
+            'secrets_manager_entry': 'secrets manager entry name'
+        },
         'frazzle': {
             'exports_from': 'lpass',
             'lastpass_entry': 'lpass entry name'
@@ -42,6 +52,15 @@
                 'a_numbered_export': 123
             },
         },
+        'fronk': {
+            'exports_from': 'secrets_manager_user_and_pass_only',
+            'secrets_manager_entry': 'secrets manager entry name',
+            'jinja_context_name': 'standard',
+            'exports': {
+                'some_additional': 'export',
+                'a_numbered_export': 123
+            },
+        },
         'gaggle': {
             'jinja_context_name': [
                 'env',

tests/test_db_info_secrets_manager.py

@@ -0,0 +1,75 @@
+from db_facts.db_info import db
+import unittest
+from unittest.mock import patch
+from .mock_dbcli_config import mock_dbcli_config
+
+
+@patch("db_facts.db_info.pull_jinja_context")
+@patch("db_facts.db_info.db_info_from_secrets_manager")
+@patch("db_facts.aws_secrets_manager.pull_aws_secrets_manager_secret")
+class TestDBInfoSecretsManager(unittest.TestCase):
+    def test_db_info_secrets_manager(
+        self,
+        mock_pull_aws_secrets_manager_secret,
+        mock_db_info_from_secrets_manager,
+        mock_pull_jinja_context,
+    ):
+
+        expected_result = {
+            "database": "database",
+            "host": "host",
+            "password": "password",
+            "port": "port",
+            "protocol": "protocol",
+            "type": "type",
+            "user": "user",
+            "connection_type": "direct",
+        }
+        mock_db_info_from_secrets_manager.return_value = {
+            "database": "database",
+            "port": "port",
+            "host": "host",
+            "type": "type",
+            "user": "user",
+            "protocol": "protocol",
+            "password": "password",
+        }
+        mock_pull_jinja_context.return_value = ({}, {})
+        db_facts = db(["fromage"], dbcli_config=mock_dbcli_config)
+        mock_db_info_from_secrets_manager.assert_called_with(
+            "secrets manager entry name"
+        )
+        self.assertEqual(expected_result, db_facts)
+        mock_pull_jinja_context.assert_called_with(
+            ["fromage"], mock_dbcli_config["dbs"]["fromage"], mock_dbcli_config
+        )
+
+    def test_db_info_pull_secrets_manager_user_and_pass_only(
+        self,
+        mock_pull_aws_secrets_manager_secret,
+        mock_db_info_from_secrets_manager,
+        mock_pull_jinja_context,
+    ):
+
+        sm_entry = {
+            "username": "user",
+            "password": "password",
+        }
+
+        mock_pull_aws_secrets_manager_secret.return_value = sm_entry
+        expected_result = {
+            "password": "password",
+            "user": "user",
+            "connection_type": "direct",
+            "some_additional": "export",
+            "a_numbered_export": 123,
+        }
+        mock_pull_jinja_context.return_value = ({}, {})
+        db_facts = db(["fronk"], dbcli_config=mock_dbcli_config)
+        mock_pull_aws_secrets_manager_secret.assert_called_with(
+            "secrets manager entry name"
+        )
+        self.assertEqual(expected_result, db_facts)
+        mock_pull_jinja_context.assert_called_with(
+            ["fronk"], mock_dbcli_config["dbs"]["fronk"], mock_dbcli_config
+        )

tests/test_secrets_manager.py

@@ -0,0 +1,29 @@
+import unittest
+from unittest.mock import patch
+
+from db_facts import aws_secrets_manager
+
+
+class TestSecretsManager(unittest.TestCase):
+    @patch("db_facts.aws_secrets_manager.pull_aws_secrets_manager_secret")
+    def test_db_info_from_secrets_manager(self, mock_pull_from_aws_sm):
+        mock_pull_from_aws_sm.return_value = {
+            "Database": "fakedatabase",
+            "Port": 123,
+            "Hostname": "fakehost",
+            "Type": "faketype",
+            "Username": "fakeuser",
+            "Password": "fakepassword",
+        }
+        db_info = aws_secrets_manager.db_info_from_secrets_manager("my_secret")
+        expected_db_info = {
+            "database": "fakedatabase",
+            "host": "fakehost",
+            "password": "fakepassword",
+            "port": 123,
+            "type": "faketype",
+            "user": "fakeuser",
+            "protocol": "faketype",  # if we don't know, just pass through
+        }
+
+        assert db_info == expected_db_info