I summarized the configuration of security services in an AWS multi-account environment

# AWS Control Tower Multi-Account Security Services Configuration Patterns ## Overview This guide explains effective configuration patterns for security services in a multi-account environment using AWS Control Tower. --- ## Architecture Overview ``` Management Account ├── Control Tower (Landing Zone) ├── Log Archive Account │ └── Centralized log storage (S3, CloudWatch Logs) └── Audit Account (Security Tooling Account) └── Centralized security monitoring Member Accounts (Workload Accounts) ├── Dev Account ├── Staging Account └── Production Account ``` --- ## 1. AWS CloudTrail ### Configuration Pattern ``` Organization Trail (Management Account) ├── Enabled at organization level (applies to all member accounts) ├── Log Storage: Log Archive Account S3 Bucket ├── CloudWatch Logs: Audit Account └── KMS Encryption: Customer Managed Key ``` ### Implementation ```yaml # CloudFormation - Organization Trail Resources: OrganizationTrail: Type: AWS::CloudTrail::Trail Properties: TrailName: organization-trail S3BucketName: !Ref CentralLogBucket IsLogging: true IsMultiRegionTrail: true IsOrganizationTrail: true EnableLogFileValidation: true IncludeGlobalServiceEvents: true CloudWatchLogsLogGroupArn: !GetAtt TrailLogGroup.Arn CloudWatchLogsRoleArn: !GetAtt TrailRole.Arn KMSKeyId: !Ref TrailKMSKey EventSelectors: - ReadWriteType: All IncludeManagementEvents: true DataResources: - Type: AWS::S3::Object Values: - "arn:aws:s3:::" - Type: AWS::Lambda::Function Values: - "arn:aws:lambda" ``` ### Log Archive Account S3 Bucket Policy ```json { "Version": "2012-10-17", "Statement": [ { "Sid": "AWSCloudTrailAclCheck", "Effect": "Allow", "Principal": { "Service": "cloudtrail.amazonaws.com" }, "Action": "s3:GetBucketAcl", "Resource": "arn:aws:s3:::central-cloudtrail-logs", "Condition": { "StringEquals": { "AWS:SourceArn": "arn:aws:cloudtrail:us-east-1:MANAGEMENT_ACCOUNT_ID:trail/organization-trail" } } }, { "Sid": "AWSCloudTrailWrite", "Effect": "Allow", "Principal": { "Service": "cloudtrail.amazonaws.com" }, "Action": "s3:PutObject", "Resource": "arn:aws:s3:::central-cloudtrail-logs/AWSLogs/*", "Condition": { "StringEquals": { "s3:x-amz-acl": "bucket-owner-full-control", "AWS:SourceArn": "arn:aws:cloudtrail:us-east-1:MANAGEMENT_ACCOUNT_ID:trail/organization-trail" } } } ] } ``` --- ## 2. AWS Config ### Configuration Pattern ``` Organization Config (Management Account) ├── Aggregator: Collects data from all member accounts ├── Conformance Packs: Applied organization-wide ├── Config Rules: Deployed via Control Tower SCPs └── Recording: Enabled in each member account └── Log Storage: Log Archive Account S3 Bucket ``` ### Configuration Aggregator ```yaml # Audit Account - Config Aggregator Resources: OrganizationConfigAggregator: Type: AWS::Config::ConfigurationAggregator Properties: ConfigurationAggregatorName: organization-aggregator OrganizationAggregationSource: RoleArn: !GetAtt ConfigAggregatorRole.Arn AllAwsRegions: true ``` ### Conformance Pack ```yaml # Organization Conformance Pack Resources: SecurityConformancePack: Type: AWS::Config::OrganizationConformancePack Properties: OrganizationConformancePackName: security-baseline DeliveryS3Bucket: !Ref ConformancePackBucket ExcludedAccounts: - !Ref ManagementAccountId TemplateBody: | Parameters: {} Rules: - id: ROOT_ACCOUNT_MFA_ENABLED type: AWS::Config::ConfigRule properties: configRuleName: root-account-mfa-enabled source: owner: AWS sourceIdentifier: ROOT_ACCOUNT_MFA_ENABLED - id: IAM_PASSWORD_POLICY type: AWS::Config::ConfigRule properties: configRuleName: iam-password-policy source: owner: AWS sourceIdentifier: IAM_PASSWORD_POLICY inputParameters: RequireUppercaseCharacters: "true" RequireLowercaseCharacters: "true" RequireSymbols: "true" RequireNumbers: "true" MinimumPasswordLength: "14" PasswordReusePrevention: "24" MaxPasswordAge: "90" - id: S3_BUCKET_PUBLIC_ACCESS_PROHIBITED type: AWS::Config::ConfigRule properties: configRuleName: s3-bucket-public-access-prohibited source: owner: AWS sourceIdentifier: S3_BUCKET_PUBLIC_ACCESS_PROHIBITED - id: ENCRYPTED_VOLUMES type: AWS::Config::ConfigRule properties: configRuleName: encrypted-volumes source: owner: AWS sourceIdentifier: ENCRYPTED_VOLUMES ``` ### Auto Remediation ```yaml # Config Rule with Auto Remediation Resources: S3PublicAccessRule: Type: AWS::Config::ConfigRule Properties: ConfigRuleName: s3-bucket-public-access-prohibited Source: Owner: AWS SourceIdentifier: S3_BUCKET_PUBLIC_ACCESS_PROHIBITED S3PublicAccessRemediation: Type: AWS::Config::RemediationConfiguration Properties: ConfigRuleName: !Ref S3PublicAccessRule TargetType: SSM_DOCUMENT TargetId: AWS-DisableS3BucketPublicReadWrite Automatic: true MaximumAutomaticAttempts: 3 RetryAttemptSeconds: 60 Parameters: AutomationAssumeRole: StaticValue: Values: - !GetAtt RemediationRole.Arn S3BucketName: ResourceValue: Value: RESOURCE_ID ``` --- ## 3. AWS Security Hub ### Configuration Pattern ``` Security Hub (Audit Account - Delegated Administrator) ├── Organization-wide auto-enable ├── Standards Enabled: │ ├── AWS Foundational Security Best Practices (FSBP) │ ├── CIS AWS Foundations Benchmark │ ├── PCI DSS (if required) │ └── NIST SP 800-53 (if required) ├── Findings Integration: │ ├── GuardDuty → Security Hub │ ├── Inspector → Security Hub │ ├── IAM Access Analyzer → Security Hub │ ├── Config → Security Hub │ └── Macie → Security Hub └── EventBridge → SNS/Lambda (alerts and automation) ``` ### Delegated Administrator Setup ```python # Python - Security Hub Organization Setup import boto3 def setup_security_hub_organization(): # Enable delegated administrator from management account org_client = boto3.client('organizations') # Register audit account as delegated administrator org_client.register_delegated_administrator( AccountId='AUDIT_ACCOUNT_ID', ServicePrincipal='securityhub.amazonaws.com' ) # Configure from audit account sh_client = boto3.client('securityhub', region_name='us-east-1') # Enable Security Hub sh_client.enable_security_hub( EnableDefaultStandards=False, ControlFindingGenerator='SECURITY_CONTROL' ) # Configure organization settings sh_client.update_organization_configuration( AutoEnable=True, AutoEnableStandards='DEFAULT' ) # Enable standards standards_to_enable = [ 'arn:aws:securityhub:us-east-1::standards/aws-foundational-security-best-practices/v/1.0.0', 'arn:aws:securityhub:us-east-1::standards/cis-aws-foundations-benchmark/v/1.4.0', ] sh_client.batch_enable_standards( StandardsSubscriptionRequests=[ {'StandardsArn': arn} for arn in standards_to_enable ] ) ``` ### Finding Automation with EventBridge ```yaml # EventBridge Rule - Critical Finding Alert Resources: CriticalFindingRule: Type: AWS::Events::Rule Properties: Name: security-hub-critical-findings EventPattern: source: - aws.securityhub detail-type: - Security Hub Findings - Imported detail: findings: Severity: Label: - CRITICAL - HIGH Workflow: Status: - NEW RecordState: - ACTIVE Targets: - Arn: !Ref SecurityAlertTopic Id: SNSAlert InputTransformer: InputPathsMap: account: "$.detail.findings[0].AwsAccountId" title: "$.detail.findings[0].Title" severity: "$.detail.findings[0].Severity.Label" region: "$.detail.findings[0].Region" resource: "$.detail.findings[0].Resources[0].Id" InputTemplate: | "Security Hub Alert Account: <account> Region: <region> Severity: <severity> Title: <title> Resource: <resource>" - Arn: !GetAtt RemediationLambda.Arn Id: AutoRemediation # Lambda for auto-remediation RemediationLambda: Type: AWS::Lambda::Function Properties: FunctionName: security-hub-auto-remediation Runtime: python3.12 Handler: index.handler Role: !GetAtt RemediationLambdaRole.Arn Code: ZipFile: | import boto3 import json def handler(event, context): findings = event['detail']['findings'] for finding in findings: account_id = finding['AwsAccountId'] region = finding['Region'] generator_id = finding['GeneratorId'] resources = finding['Resources'] # Route remediation based on finding type if 'S3.1' in generator_id: remediate_s3_public_access(resources, account_id, region) elif 'IAM.1' in generator_id: remediate_iam_policy(resources, account_id, region) # Update finding workflow status sh_client = boto3.client('securityhub', region_name=region) sh_client.batch_update_findings( FindingIdentifiers=[{ 'Id': finding['Id'], 'ProductArn': finding['ProductArn'] }], Workflow={'Status': 'IN_PROGRESS'}, Note={ 'Text': 'Auto-remediation initiated', 'UpdatedBy': 'AutoRemediationLambda' } ) def remediate_s3_public_access(resources, account_id, region): sts = boto3.client('sts') assumed_role = sts.assume_role( RoleArn=f'arn:aws:iam::{account_id}:role/RemediationRole', RoleSessionName='RemediationSession' ) credentials = assumed_role['Credentials'] s3_client = boto3.client( 's3', aws_access_key_id=credentials['AccessKeyId'], aws_secret_access_key=credentials['SecretAccessKey'], aws_session_token=credentials['SessionToken'] ) for resource in resources: if resource['Type'] == 'AwsS3Bucket': bucket_name = resource['Id'].split(':::')[1] s3_client.put_public_access_block( Bucket=bucket_name, PublicAccessBlockConfiguration={ 'BlockPublicAcls': True, 'IgnorePublicAcls': True, 'BlockPublicPolicy': True, 'RestrictPublicBuckets': True } ) ``` --- ## 4. Amazon GuardDuty ### Configuration Pattern ``` GuardDuty (Audit Account - Delegated Administrator) ├── Organization-wide auto-enable ├── Protection Plans: │ ├── S3 Protection (enabled) │ ├── EKS Protection (enabled) │ ├── Malware Protection (enabled) │ ├── RDS Protection (enabled) │ ├── Lambda Protection (enabled) │ └── Runtime Monitoring (enabled) ├── Threat Intelligence: │ ├── Custom Threat Lists │ └── Trusted IP Lists └── Findings → Security Hub → EventBridge ``` ### Organization Setup ```python # GuardDuty Organization Setup import boto3 def setup_guardduty_organization(): gd_client = boto3.client('guardduty') # Create detector in audit account detector_response = gd_client.create_detector( Enable=True, FindingPublishingFrequency='FIFTEEN_MINUTES', DataSources={ 'S3Logs': {'Enable': True}, 'Kubernetes': { 'AuditLogs': {'Enable': True} }, 'MalwareProtection': { 'ScanEc2InstanceWithFindings': { 'EbsVolumes': True } } }, Features=[ {'Name': 'S3_DATA_EVENTS', 'Status': 'ENABLED'}, {'Name': 'EKS_AUDIT_LOGS', 'Status': 'ENABLED'}, {'Name': 'EBS_MALWARE_PROTECTION', 'Status': 'ENABLED'}, {'Name': 'RDS_LOGIN_EVENTS', 'Status': 'ENABLED'}, {'Name': 'EKS_RUNTIME_MONITORING', 'Status': 'ENABLED', 'AdditionalConfiguration': [ {'Name': 'EKS_ADDON_MANAGEMENT', 'Status': 'ENABLED'} ]}, {'Name': 'LAMBDA_NETWORK_LOGS', 'Status': 'ENABLED'}, {'Name': 'RUNTIME_MONITORING', 'Status': 'ENABLED', 'AdditionalConfiguration': [ {'Name': 'EC2_AGENT_MANAGEMENT', 'Status': 'ENABLED'}, {'Name': 'ECS_FARGATE_AGENT_MANAGEMENT', 'Status': 'ENABLED'} ]}, ] ) detector_id = detector_response['DetectorId'] # Configure organization auto-enable gd_client.update_organization_configuration( DetectorId=detector_id, AutoEnable=False, # Deprecated, use AutoEnableOrganizationMembers AutoEnableOrganizationMembers='ALL', DataSources={ 'S3Logs': {'AutoEnable': True}, 'Kubernetes': { 'AuditLogs': {'AutoEnable': True} }, 'MalwareProtection': { 'ScanEc2InstanceWithFindings': { 'EbsVolumes': {'AutoEnable': True} } } }, Features=[ {'Name': 'S3_DATA_EVENTS', 'AutoEnable': 'NEW'}, {'Name': 'EKS_AUDIT_LOGS', 'AutoEnable': 'NEW'}, {'Name': 'EBS_MALWARE_PROTECTION', 'AutoEnable': 'NEW'}, {'Name': 'RUNTIME_MONITORING', 'AutoEnable': 'NEW', 'AdditionalConfiguration': [ {'Name': 'EC2_AGENT_MANAGEMENT', 'AutoEnable': 'NEW'}, {'Name': 'ECS_FARGATE_AGENT_MANAGEMENT', 'AutoEnable': 'NEW'} ]}, ] ) # Add trusted IP list gd_client.create_ip_set( DetectorId=detector_id, Name='TrustedIPs', Format='TXT', Location=f's3://security-config-bucket/trusted-ips.txt', Activate=True ) # Add threat intelligence list gd_client.create_threat_intel_set( DetectorId=detector_id, Name='CustomThreatIntel', Format='TXT', Location=f's3://security-config-bucket/threat-intel.txt', Activate=True ) ``` ### Finding-Based Automated Response ```python # Lambda - GuardDuty Finding Response import boto3 import json def handler(event, context): finding = event['detail'] finding_type = finding['type'] severity = finding['severity'] account_id = finding['accountId'] region = finding['region'] print(f"GuardDuty Finding: {finding_type}, Severity: {severity}") # High severity response if severity >= 7.0: if 'UnauthorizedAccess:IAMUser' in finding_type: handle_unauthorized_iam_access(finding, account_id, region) elif 'CryptoCurrency:EC2' in finding_type: handle_crypto_mining(finding, account_id, region) elif 'Trojan:EC2' in finding_type: handle_malware(finding, account_id, region) elif 'Backdoor:EC2' in finding_type: handle_backdoor(finding, account_id, region) # All findings go to Security Hub (automatic integration) # Send notifications send_notification(finding, severity) def handle_unauthorized_iam_access(finding, account_id, region): """Disable compromised IAM user""" sts = boto3.client('sts') assumed_role = sts.assume_role( RoleArn=f'arn:aws:iam::{account_id}:role/SecurityResponseRole', RoleSessionName='GuardDutyResponse' ) credentials = assumed_role['Credentials'] iam_client = boto3.client('iam', **get_credentials(credentials)) # Get affected user resource = finding['resource'] if resource['resourceType'] == 'AccessKey': username = resource['accessKeyDetails']['userName'] access_key_id = resource['accessKeyDetails']['accessKeyId'] # Disable access key iam_client.update_access_key( UserName=username, AccessKeyId=access_key_id, Status='Inactive' ) # Attach deny all policy iam_client.put_user_policy( UserName=username, PolicyName='EmergencyDenyAll', PolicyDocument=json.dumps({ "Version": "2012-10-17", "Statement": [{ "Effect": "Deny", "Action": "*", "Resource": "*" }] }) ) def handle_crypto_mining(finding, account_id, region): """Isolate suspected instance""" sts = boto3.client('sts') assumed_role = sts.assume_role( RoleArn=f'arn:aws:iam::{account_id}:role/SecurityResponseRole', RoleSessionName='GuardDutyResponse' ) credentials = assumed_role['Credentials'] ec2_client = boto3.client('ec2', region_name=region, **get_credentials(credentials)) instance_id = finding['resource']['instanceDetails']['instanceId'] # Create isolation security group vpc_id = finding['resource']['instanceDetails']['networkInterfaces'][0]['vpcId'] sg_response = ec2_client.create_security_group( GroupName=f'isolation-{instance_id}', Description='Isolation security group - GuardDuty response', VpcId=vpc_id ) isolation_sg_id = sg_response['GroupId'] # No inbound/outbound rules (deny all) # Apply isolation security group to instance ec2_client.modify_instance_attribute( InstanceId=instance_id, Groups=[isolation_sg_id] ) # Create snapshot for forensic analysis for interface in finding['resource']['instanceDetails']['networkInterfaces']: pass # Snapshot all attached volumes instance_info = ec2_client.describe_instances(InstanceIds=[instance_id]) for reservation in instance_info['Reservations']: for instance in reservation['Instances']: for volume in instance.get('BlockDeviceMappings', []): volume_id = volume['Ebs']['VolumeId'] ec2_client.create_snapshot( VolumeId=volume_id, Description=f'Forensic snapshot - GuardDuty finding - {instance_id}' ) def get_credentials(credentials): return { 'aws_access_key_id': credentials['AccessKeyId'], 'aws_secret_access_key': credentials['SecretAccessKey'], 'aws_session_token': credentials['SessionToken'] } def send_notification(finding, severity): sns = boto3.client('sns') sns.publish( TopicArn='arn:aws:sns:us-east-1:AUDIT_ACCOUNT:security-alerts', Subject=f'GuardDuty Alert - Severity {severity}', Message=json.dumps(finding, indent=2) ) ``` --- ## 5. IAM Access Analyzer ### Configuration Pattern ``` IAM Access Analyzer (Audit Account - Delegated Administrator) ├── Organization-wide analyzer │ ├── External Access Analyzer │ └── Unused Access Analyzer ├── Findings: │ ├── Cross-account access │ ├── Public access │ ├── Unused IAM roles │ ├── Unused access keys │ └── Unused passwords └── Archive Rules: Auto-archive expected patterns ``` ### Setup ```python # IAM Access Analyzer Organization Setup import boto3 def setup_access_analyzer(): analyzer_client = boto3.client('accessanalyzer') # Create organization-wide external access analyzer external_analyzer = analyzer_client.create_analyzer( analyzerName='organization-external-access', type='ORGANIZATION', tags={ 'Purpose': 'External Access Analysis', 'ManagedBy': 'SecurityTeam' } ) # Create unused access analyzer unused_analyzer = analyzer_client.create_analyzer( analyzerName='organization-unused-access', type='ORGANIZATION_UNUSED_ACCESS', configuration={ 'unusedAccess': { 'unusedAccessAge': 90 # Flag access unused for 90+ days } } ) # Configure archive rules (expected patterns) analyzer_arn = external_analyzer['arn'] # Archive rule: Allow cross-account access for specific services analyzer_client.create_archive_rule( analyzerName='organization-external-access', ruleName='allow-monitoring-cross-account', filter={ 'principal.AWS': { 'contains': ['arn:aws:iam::MONITORING_ACCOUNT:role/MonitoringRole'] }, 'isPublic': { 'eq': ['false'] } } ) # Archive rule: Allow known third-party access analyzer_client.create_archive_rule( analyzerName='organization-external-access', ruleName='allow-trusted-third-party', filter={ 'principal.AWS': { 'contains': ['arn:aws:iam::THIRD_PARTY_ACCOUNT:root'] }, 'condition.aws:PrincipalOrgID': { 'exists': True } } ) def review_unused_access_findings(): """Review and process unused access findings""" analyzer_client = boto3.client('accessanalyzer') # List unused access findings paginator = analyzer_client.get_paginator('list_findings_v2') findings = [] for page in paginator.paginate( analyzerArn='arn:aws:accessanalyzer:us-east-1:AUDIT_ACCOUNT:analyzer/organization-unused-access', filter={ 'findingType': { 'eq': ['UnusedIAMRole', 'UnusedIAMUserAccessKey', 'UnusedPermission'] }, 'status': { 'eq': ['ACTIVE'] } } ): findings.extend(page['findings']) # Generate cleanup report report = { 'unused_roles': [], 'unused_access_keys': [], 'unused_permissions': [] } for finding in findings: finding_type = finding['findingType'] resource = finding['resource'] last_accessed = finding.get('findingDetails', {}).get('unusedIamRoleDetails', {}).get('lastAccessed') if finding_type == 'UnusedIAMRole': report['unused_roles'].append({ 'resource': resource, 'last_accessed': last_accessed, 'account': finding['resourceOwnerAccount'] }) elif finding_type == 'UnusedIAMUserAccessKey': report['unused_access_keys'].append({ 'resource': resource, 'account': finding['resourceOwnerAccount'] }) return report ``` --- ## 6. Amazon Detective ### Configuration Pattern ``` Amazon Detective (Audit Account - Delegated Administrator) ├── Organization-wide behavior graph ├── Data Sources: │ ├── CloudTrail (automatic) │ ├── VPC Flow Logs (automatic) │ ├── GuardDuty findings (automatic) │ ├── EKS audit logs (if enabled) │ └── WAF logs (if enabled) ├── Investigation Workflow: │ ├── GuardDuty finding → Detective investigation │ ├── Security Hub finding → Detective pivot │ └── Manual investigation └── Member account data retention: 1 year ``` ### Setup ```python # Amazon Detective Organization Setup import boto3 def setup_detective(): detective_client = boto3.client('detective') # Create behavior graph (in audit account) graph_response = detective_client.create_graph( Tags={ 'Purpose': 'Security Investigation', 'ManagedBy': 'SecurityTeam' } ) graph_arn = graph_response['GraphArn'] # Enable organization integration detective_client.enable_organization_admin_account( AccountId='AUDIT_ACCOUNT_ID' ) # Update organization configuration detective_client.update_organization_configuration( GraphArn=graph_arn, AutoEnable=True ) # Add all organization members org_client = boto3.client('organizations') accounts = [] paginator = org_client.get_paginator('list_accounts') for page in paginator.paginate(): for account in page['Accounts']: if account['Status'] == 'ACTIVE' and account['Id'] != 'MANAGEMENT_ACCOUNT_ID': accounts.append({ 'AccountId': account['Id'], 'EmailAddress': account['Email'] }) # Add members in batches of 50 batch_size = 50 for i in range(0, len(accounts), batch_size): batch = accounts[i:i+batch_size] detective_client.create_members( GraphArn=graph_arn, Accounts=batch ) return graph_arn def investigate_guardduty_finding(finding_id, graph_arn): """Start investigation from GuardDuty finding""" detective_client = boto3.client('detective') # Create investigation investigation = detective_client.create_investigation( GraphArn=graph_arn, EntityArn=f'arn:aws:iam::ACCOUNT_ID:role/SuspiciousRole', ScopeStartTime='2024-01-01T00:00:00Z', ScopeEndTime='2024-01-31T23:59:59Z' ) investigation_id = investigation['InvestigationId'] # Get investigation findings indicators = detective_client.list_investigation_details( GraphArn=graph_arn, InvestigationId=investigation_id ) return { 'investigation_id': investigation_id, 'indicators': indicators } ``` --- ## 7. Amazon Inspector ### Configuration Pattern ``` Amazon Inspector (Audit Account - Delegated Administrator) ├── Organization-wide auto-enable ├── Scan Types: │ ├── EC2 Instance Scanning │ │ └── SSM Agent required │ ├── ECR Container Image Scanning │ │ └── On push + continuous scanning │ ├── Lambda Function Scanning │ │ └── Code + dependencies │ └── Lambda Code Scanning (CodeGuru) ├── Findings → Security Hub (automatic) └── Suppression Rules: Reduce false positives ``` ### Setup ```python # Amazon Inspector Organization Setup import boto3 def setup_inspector(): inspector_client = boto3.client('inspector2') # Enable delegated administrator inspector_client.enable_delegated_admin_account( delegatedAdminAccountId='AUDIT_ACCOUNT_ID' ) # Enable Inspector for organization inspector_client.enable( resourceTypes=[ 'ECR', 'EC2', 'LAMBDA', 'LAMBDA_CODE' ] ) # Configure auto-enable for new accounts inspector_client.update_organization_configuration( autoEnable={ 'ec2': True, 'ecr': True, 'lambda': True, 'lambdaCode': True } ) # Configure ECR scan frequency ecr_client = boto3.client('ecr') ecr_client.put_registry_scanning_configuration( scanType='ENHANCED', rules=[ { 'scanFrequency': 'CONTINUOUS_SCAN', 'repositoryFilters': [ { 'filter': '*', 'filterType': 'WILDCARD' } ] } ] ) def create_suppression_rules(): """Create suppression rules for known false positives""" inspector_client = boto3.client('inspector2') # Suppress specific CVE for dev accounts inspector_client.create_filter( name='suppress-dev-low-severity', description='Suppress low severity findings in dev accounts', action='SUPPRESS', filterCriteria={ 'awsAccountId': [ {'comparison': 'EQUALS', 'value': 'DEV_ACCOUNT_ID'} ], 'severity': [ {'comparison': 'EQUALS', 'value': 'LOW'}, {'comparison': 'EQUALS', 'value': 'INFORMATIONAL'} ] } ) # Suppress findings for specific base image inspector_client.create_filter( name='suppress-base-image-cve', description='Suppress known base image CVEs pending vendor fix', action='SUPPRESS', filterCriteria={ 'vulnerabilityId': [ {'comparison': 'EQUALS', 'value': 'CVE-2023-XXXXX'} ], 'resourceType': [ {'comparison': 'EQUALS', 'value': 'AWS_ECR_CONTAINER_IMAGE'} ] } ) def get_vulnerability_report(): """Generate vulnerability report across organization""" inspector_client = boto3.client('inspector2') # Get critical and high findings paginator = inspector_client.get_paginator('list_findings') critical_findings = [] for page in paginator.paginate( filterCriteria={ 'severity': [ {'comparison': 'EQUALS', 'value': 'CRITICAL'}, {'comparison': 'EQUALS', 'value': 'HIGH'} ], 'findingStatus': [ {'comparison': 'EQUALS', 'value': 'ACTIVE'} ] }, sortCriteria={ 'field': 'SEVERITY', 'sortOrder': 'DESC' } ): critical_findings.extend(page['findings']) # Aggregate by account and resource type by_account = {} for finding in critical_findings: account = finding['awsAccountId'] if account not in by_account: by_account[account] = {'CRITICAL': 0, 'HIGH': 0} severity = finding['severity'] by_account[account][severity] = by_account[account].get(severity, 0) + 1 return by_account ``` --- ## 8. Integrated Architecture ### Service Integration Flow ``` ┌─────────────────────────────────────────────────────────────┐ │ Audit Account │ │ │ │ ┌──────────┐ ┌──────────────┐ ┌──────────────────┐ │ │ │GuardDuty │───▶│ Security Hub │◀───│ Inspector │ │ │ └──────────┘ └──────┬───────┘ └──────────────────┘ │ │ │ │ │ ┌──────────┐ │ ┌──────────────────┐ │ │ │ Config │───────────┤ │ Access Analyzer │ │ │ └──────────┘ │ └────────┬─────────┘ │ │ │ │ │ │ ┌──────────┐ ▼ │ │ │ │Detective │ ┌──────────────┐ │ │ │ └──────────┘ │ EventBridge │◀────────────┘ │ │ ▲ └──────┬───────┘ │ │ │ │ │ │ │ ┌───────────┼───────────┐ │ │ │ ▼ ▼ ▼ │ │ │ ┌────┐ ┌──────┐ ┌────────┐ │ │ │ │SNS │ │Lambda│ │Step │ │ │ │ │ │ │ │ │Functions│ │ │ │ └────┘ └──────┘ └────────┘ │ │ │ │ │ │ │ └──────────────┘ │ │ │ Investigation │ Orchestrated │ │ ▼ Remediation │ └─────────────────────────────────────────────────────────────┘ ▲ Findings/Events from all member accounts ``` ### Unified Security Dashboard ```python # Lambda - Unified Security Metrics Collection import boto3 import json from datetime import datetime, timedelta def collect_security_metrics(): metrics = {} # Security Hub - FSBP score by account sh_client = boto3.client('securityhub') scores = sh_client.list_standards_control_associations( StandardsArn='arn:aws:securityhub:::standards/aws-foundational-security-best-practices/v/1.0.0' ) # GuardDuty - Active findings count gd_client = boto3.client('guardduty') detectors = gd_client.list_detectors() for detector_id in detectors['DetectorIds']: findings = gd_client.list_findings( DetectorId=detector_id, FindingCriteria={ 'Criterion': { 'service.archived': {'Eq': ['false']}, 'severity': {'Gte': 4} } } ) metrics['guardduty_active_findings'] = len(findings['FindingIds']) # Inspector - Critical vulnerabilities count inspector_client = boto3.client('inspector2') counts = inspector_client.get_counts( filterCriteria={ 'severity': [{'comparison': 'EQUALS', 'value': 'CRITICAL'}], 'findingStatus': [{'comparison': 'EQUALS', 'value': 'ACTIVE'}] } ) metrics['inspector_critical_vulns'] = counts['counts'][0]['count'] # Config - Non-compliant resources count config_client = boto3.client('config') compliance = config_client.describe_compliance_by_config_rule( ComplianceTypes=['NON_COMPLIANT'] ) metrics['config_non_compliant_rules'] = len(compliance['ComplianceByConfigRules']) # Publish to CloudWatch cw_client = boto3.client('cloudwatch') cw_client.put_metric_data( Namespace='SecurityPosture', MetricData=[ { 'MetricName': 'GuardDutyActiveFindings', 'Value': metrics.get('guardduty_active_findings', 0), 'Unit': 'Count', 'Timestamp': datetime.utcnow() }, { 'MetricName': 'InspectorCriticalVulnerabilities', 'Value': metrics.get('inspector_critical_vulns', 0), 'Unit': 'Count', 'Timestamp': datetime.utcnow() }, { 'MetricName': 'ConfigNonCompliantRules', 'Value': metrics.get('config_non_compliant_rules', 0), 'Unit': 'Count', 'Timestamp': datetime.utcnow() } ] ) return metrics ``` --- ## 9. Control Tower Integration ### Control Tower Customizations (CfCT) ```yaml # manifest.yaml - Security Service Deployment --- region: us-east-1 version: 2021-03-15 resources: # Deploy security baseline to all accounts - name: security-baseline resource_file: templates/security-baseline.yaml deploy_method: stack_set deployment_targets: organizational_units: - Security - Workloads regions: - us-east-1 - us-west-2 - eu-west-1 parameters: - parameter_key: AuditAccountId parameter_value: $[alfred_ssm_/org/audit-account-id] - parameter_key: LogArchiveAccountId parameter_value: $[alfred_ssm_/org/log-archive-account-id] # GuardDuty member configuration - name: guardduty-member-config resource_file: templates/guardduty-member.yaml deploy_method: stack_set deployment_targets: organizational_units: - Workloads/Production - Workloads/Staging regions: - us-east-1 # Security notification configuration - name: security-notifications resource_file: templates/security-notifications.yaml deploy_method: stack_set deployment_targets: accounts: - AUDIT_ACCOUNT_ID regions: - us-east-1 ``` ### Security Baseline Template ```yaml # templates/security-baseline.yaml AWSTemplateFormatVersion: '2010-09-09' Description: Security Baseline for Member Accounts Parameters: AuditAccountId: Type: String LogArchiveAccountId: Type: String Resources: # IAM Password Policy PasswordPolicy: Type: AWS::IAM::AccountPasswordPolicy Properties: MinimumPasswordLength: 14 RequireUppercaseCharacters: true RequireLowercaseCharacters: true RequireNumbers: true RequireSymbols: true MaxPasswordAge: 90 PasswordReusePrevention: 24 HardExpiry: false AllowUsersToChangePassword: true # Security Hub - Enable standards SecurityHubFSBP: Type: AWS::SecurityHub::Standard Properties: StandardsArn: !Sub 'arn:aws:securityhub:${AWS::Region}::standards/aws-foundational-security-best-practices/v/1.0.0' # Default EBS encryption EBSEncryptionDefault: Type: AWS::EC2::EncryptionByDefault Properties: Enabled: true # S3 Account Public Access Block S3PublicAccessBlock: Type: AWS::S3::AccountPublicAccessBlock Properties: BlockPublicAcls: true IgnorePublicAcls: true BlockPublicPolicy: true RestrictPublicBuckets: true # IAM Access Analyzer AccessAnalyzer: Type: AWS::AccessAnalyzer::Analyzer Properties: AnalyzerName: account-analyzer Type: ACCOUNT Tags: - Key: ManagedBy Value: ControlTower # Remediation role for audit account SecurityRemediationRole: Type: AWS::IAM::Role Properties: RoleName: SecurityRemediationRole AssumeRolePolicyDocument: Version: '2012-10-17' Statement: - Effect: Allow Principal: AWS: !Sub 'arn:aws:iam::${AuditAccountId}:root' Action: sts:AssumeRole Condition: StringEquals: sts:ExternalId: SecurityRemediation ManagedPolicyArns: - arn:aws:iam::aws:policy/ReadOnlyAccess Policies: - PolicyName: RemediationPolicy PolicyDocument: Version: '2012-10-17' Statement: - Effect: Allow Action: - s3:PutBucketPublicAccessBlock - s3:PutEncryptionConfiguration - ec2:ModifyInstanceAttribute - ec2:CreateSecurityGroup - iam:UpdateAccessKey - iam:PutUserPolicy - guardduty:CreateSampleFindings Resource: '*' # CloudWatch Alarms for root account usage RootAccountUsageAlarm: Type: AWS::CloudWatch::Alarm Properties: AlarmName: RootAccountUsage AlarmDescription: Alert on root account usage MetricName: RootAccountUsageCount Namespace: CloudTrailMetrics Statistic: Sum Period: 300 EvaluationPeriods: 1 Threshold: 1 ComparisonOperator: GreaterThanOrEqualToThreshold AlarmActions: - !Sub 'arn:aws:sns:${AWS::Region}:${AuditAccountId}:security-alerts' TreatMissingData: notBreaching # Root account metric filter RootAccountMetricFilter: Type: AWS::Logs::MetricFilter Properties: LogGroupName: /aws/cloudtrail/organization-trail FilterPattern: >- {$.userIdentity.type="Root" && $.userIdentity.invokedBy NOT EXISTS && $.eventType != "AwsServiceEvent"} MetricTransformations: - MetricName: RootAccountUsageCount MetricNamespace: CloudTrailMetrics MetricValue: '1' ``` --- ## 10. Compliance and Reporting ### Weekly Security Report Generator ```python # Lambda - Weekly Security Report import boto3 import json from datetime import datetime, timedelta from collections import defaultdict def generate_weekly_report(): end_time = datetime.utcnow() start_time = end_time - timedelta(days=7) report = { 'period': { 'start': start_time.isoformat(), 'end': end_time.isoformat() }, 'executive_summary': {}, 'guardduty_summary': {}, 'security_hub_summary': {}, 'inspector_summary': {}, 'config_compliance': {}, 'access_analyzer_summary': {} } # GuardDuty statistics gd_client = boto3.client('guardduty') detectors = gd_client.list_detectors() finding_stats = defaultdict(int) for detector_id in detectors['DetectorIds']: stats = gd_client.get_findings_statistics( DetectorId=detector_id, FindingStatisticTypes=['COUNT_BY_SEVERITY'], FindingCriteria={ 'Criterion': { 'updatedAt': { 'Gte': int(start_time.timestamp() * 1000), 'Lte': int(end_time.timestamp() * 1000) }, 'service.archived': {'Eq': ['false']} } } ) for severity, count in stats['FindingStatistics']['CountBySeverity'].items(): finding_stats[severity] += count report['guardduty_summary'] = dict(finding_stats) # Security Hub statistics sh_client = boto3.client('securityhub') sh_stats = sh_client.get_insight_results( InsightArn='arn:aws:securityhub:::insight/securityhub/default/1' ) report['security_hub_summary'] = { 'top_findings': sh_stats['InsightResults']['ResultValues'][:10] } # Config compliance config_client = boto3.client('config') compliance_summary = config_client.get_compliance_summary_by_config_rule() report['config_compliance'] = { 'compliant_rules': compliance_summary['ComplianceSummary']['CompliantResourceCount']['CappedCount'], 'non_compliant_rules': compliance_summary['ComplianceSummary']['NonCompliantResourceCount']['CappedCount'] } # Executive summary total_critical = report['guardduty_summary'].get('8.0', 0) + report['guardduty_summary'].get('9.0', 0) report['executive_summary'] = { 'critical_findings_week': total_critical, 'config_compliance_rate': calculate_compliance_rate(report['config_compliance']), 'trend': 'stable' # Compare with previous week } # Send report ses_client = boto3.client('ses') ses_client.send_email( Source='security@company.com', Destination={ 'ToAddresses': ['ciso@company.com', 'secteam@company.com'] }, Message={ 'Subject': {'Data': f'Weekly Security Report - {end_time.strftime("%Y-%m-%d")}'}, 'Body': { 'Html': {'Data': generate_html_report(report)}, 'Text': {'Data': json.dumps(report, indent=2)} } } ) # Store report in S3 s3_client = boto3.client('s3') s3_client.put_object( Bucket='security-reports-bucket', Key=f'weekly-reports/{end_time.strftime("%Y/%m/%d")}/security-report.json', Body=json.dumps(report, indent=2), ContentType='application/json' ) return report def calculate_compliance_rate(compliance_data): total = compliance_data['compliant_rules'] + compliance_data['non_compliant_rules'] if total == 0: return 100.0 return round(compliance_data['compliant_rules'] / total * 100, 2) ``` --- ## Summary ### Service Responsibility Matrix | Service | Delegated Admin | Key Features | Integration | |---------|----------------|--------------|-------------| | **CloudTrail** | Management Account | Org-wide API logging | → S3, CloudWatch Logs | | **Config** | Management Account | Compliance rules, aggregator | → S3, Security Hub | | **Security Hub** | Audit Account | Central finding aggregation | ← All services | | **GuardDuty** | Audit Account | Threat detection, ML | → Security Hub | | **Access Analyzer** | Audit Account | Excess permission detection | → Security Hub | | **Detective** | Audit Account | Security investigation | ← GuardDuty | | **Inspector** | Audit Account | Vulnerability scanning | → Security Hub, ECR | ### Key Implementation Points 1. **Single Delegated Administrator**: Consolidate security services in the audit account 2. **Automated Remediation**: Use EventBridge + Lambda for auto-response 3. **Layered Defense**: Multiple services catching different threat types 4. **Minimize Alert Fatigue**: Use archive rules and suppression rules 5. **Centralized Logging**: Aggregate all logs to log archive account 6. **Regular Reviews**: Automate weekly/monthly security reports