remove/update/eval/defer todos

Author: cyphronix

aws_sra_examples/solutions/genai/bedrock_org/lambda/rules/sra_bedrock_check_cloudwatch_endpoints/app.py

@@ -21,11 +21,11 @@
 LOGGER.info(f"boto3 version: {boto3.__version__}")
 
 # Get AWS region from environment variable
-AWS_REGION = os.environ.get('AWS_REGION')
+AWS_REGION = os.environ.get("AWS_REGION")
 
 # Initialize AWS clients
-ec2_client = boto3.client('ec2', region_name=AWS_REGION)
-config_client = boto3.client('config', region_name=AWS_REGION)
+ec2_client = boto3.client("ec2", region_name=AWS_REGION)
+config_client = boto3.client("config", region_name=AWS_REGION)
 
 
 def evaluate_compliance(vpc_id: str) -> tuple[str, str]:
@@ -39,22 +39,19 @@ def evaluate_compliance(vpc_id: str) -> tuple[str, str]:
     """
     try:
         response = ec2_client.describe_vpc_endpoints(
-            Filters=[
-                {'Name': 'vpc-id', 'Values': [vpc_id]},
-                {'Name': 'service-name', 'Values': [f'com.amazonaws.{AWS_REGION}.logs']}
-            ]
+            Filters=[{"Name": "vpc-id", "Values": [vpc_id]}, {"Name": "service-name", "Values": [f"com.amazonaws.{AWS_REGION}.logs"]}]
         )
 
-        endpoints = response['VpcEndpoints']
+        endpoints = response["VpcEndpoints"]
 
         if endpoints:
-            endpoint_id = endpoints[0]['VpcEndpointId']
-            return 'COMPLIANT', f"CloudWatch gateway endpoint is in place for VPC {vpc_id}. Endpoint ID: {endpoint_id}"
-        return 'NON_COMPLIANT', f"No CloudWatch gateway endpoint found for VPC {vpc_id}"
+            endpoint_id = endpoints[0]["VpcEndpointId"]
+            return "COMPLIANT", f"CloudWatch gateway endpoint is in place for VPC {vpc_id}. Endpoint ID: {endpoint_id}"
+        return "NON_COMPLIANT", f"No CloudWatch gateway endpoint found for VPC {vpc_id}"
 
     except Exception as e:
         LOGGER.error(f"Error evaluating CloudWatch gateway endpoint for VPC {vpc_id}: {str(e)}")
-        return 'ERROR', f"Error evaluating compliance: {str(e)}"
+        return "ERROR", f"Error evaluating compliance: {str(e)}"
 
 
 def lambda_handler(event: dict, context: Any) -> None:  # noqa: U100
@@ -64,48 +61,49 @@ def lambda_handler(event: dict, context: Any) -> None:  # noqa: U100
         event (dict): Lambda event object
         context (Any): Lambda context object
     """
-    LOGGER.info('Evaluating compliance for AWS Config rule')
+    LOGGER.info("Evaluating compliance for AWS Config rule")
     LOGGER.info(f"Event: {json.dumps(event)}")
 
-    invoking_event = json.loads(event['invokingEvent'])
+    invoking_event = json.loads(event["invokingEvent"])
 
     evaluations = []
 
-    if invoking_event['messageType'] == 'ScheduledNotification':
+    if invoking_event["messageType"] == "ScheduledNotification":
         # This is a scheduled run, evaluate all VPCs
         vpcs = ec2_client.describe_vpcs()
-        for vpc in vpcs['Vpcs']:
-            vpc_id = vpc['VpcId']
+        for vpc in vpcs["Vpcs"]:
+            vpc_id = vpc["VpcId"]
             compliance_type, annotation = evaluate_compliance(vpc_id)
-            evaluations.append({
-                'ComplianceResourceType': 'AWS::EC2::VPC',
-                'ComplianceResourceId': vpc_id,
-                'ComplianceType': compliance_type,
-                'Annotation': annotation,
-                'OrderingTimestamp': invoking_event['notificationCreationTime']
-            })
+            evaluations.append(
+                {
+                    "ComplianceResourceType": "AWS::EC2::VPC",
+                    "ComplianceResourceId": vpc_id,
+                    "ComplianceType": compliance_type,
+                    "Annotation": annotation,
+                    "OrderingTimestamp": invoking_event["notificationCreationTime"],
+                }
+            )
     else:
         # This is a configuration change event
