SNS notification for GitOps certificate issued

modules/terraform-aws-ca-iam/main.tf

@@ -17,5 +17,6 @@ resource "aws_iam_role_policy" "lambda" {
     ddb_table_arn          = var.ddb_table_arn,
     external_s3_bucket_arn = var.external_s3_bucket_arn,
     internal_s3_bucket_arn = var.internal_s3_bucket_arn
+    sns_topic_arn          = var.sns_topic_arn
   })
 }

modules/terraform-aws-ca-iam/templates/tls_cert_policy.json.tpl

@@ -108,6 +108,16 @@
       "Resource": [
         "${internal_s3_bucket_arn}/*"
       ]
+    },
+    {
+      "Sid": "SNSPublish",
+      "Effect": "Allow",
+      "Action": [
+        "sns:Publish"
+      ],
+      "Resource": [
+        "${sns_topic_arn}"
+      ]
     }
   ]
 }

modules/terraform-aws-ca-iam/variables.tf

@@ -54,4 +54,9 @@ variable "internal_s3_bucket_arn" {
 variable "aws_principals" {
   description = "List of ARNs for AWS principals allowed to assume role"
   default     = []
-}
+}
+
+variable "sns_topic_arn" {
+  description = "SNS Topic ARN"
+  default     = ""
+}

modules/terraform-aws-ca-lambda/lambda_code/tls_cert/tls_cert.py

@@ -1,6 +1,7 @@
 import base64
 import os
 
+from utils.aws.sns import publish_to_sns
 from utils.certs.kms import kms_get_kms_key_id, kms_describe_key
 from utils.certs.crypto import (
     crypto_cert_request_info,
@@ -19,7 +20,7 @@
     db_list_certificates,
     db_issue_certificate,
 )
-from utils.certs.s3 import s3_download
+from utils.certs.s3 import s3_download, is_cert_gitops
 from cryptography.x509 import load_pem_x509_certificate, load_pem_x509_csr
 from cryptography.hazmat.primitives import serialization
 from dataclasses import dataclass, field
@@ -206,12 +207,22 @@ def create_ca_chain_response(project: str, env_name: str, root_ca_name: str, iss
     )
 
 
+def sns_notify_cert_issued(cert_json, sns_topic_arn):
+    keys_to_publish = ["CertificateInfo", "Base64Certificate", "Subject"]
+    response = publish_to_sns(cert_json, "Certificate Issued", sns_topic_arn, keys_to_publish)
+
+    assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
+    common_name = cert_json["CertificateInfo"]["CommonName"]
+    print(f"Certificate details for {common_name} published to SNS")
+
+
 def lambda_handler(event, context):  # pylint:disable=unused-argument,too-many-locals
     project = os.environ["PROJECT"]
     env_name = os.environ["ENVIRONMENT_NAME"]
     external_s3_bucket_name = os.environ["EXTERNAL_S3_BUCKET"]
     internal_s3_bucket_name = os.environ["INTERNAL_S3_BUCKET"]
     max_cert_lifetime = int(os.environ["MAX_CERT_LIFETIME"])
+    sns_topic_arn = os.environ["SNS_TOPIC_ARN"]
     domain = os.environ.get("DOMAIN")
 
     public_crl = os.environ.get("PUBLIC_CRL")
@@ -271,9 +282,12 @@ def lambda_handler(event, context):  # pylint:disable=unused-argument,too-many-l
         certificate_info=cert_info,
         base64_certificate=base64_certificate.decode("utf-8"),
         subject=load_pem_x509_certificate(base64.b64decode(base64_certificate)).subject.rfc4514_string(),
-        base64_root_ca_certificate=ca_chain_response.base64_root_ca_certificate,
-        base64_issuing_ca_certificate=ca_chain_response.base64_issuing_ca_certificate,
+        base64_root_ca_certificate=ca_chain_response.base64_root_ca_certificate.decode("utf-8"),
+        base64_issuing_ca_certificate=ca_chain_response.base64_issuing_ca_certificate.decode("utf-8"),
         base64_ca_chain=ca_chain_response.base64_ca_chain,
     )
 
+    if is_cert_gitops(internal_s3_bucket_name, response.subject):
+        sns_notify_cert_issued(response.to_dict(), sns_topic_arn)
+
     return response.to_dict()

modules/terraform-aws-ca-lambda/utils/aws/sns.py

@@ -0,0 +1,21 @@
+import boto3
+import json
+
+
+def publish_to_sns(json_data, subject, sns_topic_arn, keys_to_publish="All"):
+    # Filter out unwanted keys
+    if keys_to_publish == "All":
+        keys_to_publish = json_data.keys()
+
+    filtered_json_data = {key: json_data[key] for key in keys_to_publish if key in json_data}
+
+    client = boto3.client("sns")
+
+    response = client.publish(
+        TargetArn=sns_topic_arn,
+        Subject=subject,
+        Message=json.dumps({"default": json.dumps(filtered_json_data)}),
+        MessageStructure="json",
+    )
+
+    return response

modules/terraform-aws-ca-lambda/utils/certs/s3.py

@@ -1,4 +1,5 @@
 import boto3
+import json
 
 
 def s3_download_file(bucket_name, key):
@@ -39,3 +40,45 @@ def s3_upload(
         return s3_upload_file(file, external_s3_bucket_name, key, content_type)
 
     return s3_upload_file(file, internal_s3_bucket_name, key, content_type)
+
+
+def convert_to_json(input_str):
+    # split string by commas
+    pairs = input_str.split(",")
+
+    # split each pair by '=' and construct dictionary
+    json_dictionary = {}
+    for pair in pairs:
+        key, value = pair.split("=")
+        json_dictionary[key] = value
+
+    return json_dictionary
+
+
+def is_cert_gitops(internal_s3_bucket_name, subject):
+    subject_json = convert_to_json(subject)
+
+    cn = subject_json["CN"]
+    o = subject_json["O"]
+    ou = subject_json["OU"]
+
+    # get list of GitOps certificates from internal S3 bucket
+    tls_file = s3_download_file(internal_s3_bucket_name, "tls.json")
+
+    # convert to json dictionary
+    gitops_certs = json.loads(tls_file["Body"].read())
+
+    for cert in gitops_certs:
+        common_name = cert["common_name"]
+        organization = cert.get("organization")
+        organizational_unit = cert.get("organizational_unit")
+
+        # check if certificate is included in tls.json
+        if (
+            cn == common_name
+            and (organization is None or o == organization)
+            and (organizational_unit is None or ou == organizational_unit)
+        ):
+            return True
+
+    return False