GR03

Author: a-shokri-ssc

src/lambda/gc03_check_iam_cloudwatch_alarms/app.py

@@ -1,32 +1,127 @@
-""" GC03 - Check trusted devices admin access
-    Confirm that administrative access to cloud environments is from approved and trusted locations and devices
+""" GC03 - Check IAM/ CloudWatch Alarms
+    https://canada-ca.github.io/cloud-guardrails/EN/03_Cloud-Console-Access.html
 """
 
 import json
 import logging
-import ipaddress
-import datetime
 
 from utils import is_scheduled_notification, check_required_parameters, check_guardrail_requirement_by_cloud_usage_profile, get_cloud_profile_from_tags, GuardrailType, GuardrailRequirementType
 from boto_util.organizations import get_account_tags, get_organizations_mgmt_account_id
 from boto_util.client import get_client
 from boto_util.config import build_evaluation, submit_evaluations
-from boto_util.iam import account_has_federated_users
-from boto_util.s3 import check_s3_object_exists, get_lines_from_s3_file
-from boto_util.cloud_trail import lookup_cloud_trail_events
+
+import botocore.exceptions
 
 # Logging setup
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
 
-def ip_is_within_ranges(ip_addr: str, ip_cidr_ranges: list[str]) -> bool:
-    """Return true if the given IP Address is within the at least one of the given CIDR ranges, otherwise returns false"""
-    for ip_range in ip_cidr_ranges:
-        ip_network = ipaddress.ip_network(ip_range)
-        if ipaddress.ip_address(ip_addr) in ip_network:
-            return True
-    return False
+def check_cloudwatch_alarms(
+    cloudwatch_client,
+    alarm_names=[
+        "AWS-IAM-Authentication-From-Unapproved-IP",
+        "AWS-SSO-Authentication-From-Unapproved-IP",
+        "AWS-Console-SignIn-Without-MFA",
+        "AWSAccelerator-AWS-IAM-Authentication-From-Unapproved-IP",
+        "AWSAccelerator-AWS-SSO-Authentication-From-Unapproved-IP",
+        "AWSAccelerator-AWS-Console-SignIn-Without-MFA"        
+    ],
+):
+    """Check CloudWatch alarms for compliance.
+    Keyword arguments:
+    alarm_names -- the list of CloudWatch alarms to check
+    """
+    result = {"status": "NON_COMPLIANT", "annotation": "No alarms found"}
+    if len(alarm_names) < 1:
+        # no alarms to check
+        result = {
+            "status": "COMPLIANT",
+            "annotation": "No alarms checked for compliance",
+        }
+        return result
+    # initialize our lists
+    alarms_not_found = alarm_names
+    alarms_found = []
+    try:
+        # describe CloudWatch alarms
+        response = cloudwatch_client.describe_alarms(
+            AlarmNames=alarm_names,
+            AlarmTypes=["MetricAlarm"],
+        )
+        # results may be paginated, and we may have to retry
+        b_more_data = True
+        i_retries = 0
+        i_retry_limit = 10
+        next_token = ""
+        while b_more_data and (i_retries < i_retry_limit):
+            # did we get a response?
+            if response:
+                # yes
+                alarms_found.extend(response.get("MetricAlarms"))
+                # results paginated?
+                next_token = response.get("NextToken")
+                if next_token:
+                    # yes
+                    response = cloudwatch_client.describe_alarms(
+                        AlarmNames=alarm_names,
+                        AlarmTypes=["MetricAlarm"],
+                        NextToken=next_token,
+                    )
+                else:
+                    # no more data
+                    b_more_data = False
+            else:
+                logger.error("Empty response. Retry call.")
+                i_retries += 1
+                if next_token:
+                    response = cloudwatch_client.describe_alarms(
+                        AlarmNames=alarm_names,
+                        AlarmTypes=["MetricAlarm"],
+                        NextToken=next_token,
+                    )
+                else:
+                    response = cloudwatch_client.describe_alarms(
+                        AlarmNames=alarm_names,
+                        AlarmTypes=["MetricAlarm"],
+                    )
+        # did we time out trying?
+        if i_retries >= i_retry_limit:
+            # yes
+            result["annotation"] = "Empty response while trying describe_alarms in CloudWatch API."
+            return result
+    except botocore.exceptions.ClientError as error:
+        logger.error("Error while trying to describe_alarms - boto3 Client error - %s", error)
+        result["annotation"] = "Error while trying to describe_alarms."
+        return result
+
+    # checking the alarms we found
+    alarms_not_found_set = set(alarms_not_found)
+    for alarm in alarms_found:
+        if not alarms_not_found_set:
+            # All alarms have been found, exit the loop
+            break
+        alarm_name = alarm.get("AlarmName")
+        if alarm_name:
+            for not_found_alarm in alarms_not_found_set:
+                if not_found_alarm in alarm_name:
+                    logger.info("CloudWatch Alarm %s found.", alarm_name)
+                    alarms_not_found_set.remove(not_found_alarm)
+
+                    # Stop the inner loop as we found a match
+                    break
+
+    # prepare the annotation (if needed)
+    if len(alarms_not_found_set) > 0:
+        annotation = "Alarms not found: "
+        for alarm in alarms_not_found_set:
+            annotation += f"{alarm}; "
+        result["annotation"] = annotation
+    else:
+        result = {"status": "COMPLIANT", "annotation": "All alarms found"}
+
+    logger.info(result)
+    return result
 
 
 def lambda_handler(event, context):
@@ -47,7 +142,7 @@ def lambda_handler(event, context):
         return
 
     rule_parameters = check_required_parameters(
-        json.loads(event.get("ruleParameters", "{}")), ["ExecutionRoleName", "s3ObjectPath"]
+        json.loads(event.get("ruleParameters", "{}")), ["ExecutionRoleName", "AlarmList"]
     )
     execution_role_name = rule_parameters.get("ExecutionRoleName")
     audit_account_id = rule_parameters.get("AuditAccountID", "")
@@ -56,111 +151,64 @@ def lambda_handler(event, context):
 
     evaluations = []
 
-    aws_organizations_client = get_client("organizations", aws_account_id, execution_role_name)
+    try:
+        client = get_client("organizations")
+        response = client.describe_account(AccountId=aws_account_id)
+        account_status = response["Account"]["Status"]
 
-    if aws_account_id != get_organizations_mgmt_account_id(aws_organizations_client):
-        logger.info(
-            "Cloud Trail events not checked in account %s as this is not the Management Account", aws_account_id
-        )
-        return
+        logger.info(f"account_status is {account_status}")
 
-    aws_config_client = get_client("config", aws_account_id, execution_role_name)
-    aws_s3_client = get_client("s3")
-    aws_cloudtrail_client = get_client("cloudtrail", aws_account_id, execution_role_name)
-    aws_iam_client = get_client("iam", aws_account_id, execution_role_name)
-
-    # Check cloud profile
-    tags = get_account_tags(get_client("organizations", assume_role=False), aws_account_id)
-    cloud_profile = get_cloud_profile_from_tags(tags)
-    gr_requirement_type = check_guardrail_requirement_by_cloud_usage_profile(GuardrailType.Guardrail3, cloud_profile)
-    
-    # If the guardrail is recommended
-    if gr_requirement_type == GuardrailRequirementType.Recommended:
-        return submit_evaluations(aws_config_client, event, [build_evaluation(
-            aws_account_id,
-            "COMPLIANT",
-            event,
-            gr_requirement_type=gr_requirement_type
-        )])
-    # If the guardrail is not required
-    elif gr_requirement_type == GuardrailRequirementType.Not_Required:
-        return submit_evaluations(aws_config_client, event, [build_evaluation(
-            aws_account_id,
-            "NOT_APPLICABLE",
-            event,
-            gr_requirement_type=gr_requirement_type
-        )])
+        if account_status != "ACTIVE":
+            return
+
+        aws_organizations_client = get_client("organizations", aws_account_id, execution_role_name)
+            
+        if aws_account_id != get_organizations_mgmt_account_id(aws_organizations_client):
+            logger.info(
+                "CloudWatch Alarms not checked in account %s as this is not the Management Account",
+                aws_account_id,
+            )
+            return
 
-    file_param_name = "s3ObjectPath"
-    vpn_ip_ranges_file_path = rule_parameters.get(file_param_name, "")
 
-    if not check_s3_object_exists(aws_s3_client, vpn_ip_ranges_file_path):
-        annotation = f"No file found for s3 path '{vpn_ip_ranges_file_path}' via '{file_param_name}' input parameter."
-        logger.info(annotation)
-        evaluations.append(build_evaluation(aws_account_id, "NON_COMPLIANT", event, annotation=annotation))
-        submit_evaluations(aws_config_client, event, evaluations)
-        return
+        aws_config_client = get_client("config", aws_account_id, execution_role_name)
+        aws_cloudwatch_client = get_client("cloudwatch", aws_account_id, execution_role_name)
 
-    vpn_ip_ranges = get_lines_from_s3_file(aws_s3_client, vpn_ip_ranges_file_path)
-    logger.info("vpn_ip_ranges from the file in s3: %s", vpn_ip_ranges)
+        # Check cloud profile
+        tags = get_account_tags(get_client("organizations", assume_role=False), aws_account_id)
+        cloud_profile = get_cloud_profile_from_tags(tags)
+        gr_requirement_type = check_guardrail_requirement_by_cloud_usage_profile(GuardrailType.Guardrail3, cloud_profile)
+        
+        # If the guardrail is recommended
+        if gr_requirement_type == GuardrailRequirementType.Recommended:
+            return submit_evaluations(aws_config_client, event, [build_evaluation(
+                aws_account_id,
+                "COMPLIANT",
+                event,
+                gr_requirement_type=gr_requirement_type
+            )])
+        # If the guardrail is not required
+        elif gr_requirement_type == GuardrailRequirementType.Not_Required:
+            return submit_evaluations(aws_config_client, event, [build_evaluation(
+                aws_account_id,
+                "NOT_APPLICABLE",
+                event,
+                gr_requirement_type=gr_requirement_type
+            )])
+            
+        results = check_cloudwatch_alarms(
+            aws_cloudwatch_client, alarm_names=str(rule_parameters["AlarmList"]).split(",")
+        )
+        if results:
+            compliance_type = results.get("status")
+            annotation = results.get("annotation")
+        else:
+            compliance_type = "NON_COMPLIANT"
+            annotation = "Unable to assess CloudWatch Alarms"
 
-    if not vpn_ip_ranges:
-        annotation = "No ip ranges found in input file."
-        logger.info(annotation)
-        evaluations.append(build_evaluation(aws_account_id, "NON_COMPLIANT", event, annotation=annotation))
+        logger.info(f"{compliance_type}: {annotation}")
+        evaluations.append(build_evaluation(aws_account_id, compliance_type, event, annotation=annotation))
         submit_evaluations(aws_config_client, event, evaluations)
-        return
-
-    bg_account_names = [rule_parameters["BgUser1"], rule_parameters["BgUser2"]]
-    lookup_attributes = [{"AttributeKey": "EventName", "AttributeValue": "ConsoleLogin"}]
-    console_login_cloud_trail_events = lookup_cloud_trail_events(aws_cloudtrail_client, lookup_attributes)
-    print("INFO 1:", console_login_cloud_trail_events)
-    cloud_trail_events = [e for e in console_login_cloud_trail_events if e.get("Username") not in bg_account_names]
-    num_compliant_rules = 0
-    logger.info("Number of events found: %s", len(cloud_trail_events))
-
-    for lookup_event in cloud_trail_events:
-        ct_event = json.loads(lookup_event.get("CloudTrailEvent", "{}"))
-        ct_event_id = ct_event.get("eventID", "")
-        ct_event_time = ct_event.get("eventTime", "")
-
-        # get event time and convert to ISO format
-        ct_event_isotime = datetime.datetime.fromisoformat(ct_event_time.replace('Z', '+00:00'))
-        now = datetime.datetime.now(datetime.timezone.utc)
-        # find difference from the current/execution time
-        time_diff = now - ct_event_isotime
-        # this is one day in ISO format
-        one_day = datetime.timedelta(days=1)
-
-        if time_diff > one_day:
-            # skip if event entry is older than 1 day
-            pass
-        elif not ip_is_within_ranges(ct_event["sourceIPAddress"], vpn_ip_ranges):
-            compliance_type = "NON_COMPLIANT"
-            annotation = f"Cloud Trail Event '{ct_event_id}' has a source IP address OUTSIDE of the allowed ranges, with event time '{ct_event_time}'."
-            logger.info(f"{compliance_type}: {annotation}")
-            evaluations.append(build_evaluation(ct_event_id, compliance_type, event, "AWS::CloudTrail::Trail", annotation))
-        else:
-            num_compliant_rules = num_compliant_rules + 1
-            compliance_type = "COMPLIANT"
-            annotation = f"Cloud Trail Event '{ct_event_id}' has a source IP address inside of the allowed ranges."
-            if account_has_federated_users(aws_iam_client):
-                annotation = f"{annotation} Dependent on the compliance of the Federated IdP."
-            logger.info(f"{compliance_type}: {annotation}")
-            evaluations.append(build_evaluation(ct_event_id, compliance_type, event, "AWS::CloudTrail::Trail", annotation))
-
-        #logger.info(f"{compliance_type}: {annotation}")
-        #evaluations.append(build_evaluation(ct_event_id, compliance_type, event, "AWS::CloudTrail::Trail", annotation))
-
-    if len(cloud_trail_events) == num_compliant_rules:
-        compliance_type = "COMPLIANT"
-        annotation = "All Cloud Trail Events are within the allowed source IP address ranges."
-        if account_has_federated_users(aws_iam_client):
-            annotation = f"All Cloud Trail Events are within the allowed source IP address ranges or are dependant on the federated identity provider."
-    else:
-        compliance_type = "NON_COMPLIANT"
-        annotation = "NOT all Cloud Trail Events are within the allowed source IP address ranges."
 
-    logger.info(f"{compliance_type}: {annotation}")
-    evaluations.append(build_evaluation(aws_account_id, compliance_type, event, annotation=annotation))
-    submit_evaluations(aws_config_client, event, evaluations)
+    except:
+        logger.info("This account Id is not active. Compliance evaluation not available for suspended accounts")