-        configuration_item = invoking_event['configurationItem']
-        if configuration_item['resourceType'] != 'AWS::EC2::VPC':
+        configuration_item = invoking_event["configurationItem"]
+        if configuration_item["resourceType"] != "AWS::EC2::VPC":
             LOGGER.info(f"Skipping non-VPC resource: {configuration_item['resourceType']}")
             return
 
-        vpc_id = configuration_item['resourceId']
+        vpc_id = configuration_item["resourceId"]
         compliance_type, annotation = evaluate_compliance(vpc_id)
-        evaluations.append({
-            'ComplianceResourceType': configuration_item['resourceType'],
-            'ComplianceResourceId': vpc_id,
-            'ComplianceType': compliance_type,
-            'Annotation': annotation,
-            'OrderingTimestamp': configuration_item['configurationItemCaptureTime']
-        })
+        evaluations.append(
+            {
+                "ComplianceResourceType": configuration_item["resourceType"],
+                "ComplianceResourceId": vpc_id,
+                "ComplianceType": compliance_type,
+                "Annotation": annotation,
+                "OrderingTimestamp": configuration_item["configurationItemCaptureTime"],
+            }
+        )
 
     # Submit compliance evaluations
     if evaluations:
-        config_client.put_evaluations(
-            Evaluations=evaluations,
-            ResultToken=event['resultToken']
-        )
+        config_client.put_evaluations(Evaluations=evaluations, ResultToken=event["resultToken"])
 
-    LOGGER.info(f"Compliance evaluation complete. Processed {len(evaluations)} evaluations.")
+    LOGGER.info(f"Compliance evaluation complete. Processed {len(evaluations)} evaluations.")

aws_sra_examples/solutions/genai/bedrock_org/lambda/rules/sra_bedrock_check_eval_job_bucket/app.py

@@ -43,73 +43,73 @@ def evaluate_compliance(event: dict, context: Any) -> tuple[str, str]:  # noqa:
     """
     LOGGER.info(f"Evaluate Compliance Event: {event}")
     # Initialize AWS clients
-    s3 = boto3.client('s3')
+    s3 = boto3.client("s3")
 
     # Get rule parameters
-    params = ast.literal_eval(event['ruleParameters'])
+    params = ast.literal_eval(event["ruleParameters"])
     LOGGER.info(f"Parameters: {params}")
-    bucket_name = params.get('BucketName', '')
-    check_retention = params.get('CheckRetention', 'true').lower() != 'false'
-    check_encryption = params.get('CheckEncryption', 'true').lower() != 'false'
-    check_logging = params.get('CheckLogging', 'true').lower() != 'false'
-    check_object_locking = params.get('CheckObjectLocking', 'true').lower() != 'false'
-    check_versioning = params.get('CheckVersioning', 'true').lower() != 'false'
+    bucket_name = params.get("BucketName", "")
+    check_retention = params.get("CheckRetention", "true").lower() != "false"
+    check_encryption = params.get("CheckEncryption", "true").lower() != "false"
+    check_logging = params.get("CheckLogging", "true").lower() != "false"
+    check_object_locking = params.get("CheckObjectLocking", "true").lower() != "false"
+    check_versioning = params.get("CheckVersioning", "true").lower() != "false"
 
     # Check if the bucket exists
     if not check_bucket_exists(bucket_name):
-        return build_evaluation('NOT_APPLICABLE', f"Bucket {bucket_name} does not exist or is not accessible")
+        return build_evaluation("NOT_APPLICABLE", f"Bucket {bucket_name} does not exist or is not accessible")
 
-    compliance_type = 'COMPLIANT'
+    compliance_type = "COMPLIANT"
     annotation = []
 
     # Check retention
     if check_retention:
         try:
             retention = s3.get_bucket_lifecycle_configuration(Bucket=bucket_name)
-            if not any(rule.get('Expiration') for rule in retention.get('Rules', [])):
-                compliance_type = 'NON_COMPLIANT'
+            if not any(rule.get("Expiration") for rule in retention.get("Rules", [])):
+                compliance_type = "NON_COMPLIANT"
                 annotation.append("Retention policy not set")
         except ClientError:
-            compliance_type = 'NON_COMPLIANT'
+            compliance_type = "NON_COMPLIANT"
             annotation.append("Retention policy not set")
 
     # Check encryption
     if check_encryption:
         try:
             encryption = s3.get_bucket_encryption(Bucket=bucket_name)
-            if 'ServerSideEncryptionConfiguration' not in encryption:
-                compliance_type = 'NON_COMPLIANT'
+            if "ServerSideEncryptionConfiguration" not in encryption:
+                compliance_type = "NON_COMPLIANT"
                 annotation.append("KMS CMK encryption not enabled")
         except ClientError:
-            compliance_type = 'NON_COMPLIANT'
+            compliance_type = "NON_COMPLIANT"
             annotation.append("KMS CMK encryption not enabled")
 
     # Check logging
     if check_logging:
         logging = s3.get_bucket_logging(Bucket=bucket_name)
-        if 'LoggingEnabled' not in logging:
-            compliance_type = 'NON_COMPLIANT'
+        if "LoggingEnabled" not in logging:
+            compliance_type = "NON_COMPLIANT"
             annotation.append("Server access logging not enabled")
 
     # Check object locking
     if check_object_locking:
         try:
             object_locking = s3.get_object_lock_configuration(Bucket=bucket_name)
-            if 'ObjectLockConfiguration' not in object_locking:
-                compliance_type = 'NON_COMPLIANT'
+            if "ObjectLockConfiguration" not in object_locking:
+                compliance_type = "NON_COMPLIANT"
                 annotation.append("Object locking not enabled")
         except ClientError:
-            compliance_type = 'NON_COMPLIANT'
+            compliance_type = "NON_COMPLIANT"
             annotation.append("Object locking not enabled")
 
     # Check versioning
     if check_versioning:
         versioning = s3.get_bucket_versioning(Bucket=bucket_name)
-        if versioning.get('Status') != 'Enabled':
-            compliance_type = 'NON_COMPLIANT'
+        if versioning.get("Status") != "Enabled":
+            compliance_type = "NON_COMPLIANT"
             annotation.append("Versioning not enabled")
 
-    annotation_str = '; '.join(annotation) if annotation else "All checked features are compliant"
+    annotation_str = "; ".join(annotation) if annotation else "All checked features are compliant"
     return build_evaluation(compliance_type, annotation_str)
 
 
@@ -122,10 +122,10 @@ def check_bucket_exists(bucket_name: str) -> Any:
     Returns:
         Any: True if the bucket exists and is accessible, False otherwise
     """
