AWSTemplateFormatVersion: '2010-09-09' Description: > Security Services and Audit Configuration Cost Review Remediator. Review-only module that inventories CloudTrail, AWS Config, GuardDuty, Security Hub, Inspector, Macie, and Detective across enabled Regions. This module intentionally does not disable, delete, or modify security services. Parameters: ProjectName: Type: String Default: security-services-audit-review Description: Short base name used for created resources. ContinueOnAccessDenied: Type: String Default: 'true' AllowedValues: ['true', 'false'] Description: Continue scanning when blocked by IAM/SCP. ExcludedRegionsCsv: Type: String Default: ap-northeast-3 Description: Comma-separated Regions to skip. ScheduleEnabled: Type: String Default: 'false' AllowedValues: ['true', 'false'] Description: Whether to create an enabled EventBridge schedule. ScheduleExpression: Type: String Default: rate(30 days) Description: EventBridge schedule expression used when ScheduleEnabled is true. Conditions: IsScheduleEnabled: !Equals [!Ref ScheduleEnabled, 'true'] Resources: ReviewRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: '2012-10-17' Statement: - Effect: Allow Principal: Service: lambda.amazonaws.com Action: sts:AssumeRole ManagedPolicyArns: - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole Policies: - PolicyName: SecurityServicesAuditReviewPolicy PolicyDocument: Version: '2012-10-17' Statement: - Sid: RegionDiscovery Effect: Allow Action: - ec2:DescribeRegions Resource: '*' - Sid: CloudTrailRead Effect: Allow Action: - cloudtrail:DescribeTrails - cloudtrail:GetTrailStatus - cloudtrail:GetEventSelectors - cloudtrail:GetInsightSelectors - cloudtrail:ListTrails Resource: '*' - Sid: ConfigRead Effect: Allow Action: - config:DescribeConfigurationRecorders - config:DescribeConfigurationRecorderStatus - config:DescribeDeliveryChannels - config:DescribeDeliveryChannelStatus - config:DescribeConformancePacks - config:DescribeConfigRules - config:DescribeAggregationAuthorizations - config:DescribeConfigurationAggregators Resource: '*' - Sid: GuardDutyRead Effect: Allow Action: - guardduty:ListDetectors - guardduty:GetDetector - guardduty:ListMembers Resource: '*' - Sid: SecurityHubRead Effect: Allow Action: - securityhub:DescribeHub - securityhub:GetEnabledStandards Resource: '*' - Sid: InspectorRead Effect: Allow Action: - inspector2:BatchGetAccountStatus Resource: '*' - Sid: MacieRead Effect: Allow Action: - macie2:GetMacieSession Resource: '*' - Sid: DetectiveRead Effect: Allow Action: - detective:ListGraphs Resource: '*' ReviewFunction: Type: AWS::Lambda::Function Properties: FunctionName: !Ref ProjectName Description: Review-only Lambda for security service and audit configuration cost visibility. Runtime: python3.12 Handler: index.handler Timeout: 900 MemorySize: 512 Role: !GetAtt ReviewRole.Arn Environment: Variables: CONTINUE_ON_ACCESS_DENIED: !Ref ContinueOnAccessDenied EXCLUDED_REGIONS_CSV: !Ref ExcludedRegionsCsv Code: ZipFile: | import json import os import traceback from datetime import datetime, timezone import boto3 from botocore.exceptions import ClientError, EndpointConnectionError def now(): return datetime.now(timezone.utc) def env_bool(name, default=False): return os.environ.get(name, str(default)).strip().lower() in ("1", "true", "yes", "y") def csv_set(name, default=""): raw = os.environ.get(name, default) or "" return {x.strip() for x in raw.split(",") if x.strip()} def error_record(service, operation, err): if isinstance(err, ClientError): return { "service": service, "operation": operation, "error_code": err.response.get("Error", {}).get("Code", "Unknown"), "error_message": err.response.get("Error", {}).get("Message", str(err)) } return { "service": service, "operation": operation, "error_code": type(err).__name__, "error_message": str(err) } def is_access_denied(err): text = str(err).lower() if isinstance(err, ClientError): code = err.response.get("Error", {}).get("Code", "").lower() msg = err.response.get("Error", {}).get("Message", "").lower() text = f"{code} {msg}" return "accessdenied" in text or "not authorized" in text or "explicit deny" in text or "service control policy" in text def safe_call(region_result, service_name, operation_name, func, default): try: return func() except EndpointConnectionError as e: region_result["failed"] += 1 region_result["failures"].append(error_record(service_name, operation_name, e)) return default except ClientError as e: if is_access_denied(e): region_result["access_denied"] += 1 else: region_result["failed"] += 1 region_result["failures"].append(error_record(service_name, operation_name, e)) return default except Exception as e: region_result["failed"] += 1 region_result["failures"].append(error_record(service_name, operation_name, e)) return default def get_enabled_regions(excluded): ec2 = boto3.client("ec2", region_name="us-east-1") response = ec2.describe_regions(AllRegions=True) regions = [] for r in response.get("Regions", []): status = r.get("OptInStatus") name = r.get("RegionName") if status in (None, "opt-in-not-required", "opted-in") and name not in excluded: regions.append(name) return sorted(regions) def scan_cloudtrail(region, rr): client = boto3.client("cloudtrail", region_name=region) trails = safe_call(rr, "CloudTrail", "DescribeTrails", lambda: client.describe_trails(includeShadowTrails=False).get("trailList", []), []) inventory = [] for trail in trails: name = trail.get("Name") arn = trail.get("TrailARN") status = safe_call(rr, "CloudTrail", "GetTrailStatus", lambda n=name: client.get_trail_status(Name=n), {}) event_selectors = safe_call(rr, "CloudTrail", "GetEventSelectors", lambda n=name: client.get_event_selectors(TrailName=n), {}) insight_selectors = safe_call(rr, "CloudTrail", "GetInsightSelectors", lambda n=name: client.get_insight_selectors(TrailName=n), {}) selectors = event_selectors.get("EventSelectors", []) or [] advanced = event_selectors.get("AdvancedEventSelectors", []) or [] inventory.append({ "name": name, "arn": arn, "home_region": trail.get("HomeRegion"), "s3_bucket_name": trail.get("S3BucketName"), "s3_key_prefix": trail.get("S3KeyPrefix"), "sns_topic_name": trail.get("SnsTopicName"), "cloudwatch_logs_log_group_arn": trail.get("CloudWatchLogsLogGroupArn"), "kms_key_id": trail.get("KmsKeyId"), "is_multi_region_trail": trail.get("IsMultiRegionTrail"), "is_organization_trail": trail.get("IsOrganizationTrail"), "log_file_validation_enabled": trail.get("LogFileValidationEnabled"), "is_logging": status.get("IsLogging"), "latest_delivery_error": status.get("LatestDeliveryError"), "latest_notification_error": status.get("LatestNotificationError"), "event_selectors_count": len(selectors), "advanced_event_selectors_count": len(advanced), "insight_selectors_count": len(insight_selectors.get("InsightSelectors", []) or []) }) rr["cloudtrail_trails"] = len(inventory) rr["cloudtrail_logging_trails"] = len([x for x in inventory if x.get("is_logging")]) rr["cloudtrail_multi_region_trails"] = len([x for x in inventory if x.get("is_multi_region_trail")]) rr["cloudtrail_org_trails"] = len([x for x in inventory if x.get("is_organization_trail")]) rr["cloudtrail_inventory"] = inventory def scan_config(region, rr): client = boto3.client("config", region_name=region) recorders = safe_call(rr, "Config", "DescribeConfigurationRecorders", lambda: client.describe_configuration_recorders().get("ConfigurationRecorders", []), []) recorder_status = safe_call(rr, "Config", "DescribeConfigurationRecorderStatus", lambda: client.describe_configuration_recorder_status().get("ConfigurationRecordersStatus", []), []) channels = safe_call(rr, "Config", "DescribeDeliveryChannels", lambda: client.describe_delivery_channels().get("DeliveryChannels", []), []) channel_status = safe_call(rr, "Config", "DescribeDeliveryChannelStatus", lambda: client.describe_delivery_channel_status().get("DeliveryChannelsStatus", []), []) rules = safe_call(rr, "Config", "DescribeConfigRules", lambda: client.describe_config_rules().get("ConfigRules", []), []) packs = safe_call(rr, "Config", "DescribeConformancePacks", lambda: client.describe_conformance_packs().get("ConformancePackDetails", []), []) aggregators = safe_call(rr, "Config", "DescribeConfigurationAggregators", lambda: client.describe_configuration_aggregators().get("ConfigurationAggregators", []), []) agg_auths = safe_call(rr, "Config", "DescribeAggregationAuthorizations", lambda: client.describe_aggregation_authorizations().get("AggregationAuthorizations", []), []) active_recorders = [x for x in recorder_status if x.get("recording")] rr["config_recorders"] = len(recorders) rr["config_recorders_recording"] = len(active_recorders) rr["config_delivery_channels"] = len(channels) rr["config_rules"] = len(rules) rr["config_conformance_packs"] = len(packs) rr["config_aggregators"] = len(aggregators) rr["config_aggregation_authorizations"] = len(agg_auths) rr["config_inventory"] = { "recorders": [{"name": x.get("name"), "role_arn": x.get("roleARN"), "recording_group": x.get("recordingGroup")} for x in recorders], "recorder_status": recorder_status, "delivery_channels": [{"name": x.get("name"), "s3_bucket_name": x.get("s3BucketName"), "sns_topic_arn": x.get("snsTopicARN")} for x in channels], "delivery_channel_status": channel_status, "rules_count": len(rules), "conformance_packs_count": len(packs), "aggregators_count": len(aggregators), "aggregation_authorizations_count": len(agg_auths) } def scan_guardduty(region, rr): client = boto3.client("guardduty", region_name=region) detector_ids = safe_call(rr, "GuardDuty", "ListDetectors", lambda: client.list_detectors().get("DetectorIds", []), []) inventory = [] member_count = 0 for detector_id in detector_ids: det = safe_call(rr, "GuardDuty", "GetDetector", lambda d=detector_id: client.get_detector(DetectorId=d), {}) members = safe_call(rr, "GuardDuty", "ListMembers", lambda d=detector_id: client.list_members(DetectorId=d, OnlyAssociated="true").get("Members", []), []) member_count += len(members) inventory.append({ "detector_id": detector_id, "status": det.get("Status"), "finding_publishing_frequency": det.get("FindingPublishingFrequency"), "features": det.get("Features", []), "associated_members": len(members) }) rr["guardduty_detectors"] = len(detector_ids) rr["guardduty_enabled_detectors"] = len([x for x in inventory if x.get("status") == "ENABLED"]) rr["guardduty_members"] = member_count rr["guardduty_inventory"] = inventory def scan_securityhub(region, rr): client = boto3.client("securityhub", region_name=region) hub = safe_call(rr, "SecurityHub", "DescribeHub", lambda: client.describe_hub(), None) standards = [] if hub: standards = safe_call(rr, "SecurityHub", "GetEnabledStandards", lambda: client.get_enabled_standards().get("StandardsSubscriptions", []), []) rr["securityhub_enabled"] = bool(hub) rr["securityhub_enabled_standards"] = len(standards) rr["securityhub_inventory"] = { "hub_arn": hub.get("HubArn") if hub else None, "auto_enable_controls": hub.get("AutoEnableControls") if hub else None, "control_finding_generator": hub.get("ControlFindingGenerator") if hub else None, "enabled_standards": [{"arn": s.get("StandardsSubscriptionArn"), "status": s.get("StandardsStatus")} for s in standards] } def scan_inspector(region, rr): client = boto3.client("inspector2", region_name=region) status = safe_call(rr, "Inspector2", "BatchGetAccountStatus", lambda: client.batch_get_account_status(), {}) accounts = status.get("accounts", []) or [] enabled_accounts = 0 for acct in accounts: resource_state = acct.get("resourceState", {}) or {} if any(v.get("status") == "ENABLED" for v in resource_state.values() if isinstance(v, dict)): enabled_accounts += 1 rr["inspector_accounts_reported"] = len(accounts) rr["inspector_enabled_accounts"] = enabled_accounts rr["inspector_inventory"] = accounts def scan_macie(region, rr): client = boto3.client("macie2", region_name=region) try: session = client.get_macie_session() rr["macie_enabled"] = bool(session and session.get("status") == "ENABLED") rr["macie_status"] = session.get("status") if session else None rr["macie_inventory"] = session or {} except ClientError as e: code = e.response.get("Error", {}).get("Code", "Unknown") msg = e.response.get("Error", {}).get("Message", str(e)) if code == "AccessDeniedException" and "macie is not enabled" in msg.lower(): rr["macie_enabled"] = False rr["macie_status"] = "NOT_ENABLED" rr["macie_inventory"] = {} elif is_access_denied(e): rr["access_denied"] += 1 rr["failures"].append(error_record("Macie2", "GetMacieSession", e)) else: rr["failed"] += 1 rr["failures"].append(error_record("Macie2", "GetMacieSession", e)) def scan_detective(region, rr): client = boto3.client("detective", region_name=region) graphs = safe_call(rr, "Detective", "ListGraphs", lambda: client.list_graphs().get("GraphList", []), []) rr["detective_graphs"] = len(graphs) rr["detective_inventory"] = graphs def summarize_review_flags(rr): flags = [] if rr.get("cloudtrail_trails", 0) > 1: flags.append("multiple_cloudtrail_trails") if rr.get("cloudtrail_logging_trails", 0) == 0 and rr.get("cloudtrail_trails", 0) > 0: flags.append("cloudtrail_not_logging") if rr.get("config_recorders_recording", 0) > 0: flags.append("config_recording_enabled") if rr.get("guardduty_enabled_detectors", 0) > 0: flags.append("guardduty_enabled") if rr.get("securityhub_enabled"): flags.append("securityhub_enabled") if rr.get("inspector_enabled_accounts", 0) > 0: flags.append("inspector_enabled") if rr.get("macie_enabled"): flags.append("macie_enabled") if rr.get("detective_graphs", 0) > 0: flags.append("detective_enabled") return flags def handler(event, context): excluded = csv_set("EXCLUDED_REGIONS_CSV", "ap-northeast-3") continue_on_access_denied = env_bool("CONTINUE_ON_ACCESS_DENIED", True) summary = { "started_at": now().isoformat(), "review_only": True, "excluded_regions": sorted(list(excluded)), "totals": { "regions_scanned": 0, "cloudtrail_trails": 0, "cloudtrail_logging_trails": 0, "cloudtrail_multi_region_trails": 0, "cloudtrail_org_trails": 0, "config_recorders": 0, "config_recorders_recording": 0, "config_delivery_channels": 0, "config_rules": 0, "config_conformance_packs": 0, "config_aggregators": 0, "config_aggregation_authorizations": 0, "guardduty_detectors": 0, "guardduty_enabled_detectors": 0, "guardduty_members": 0, "securityhub_enabled_regions": 0, "securityhub_enabled_standards": 0, "inspector_accounts_reported": 0, "inspector_enabled_accounts": 0, "macie_enabled_regions": 0, "detective_graphs": 0, "cost_review_regions": 0, "access_denied": 0, "failed": 0 }, "region_results": [], "failures": [] } try: regions = get_enabled_regions(excluded) except Exception as e: summary["totals"]["failed"] += 1 summary["failures"].append(error_record("EC2", "DescribeRegions", e)) summary["ended_at"] = now().isoformat() print(json.dumps(summary, indent=2, default=str)) return json.loads(json.dumps(summary, default=str)) for region in regions: rr = { "region": region, "cloudtrail_trails": 0, "cloudtrail_logging_trails": 0, "cloudtrail_multi_region_trails": 0, "cloudtrail_org_trails": 0, "config_recorders": 0, "config_recorders_recording": 0, "config_delivery_channels": 0, "config_rules": 0, "config_conformance_packs": 0, "config_aggregators": 0, "config_aggregation_authorizations": 0, "guardduty_detectors": 0, "guardduty_enabled_detectors": 0, "guardduty_members": 0, "securityhub_enabled": False, "securityhub_enabled_standards": 0, "inspector_accounts_reported": 0, "inspector_enabled_accounts": 0, "macie_enabled": False, "macie_status": None, "detective_graphs": 0, "review_flags": [], "access_denied": 0, "failed": 0, "failures": [] } try: scan_cloudtrail(region, rr) scan_config(region, rr) scan_guardduty(region, rr) scan_securityhub(region, rr) scan_inspector(region, rr) scan_macie(region, rr) scan_detective(region, rr) rr["review_flags"] = summarize_review_flags(rr) except Exception as e: rr["failed"] += 1 rr["failures"].append({ "service": "RegionScan", "operation": "scan_region", "error_code": type(e).__name__, "error_message": str(e), "trace": traceback.format_exc() }) if not continue_on_access_denied: raise for key in [ "cloudtrail_trails", "cloudtrail_logging_trails", "cloudtrail_multi_region_trails", "cloudtrail_org_trails", "config_recorders", "config_recorders_recording", "config_delivery_channels", "config_rules", "config_conformance_packs", "config_aggregators", "config_aggregation_authorizations", "guardduty_detectors", "guardduty_enabled_detectors", "guardduty_members", "securityhub_enabled_standards", "inspector_accounts_reported", "inspector_enabled_accounts", "detective_graphs", "access_denied", "failed" ]: summary["totals"][key] += int(rr.get(key, 0) or 0) if rr.get("securityhub_enabled"): summary["totals"]["securityhub_enabled_regions"] += 1 if rr.get("macie_enabled"): summary["totals"]["macie_enabled_regions"] += 1 if rr.get("review_flags"): summary["totals"]["cost_review_regions"] += 1 summary["totals"]["regions_scanned"] += 1 summary["failures"].extend([dict(f, region=region) for f in rr.get("failures", [])]) summary["region_results"].append(rr) summary["ended_at"] = now().isoformat() print(json.dumps(summary, indent=2, default=str)) return json.loads(json.dumps(summary, default=str)) ReviewScheduleRule: Type: AWS::Events::Rule Condition: IsScheduleEnabled Properties: Name: !Sub '${ProjectName}-schedule' Description: Scheduled run for Security Services Audit Review Remediator. ScheduleExpression: !Ref ScheduleExpression State: ENABLED Targets: - Arn: !GetAtt ReviewFunction.Arn Id: SecurityServicesAuditReviewTarget Input: '{"source":"eventbridge-schedule"}' ReviewSchedulePermission: Type: AWS::Lambda::Permission Condition: IsScheduleEnabled Properties: FunctionName: !Ref ReviewFunction Action: lambda:InvokeFunction Principal: events.amazonaws.com SourceArn: !GetAtt ReviewScheduleRule.Arn Outputs: ReviewFunctionName: Description: Lambda function name. Value: !Ref ReviewFunction ReviewFunctionArn: Description: Lambda function ARN. Value: !GetAtt ReviewFunction.Arn ReviewOnly: Description: This module is intentionally review-only. Value: 'true'