Add optional read only user

Author: Tammo-Feldmann

cdk_bootstrapped_db/constructs.py

@@ -19,7 +19,7 @@
 class BootstrappedDb(Construct):
     """
     Given an RDS database, connect to DB and create a database, user, and
-    password
+    password, optionally a read-only user if `read_only_username` is provided.
     """
 
     def __init__(
@@ -32,6 +32,7 @@ def __init__(
         secrets_prefix: str,
         handler: Union[aws_lambda.Function, aws_lambda.SingletonFunction],
         db_engine: Optional[str] = None,
+        read_only_username: Optional[str] = None,
     ) -> None:
         super().__init__(scope, id)
 
@@ -44,6 +45,7 @@ def __init__(
                 " you must provide a value for `db_engine` explicitly."
             )
 
+        # Create secret for the main user
         self.secret = secretsmanager.Secret(
             self,
             id,
@@ -66,27 +68,66 @@ def __init__(
             description=f"Deployed by {Stack.of(self).stack_name}",
         )
 
+        # Optionally create a secret for the read-only user
+        self.read_only_secret = None
+        if read_only_username:
+            self.read_only_secret = secretsmanager.Secret(
+                self,
+                f"{id}-readonly-secret",
+                secret_name=os.path.join(
+                    secrets_prefix,
+                    f"{id}_readonly".replace(" ", "_"),
+                    self.node.id[-8:],
+                ),
+                generate_secret_string=secretsmanager.SecretStringGenerator(
+                    secret_string_template=json.dumps(
+                        {
+                            "dbname": new_dbname,
+                            "engine": db_engine if db_engine else db.engine.engine_type,  # type: ignore
+                            "port": db.instance_endpoint.port,
+                            "host": db.instance_endpoint.hostname,
+                            "username": read_only_username,
+                        },
+                    ),
+                    generate_string_key="password",
+                    exclude_punctuation=True,
+                ),
+                description=f"Read-only user secret deployed by {Stack.of(self).stack_name}",
+            )
+
         self.provider = custom_resources.Provider(
             scope, "BootstrapProvider", on_event_handler=handler
         )
 
+        # Prepare the properties for the custom resource
+        resource_properties = {
+            "conn_secret_arn": db.secret.secret_arn,
+            "new_user_secret_arn": self.secret.secret_arn,
+            "version": handler.current_version.version,
+        }
+
+        # Optionally include the read-only secret ARN
+        if self.read_only_secret:
+            resource_properties["read_only_user_secret_arn"] = (
+                self.read_only_secret.secret_arn
+            )
+
         self.resource = CustomResource(
             scope=scope,
             id="BootstrapHandlerResource",
             service_token=self.provider.service_token,
-            properties={
-                "conn_secret_arn": db.secret.secret_arn,
-                "new_user_secret_arn": self.secret.secret_arn,
-                "version": handler.current_version.version,
-            },
+            properties=resource_properties,
         )
 
-        # Allow lambda to...
-        # read new user secret
+        # Grant the Lambda handler permissions to read the main user secret
         self.secret.grant_read(handler)
-        # read database secret
+
+        # If the read-only secret exists, grant the handler permissions to read it
+        if self.read_only_secret:
+            self.read_only_secret.grant_read(handler)
+
+        # Grant the Lambda handler permission to connect to the database
         db.secret.grant_read(handler)
-        # connect to database
         db.connections.allow_from(handler, port_range=ec2.Port.tcp(5432))
 
     def is_required_by(self, construct: Construct):