feat: remove secrets listing to reduce required AWS permissions

closes #107

app/handler.py

@@ -6,28 +6,24 @@
 
 from app.settings import load_settings
 from app.services.certbot import obtain_certbot_certs
-from app.services.aws import list_secret_names, upload_certs_as_secrets
+from app.services.aws import upload_certs_as_secrets
 
 
 def handler(_event, _context):
-    if environ.get('TESTMODE') == 'true':
+    if environ.get("TESTMODE") == "true":
         plugins = list(plugins_disco.PluginsRegistry.find_all())
-        dns_plugins = [v for v in plugins if v.startswith('dns-')]
+        dns_plugins = [v for v in plugins if v.startswith("dns-")]
 
         if len(dns_plugins) != 14:
-            raise Exception('Failed to discover all certbot DNS plugins')
-        
+            raise Exception("Failed to discover all certbot DNS plugins")
+
         return
     else:
         settings = load_settings()
 
-
         try:
             shutil.rmtree(str(settings.CERTBOT_DIR), ignore_errors=True)
 
-            # Load secret names early to check if aws client is configured correctly
-            secret_names = list_secret_names()
-
             certs = obtain_certbot_certs(
                 emails=settings.CERTBOT_EMAILS,
                 domains=settings.CERTBOT_DOMAINS,
@@ -43,7 +39,6 @@ def handler(_event, _context):
             upload_certs_as_secrets(
                 certs,
                 name=settings.AWS_SECRET_NAME,
-                secret_names=secret_names,
                 description=settings.AWS_SECRET_DESCRIPTION,
             )
         finally:

app/services/aws.py

@@ -5,43 +5,34 @@
 from .certbot import Cert
 
 
-def list_secret_names() -> list[str]:
-    secretsmanager = client("secretsmanager")
-    return {v["Name"] for v in secretsmanager.list_secrets()["SecretList"]}
-
-
 def upload_certs_as_secrets(
-    certs: list[Cert], name: str, secret_names: list[str] = None, description: str = ""
+    certs: list[Cert], name: str, description: str = ""
 ) -> None:
     for cert in certs:
         name = name.format(domain=slugify(cert.domain))
 
         create_or_update_secret(
             name=name,
             data={f.name: f.content for f in cert.files},
-            secret_names=secret_names,
             description=description,
         )
 
 
 def create_or_update_secret(
     name: str,
     data: dict[str, str],
-    secret_names: list[str] = None,
     description: str = "",
 ):
     secretsmanager = client("secretsmanager")
-    secret_names = secret_names if secret_names is not None else list_secret_names()
-
-    if name in secret_names:
-        print(f"Updating secret {name} with new certs")
-
-        secretsmanager.put_secret_value(SecretId=name, SecretString=json.dumps(data))
-    else:
-        print(f"Creating a new secret {name}")
 
+    try:
         secretsmanager.create_secret(
             Name=name,
             Description=description,
             SecretString=json.dumps(data),
         )
+        print(f"Creating a new secret {name}")
+    except secretsmanager.exceptions.ResourceExistsException:
+        print(f"Updating secret {name} with new certs")
+
+        secretsmanager.put_secret_value(SecretId=name, SecretString=json.dumps(data))

app/services/certbot.py

@@ -27,7 +27,7 @@ def obtain_certbot_certs(
     credentials: str = None,
     propagation_seconds: int = None,
 ) -> list[Cert]:
-    with NamedTemporaryFile(mode = "w") as tmp:
+    with NamedTemporaryFile(mode="w") as tmp:
         if credentials:
             tmp.write(credentials)
             tmp.flush()
@@ -63,11 +63,7 @@ def obtain_certbot_certs(
             # Rewrite preferred chain
             *(["--preferred-chain", preferred_chain] if preferred_chain else []),
             # Credentials file
-            *(
-                [f"--{dns_plugin}-credentials", tmp.name]
-                if credentials
-                else []
-            ),
+            *([f"--{dns_plugin}-credentials", tmp.name] if credentials else []),
             # The number of seconds to wait for DNS
             *(
                 [f"--{dns_plugin}-propagation-seconds", propagation_seconds]

pyproject.toml

@@ -47,6 +47,7 @@ build-pex = "pex . -o dist/certbot-lambda.zip"
 build-lambda = "lambdex build -e app.handler:handler -M main.py dist/certbot-lambda.zip"
 test = "pytest --cov=app --cov-report xml tests/"
 test-build = "task build && TESTMODE=true lambdex test --empty dist/certbot-lambda.zip"
+run-build = "task build && lambdex test --empty dist/certbot-lambda.zip"
 dev = "python -c 'from app.handler import handler; handler(None, None)'"
 prettify = "black ./app ./tests"
 release = "semantic-release publish"

tests/test_hadler.py

@@ -1,4 +1,4 @@
-from unittest.mock import patch, MagicMock
+from unittest.mock import patch
 import os
 from tempfile import TemporaryDirectory
 from pathlib import Path
@@ -14,6 +14,10 @@
 }
 
 
+class ResourceExistsException(Exception):
+    pass
+
+
 @patch("app.settings.load_dotenv")
 @patch("app.services.aws.client")
 @patch("app.services.certbot.main.main")
@@ -80,9 +84,8 @@ def certbot_side_effect(args: list[str]):
 @patch("app.services.aws.client")
 @patch("app.services.certbot.main.main")
 def test_existing_aws_cert(certbot, client, _dotenv):
-    client.return_value.list_secrets.return_value = {
-        "SecretList": [{"Name": "certbot-domain-com"}]
-    }
+    client.return_value.create_secret.side_effect = ResourceExistsException("")
+    client.return_value.exceptions.ResourceExistsException = ResourceExistsException
 
     with TemporaryDirectory() as tmpdir:
         path = Path(tmpdir).resolve()