-    s3 = boto3.client('s3')
+    s3 = boto3.client("s3")
     try:
         response = s3.list_buckets()
-        buckets = [bucket['Name'] for bucket in response['Buckets']]
+        buckets = [bucket["Name"] for bucket in response["Buckets"]]
         return bucket_name in buckets
     except ClientError as e:
         LOGGER.info(f"An error occurred: {e}")
@@ -143,11 +143,7 @@ def build_evaluation(compliance_type: str, annotation: str) -> Any:
         Any: The evaluation compliance type and annotation
     """
     LOGGER.info(f"Build Evaluation Compliance Type: {compliance_type} Annotation: {annotation}")
-    return {
-        'ComplianceType': compliance_type,
-        'Annotation': annotation,
-        'OrderingTimestamp': datetime.now().isoformat()
-    }
+    return {"ComplianceType": compliance_type, "Annotation": annotation, "OrderingTimestamp": datetime.now().isoformat()}
 
 
 def lambda_handler(event: dict, context: Any) -> None:
@@ -160,17 +156,17 @@ def lambda_handler(event: dict, context: Any) -> None:
     LOGGER.info(f"Lambda Handler Context: {context}")
     LOGGER.info(f"Lambda Handler Event: {event}")
     evaluation = evaluate_compliance(event, context)
-    config = boto3.client('config')
-    params = ast.literal_eval(event['ruleParameters'])
+    config = boto3.client("config")
+    params = ast.literal_eval(event["ruleParameters"])
     config.put_evaluations(
         Evaluations=[
             {
-                'ComplianceResourceType': 'AWS::S3::Bucket',
-                'ComplianceResourceId': params.get('BucketName'),
-                'ComplianceType': evaluation['ComplianceType'],  # type: ignore
-                'Annotation': evaluation['Annotation'],  # type: ignore
-                'OrderingTimestamp': evaluation['OrderingTimestamp']  # type: ignore
+                "ComplianceResourceType": "AWS::S3::Bucket",
+                "ComplianceResourceId": params.get("BucketName"),
+                "ComplianceType": evaluation["ComplianceType"],  # type: ignore
+                "Annotation": evaluation["Annotation"],  # type: ignore
+                "OrderingTimestamp": evaluation["OrderingTimestamp"],  # type: ignore
             }
         ],
-        ResultToken=event['resultToken']
+        ResultToken=event["resultToken"],
     )

aws_sra_examples/solutions/genai/bedrock_org/lambda/rules/sra_bedrock_check_guardrail_encryption/app.py

@@ -21,11 +21,11 @@
 LOGGER.info(f"boto3 version: {boto3.__version__}")
 
 # Get AWS region from environment variable
-AWS_REGION = os.environ.get('AWS_REGION')
+AWS_REGION = os.environ.get("AWS_REGION")
 
 # Initialize AWS clients
-bedrock_client = boto3.client('bedrock', region_name=AWS_REGION)
-config_client = boto3.client('config', region_name=AWS_REGION)
+bedrock_client = boto3.client("bedrock", region_name=AWS_REGION)
+config_client = boto3.client("config", region_name=AWS_REGION)
 
 
 def evaluate_compliance(rule_parameters: dict) -> tuple[str, str]:  # noqa: CFQ004
@@ -40,27 +40,27 @@ def evaluate_compliance(rule_parameters: dict) -> tuple[str, str]:  # noqa: CFQ0
     LOGGER.info(f"Rule parameters: {json.dumps(rule_parameters)}")
     try:
         response = bedrock_client.list_guardrails()
-        guardrails = response.get('guardrails', [])
+        guardrails = response.get("guardrails", [])
 
         if not guardrails:
-            return 'NON_COMPLIANT', "No Bedrock guardrails found"
+            return "NON_COMPLIANT", "No Bedrock guardrails found"
 
         unencrypted_guardrails: list[str] = []
         for guardrail in guardrails:
-            guardrail_id = guardrail['id']
-            guardrail_name = guardrail['name']
+            guardrail_id = guardrail["id"]
+            guardrail_name = guardrail["name"]
             guardrail_detail = bedrock_client.get_guardrail(guardrailIdentifier=guardrail_id)
 
-            if 'kmsKeyArn' not in guardrail_detail:
+            if "kmsKeyArn" not in guardrail_detail:
                 unencrypted_guardrails.append(guardrail_name)
 
         if unencrypted_guardrails:
-            return 'NON_COMPLIANT', f"The following Bedrock guardrails are not encrypted with a KMS key: {', '.join(unencrypted_guardrails)}"
-        return 'COMPLIANT', "All Bedrock guardrails are encrypted with a KMS key"
+            return "NON_COMPLIANT", f"The following Bedrock guardrails are not encrypted with a KMS key: {', '.join(unencrypted_guardrails)}"
+        return "COMPLIANT", "All Bedrock guardrails are encrypted with a KMS key"
 
     except Exception as e:
         LOGGER.error(f"Error evaluating Bedrock guardrails encryption: {str(e)}")
-        return 'ERROR', f"Error evaluating compliance: {str(e)}"
+        return "ERROR", f"Error evaluating compliance: {str(e)}"
 
 
 def lambda_handler(event: dict, context: Any) -> None:  # noqa: U100
@@ -70,28 +70,25 @@ def lambda_handler(event: dict, context: Any) -> None:  # noqa: U100
         event (dict): Lambda event object
         context (Any): Lambda context object
     """
