Refactor all gc07

Author: david-dearden-ssc

src/lambda/gc07_check_certificate_authorities/app.py

@@ -3,176 +3,17 @@
 
 import json
 import logging
-import time
-import re
 
-import boto3
-import botocore
+from utils import is_scheduled_notification, check_required_parameters
+from boto_util.client import get_client
+from boto_util.config import build_evaluation, submit_evaluations
+from boto_util.s3 import check_s3_object_exists, get_lines_from_s3_file
+from boto_util.acm import list_all_acm_certificates, describe_acm_certificate
 
-ASSUME_ROLE_MODE = True
-ACCOUNT_RESOURCE_TYPE = "AWS::::Account"
 
-
-def evaluate_parameters(rule_parameters):
-    """Evaluate the rule parameters dictionary validity. Raise a Exception for invalid parameters.
-    Keyword arguments:
-    rule_parameters -- the Key/Value dictionary of the Config Rule parameters
-    """
-    if "S3CasCurrentlyInUsePath" not in rule_parameters:
-        logger.error('The parameter with "S3CasCurrentlyInUsePath" as key must be defined.')
-        raise ValueError('The parameter with "S3CasCurrentlyInUsePath" as key must be defined.')
-    if not rule_parameters["S3CasCurrentlyInUsePath"]:
-        logger.error('The parameter "S3CasCurrentlyInUsePath" must have a defined value.')
-        raise ValueError('The parameter "S3CasCurrentlyInUsePath" must have a defined value.')
-    return rule_parameters
-
-
-def build_evaluation(resource_id, compliance_type, event, resource_type, annotation=None):
-    """Form an evaluation as a dictionary. Usually suited to report on scheduled rules.
-    Keyword arguments:
-    resource_id -- the unique id of the resource to report
-    compliance_type -- either COMPLIANT, NON_COMPLIANT or NOT_APPLICABLE
-    event -- the event variable given in the lambda handler
-    resource_type -- the CloudFormation resource type (or AWS::::Account) to report on the rule
-    annotation -- an annotation to be added to the evaluation (default None)
-    """
-    eval_cc = {}
-    if annotation:
-        eval_cc["Annotation"] = annotation
-    eval_cc["ComplianceResourceType"] = resource_type
-    eval_cc["ComplianceResourceId"] = resource_id
-    eval_cc["ComplianceType"] = compliance_type
-    eval_cc["OrderingTimestamp"] = str(json.loads(event["invokingEvent"])["notificationCreationTime"])
-    return eval_cc
-
-
-def get_client(service, event, region="ca-central-1"):
-    """Return the service boto client. It should be used instead of directly calling the client.
-    Keyword arguments:
-    service -- the service name used for calling the boto.client()
-    event -- the event variable given in the lambda handler
-    """
-    if not ASSUME_ROLE_MODE or (AWS_ACCOUNT_ID == AUDIT_ACCOUNT_ID):
-        return boto3.client(service, region_name=region)
-    execution_role_arn = f"arn:aws:iam::{AWS_ACCOUNT_ID}:role/{EXECUTION_ROLE_NAME}"
-    credentials = get_assume_role_credentials(execution_role_arn, region)
-    return boto3.client(
-        service,
-        region_name=region,
-        aws_access_key_id=credentials["AccessKeyId"],
-        aws_secret_access_key=credentials["SecretAccessKey"],
-        aws_session_token=credentials["SessionToken"],
-    )
-
-
-def get_assume_role_credentials(role_arn, region="ca-central-1"):
-    """Return the service boto client. It should be used instead of directly calling the client.
-    Keyword arguments:
-    service -- the service name used for calling the boto.client()
-    event -- the event variable given in the lambda handler
-    """
-    sts_client = boto3.client("sts", region_name=region)
-    try:
-        assume_role_response = sts_client.assume_role(RoleArn=role_arn, RoleSessionName="configLambdaExecution")
-        return assume_role_response["Credentials"]
-    except botocore.exceptions.ClientError as ex:
-        if "AccessDenied" in ex.response["Error"]["Code"]:
-            ex.response["Error"]["Message"] = "AWS Config does not have permission to assume the IAM role."
-        else:
-            ex.response["Error"]["Message"] = "InternalError"
-            ex.response["Error"]["Code"] = "InternalError"
-        raise ex
-
-
-def is_scheduled_notification(message_type):
-    """Check whether the message is a ScheduledNotification or not.
-    Keyword arguments:
-    message_type -- the message type
-    """
-    return message_type == "ScheduledNotification"
-
-
-def check_s3_object_exists(aws_s3_client, object_path: str) -> bool:
-    """Check if the S3 object exists
-    Keyword arguments:
-    object_path -- the S3 object path
-    """
-    # parse the S3 path
-    match = re.match(r"s3:\/\/([^/]+)\/((?:[^/]*/)*.*)", object_path)
-    if match:
-        bucket_name = match.group(1)
-        key_name = match.group(2)
-    else:
-        logger.error("Unable to parse S3 object path %s", object_path)
-        raise ValueError(f"Unable to parse S3 object path {object_path}")
-    try:
-        aws_s3_client.head_object(Bucket=bucket_name, Key=key_name)
-    except botocore.exceptions.ClientError as err:
-        if err.response["Error"]["Code"] == "404":
-            # The object does not exist.
-            logger.info("Object %s not found in bucket %s", key_name, bucket_name)
-            return False
-        elif err.response["Error"]["Code"] == "403":
-            # AccessDenied
-            logger.info("Access denied to bucket %s", bucket_name)
-            return False
-        else:
-            # Something else has gone wrong.
-            logger.error("Error trying to find object %s in bucket %s", key_name, bucket_name)
-            raise ValueError(f"Error trying to find object {key_name} in bucket {bucket_name}") from err
-    else:
-        # The object does exist.
-        return True
-
-
-def extract_bucket_name_and_key(object_path: str) -> tuple[str, str]:
-    match = re.match(r"s3:\/\/([^/]+)\/((?:[^/]*/)*.*)", object_path)
-    if match:
-        bucket_name = match.group(1)
-        key_name = match.group(2)
-    else:
-        logger.error("Unable to parse S3 object path %s", object_path)
-        raise ValueError(f"Unable to parse S3 object path {object_path}")
-    return bucket_name, key_name
-
-
-def get_lines_from_s3_file(aws_s3_client, s3_file_path: str) -> list[str]:
-    bucket, key = extract_bucket_name_and_key(s3_file_path)
-    response = aws_s3_client.get_object(Bucket=bucket, Key=key)
-    return response.get("Body").read().decode("utf-8").splitlines()
-
-
-def acm_list_all_certificates(acm_client, page_size: int = 50, interval_between_calls: float = 0.1) -> list[dict]:
-    """
-    Get a list of all the AWS Certificate Manager Certificates
-    https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/acm/paginator/ListCertificates.html
-    """
-    resources: list[dict] = []
-    paginator = acm_client.get_paginator("list_certificates")
-    # Override the default 'keyTypes' so that we get all the types
-    page_iterator = paginator.paginate(Includes={"keyTypes": []}, PaginationConfig={"PageSize": page_size})
-    for page in page_iterator:
-        resources.extend(page.get("CertificateSummaryList", []))
-        time.sleep(interval_between_calls)
-    return resources
-
-
-def acm_describe_certificate(acm_client, certificate_arn: str) -> dict | None:
-    try:
-        response: dict | None = acm_client.describe_certificate(CertificateArn=certificate_arn)
-        return None if not response else response.get("Certificate", None)
-    except botocore.exceptions.ClientError as ex:
-        # Scrub error message for any internal account info leaks
-        if "AccessDenied" in ex.response["Error"]["Code"]:
-            ex.response["Error"]["Message"] = "AWS Config does not have permission to assume the IAM role."
-        elif "ResourceNotFound" in ex.response["Error"]["Code"]:
-            ex.response["Error"]["Message"] = "ResourceNotFound Error calling 'describe_certificate'"
-        elif "InvalidArn" in ex.response["Error"]["Code"]:
-            ex.response["Error"]["Message"] = "InvalidArn Error calling 'describe_certificate'"
-        else:
-            ex.response["Error"]["Message"] = "InternalError"
-            ex.response["Error"]["Code"] = "InternalError"
-        raise ex
+# Logging setup
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
 
 
 def assess_certificate_manager_enforcement(
@@ -201,75 +42,57 @@ def assess_certificate_manager_enforcement(
     return evaluations, all_resources_are_compliant
 
 
-def submit_evaluations(
-    aws_config_client, result_token: str, evaluations: list[dict], interval_between_calls: float = 0.1
-):
-    max_evaluations_per_call = 100
-    while evaluations:
-        batch_of_evaluations, evaluations = (
-            evaluations[:max_evaluations_per_call],
-            evaluations[max_evaluations_per_call:],
-        )
-        aws_config_client.put_evaluations(Evaluations=batch_of_evaluations, ResultToken=result_token)
-        if evaluations:
-            time.sleep(interval_between_calls)
-
-
 def lambda_handler(event, context):
-    """Lambda handler to check CloudTrail trails are logging.
-    Keyword arguments:
-    event -- the event variable given in the lambda handler
-    context -- the context variable given in the lambda handler
     """
-    global logger
-    logger = logging.getLogger()
-    logger.setLevel(logging.INFO)
-
-    global AWS_ACCOUNT_ID
-    global AUDIT_ACCOUNT_ID
-    global EXECUTION_ROLE_NAME
-
-    page_size = 100
-    interval_between_api_calls = 0.1
+    This function is the main entry point for Lambda.
 
-    rule_parameters = json.loads(event.get("ruleParameters", "{}"))
-    invoking_event = json.loads(event.get("invokingEvent", "{}"))
-    logger.info("Received event: %s", json.dumps(event, indent=2))
+    Keyword arguments:
 
-    AWS_ACCOUNT_ID = event["accountId"]
-    logger.info("Assessing account %s", AWS_ACCOUNT_ID)
+    event -- the event variable given in the lambda handler
 
-    valid_rule_parameters = evaluate_parameters(rule_parameters)
-    EXECUTION_ROLE_NAME = valid_rule_parameters.get("ExecutionRoleName", "AWSA-GCLambdaExecutionRole")
-    AUDIT_ACCOUNT_ID = valid_rule_parameters.get("AuditAccountID", "")
+    context -- the context variable given in the lambda handler
+    """
+    logger.info("Received Event: %s", json.dumps(event, indent=2))
 
+    invoking_event = json.loads(event["invokingEvent"])
     if not is_scheduled_notification(invoking_event["messageType"]):
         logger.error("Skipping assessments as this is not a scheduled invocation")
         return
 
-    aws_config_client = get_client("config", event)
-    # Not using get_client to get S3 client for the Audit account
-    aws_s3_client = boto3.client("s3")
-    aws_acm_client = get_client("acm", event)
-
     file_param_name = "S3CasCurrentlyInUsePath"
-    cas_currently_in_use_file_path = valid_rule_parameters.get(file_param_name, "")
+    rule_parameters = check_required_parameters(
+        json.loads(event.get("ruleParameters", "{}")), ["ExecutionRoleName", file_param_name]
+    )
+    execution_role_name = rule_parameters.get("ExecutionRoleName")
+    audit_account_id = rule_parameters.get("AuditAccountID", "")
+    aws_account_id = event["accountId"]
+    is_not_audit_account = aws_account_id != audit_account_id
+
+    page_size = 100
+    interval_between_api_calls = 0.1
+
+    aws_config_client = get_client("config", aws_account_id, execution_role_name, is_not_audit_account)
+    # Get the S3 client for the current (Audit) account where this lambda runs from
+    aws_s3_client = get_client("s3")
+    aws_acm_client = get_client("acm", aws_account_id, execution_role_name, is_not_audit_account)
+
+    cas_currently_in_use_file_path = rule_parameters.get(file_param_name, "")
 
     if not check_s3_object_exists(aws_s3_client, cas_currently_in_use_file_path):
         annotation = (
             f"No file found for s3 path '{cas_currently_in_use_file_path}' via '{file_param_name}' input parameter."
         )
         logger.info(annotation)
-        evaluations = [build_evaluation(AWS_ACCOUNT_ID, "NON_COMPLIANT", event, ACCOUNT_RESOURCE_TYPE, annotation)]
-        aws_config_client.put_evaluations(Evaluations=evaluations, ResultToken=event["resultToken"])
+        evaluations = [build_evaluation(aws_account_id, "NON_COMPLIANT", event, annotation=annotation)]
+        submit_evaluations(aws_config_client, event["resultToken"], evaluations, interval_between_api_calls)
         return
 
     cas_currently_in_use = get_lines_from_s3_file(aws_s3_client, cas_currently_in_use_file_path)
     logger.info("cas_currently_in_use from the file in s3: %s", cas_currently_in_use)
 
-    certificates_summaries = acm_list_all_certificates(aws_acm_client, page_size, interval_between_api_calls)
+    certificates_summaries = list_all_acm_certificates(aws_acm_client, page_size, interval_between_api_calls)
     certificates_descriptions = [
-        acm_describe_certificate(aws_acm_client, x.get("CertificateArn", "")) for x in certificates_summaries
+        describe_acm_certificate(aws_acm_client, x.get("CertificateArn", "")) for x in certificates_summaries
     ]
 
     evaluations, all_acm_resources_are_compliant = assess_certificate_manager_enforcement(
@@ -284,5 +107,5 @@ def lambda_handler(event, context):
         annotation = "Non-compliant resources found in scope."
 
     logger.info(f"{compliance_type}: {annotation}")
-    evaluations.append(build_evaluation(AWS_ACCOUNT_ID, compliance_type, event, ACCOUNT_RESOURCE_TYPE, annotation))
+    evaluations.append(build_evaluation(aws_account_id, compliance_type, event, annotation=annotation))
     submit_evaluations(aws_config_client, event["resultToken"], evaluations, interval_between_api_calls)