-    LOGGER.info('Evaluating compliance for AWS Config rule')
+    LOGGER.info("Evaluating compliance for AWS Config rule")
     LOGGER.info(f"Event: {json.dumps(event)}")
 
-    invoking_event = json.loads(event['invokingEvent'])
-    rule_parameters = json.loads(event['ruleParameters']) if 'ruleParameters' in event else {}
+    invoking_event = json.loads(event["invokingEvent"])
+    rule_parameters = json.loads(event["ruleParameters"]) if "ruleParameters" in event else {}
 
     compliance_type, annotation = evaluate_compliance(rule_parameters)
 
     evaluation = {
-        'ComplianceResourceType': 'AWS::::Account',
-        'ComplianceResourceId': event['accountId'],
-        'ComplianceType': compliance_type,
-        'Annotation': annotation,
-        'OrderingTimestamp': invoking_event['notificationCreationTime']
+        "ComplianceResourceType": "AWS::::Account",
+        "ComplianceResourceId": event["accountId"],
+        "ComplianceType": compliance_type,
+        "Annotation": annotation,
+        "OrderingTimestamp": invoking_event["notificationCreationTime"],
     }
 
     LOGGER.info(f"Compliance evaluation result: {compliance_type}")
     LOGGER.info(f"Annotation: {annotation}")
 
-    config_client.put_evaluations(
-        Evaluations=[evaluation],
-        ResultToken=event['resultToken']
-    )
+    config_client.put_evaluations(Evaluations=[evaluation], ResultToken=event["resultToken"])
 
-    LOGGER.info("Compliance evaluation complete.")
+    LOGGER.info("Compliance evaluation complete.")