Monitoring, Security, and Incident Response¶

import boto3
from datetime import datetime

cloudwatch = boto3.client('cloudwatch')

# Put single metric
cloudwatch.put_metric_data(
    Namespace='MyApp',
    MetricData=[
        {
            'MetricName': 'OrdersProcessed',
            'Value': 42,
            'Unit': 'Count',
            'Timestamp': datetime.utcnow(),
            'Dimensions': [
                {
                    'Name': 'Environment',
                    'Value': 'Production'
                },
                {
                    'Name': 'Region',
                    'Value': 'us-east-1'
                }
            ]
        }
    ]
)

# Put multiple metrics
cloudwatch.put_metric_data(
    Namespace='MyApp',
    MetricData=[
        {
            'MetricName': 'ProcessingTime',
            'Value': 125.5,
            'Unit': 'Milliseconds',
            'StorageResolution': 1  # High-resolution (1-second)
        },
        {
            'MetricName': 'ErrorRate',
            'Value': 0.02,
            'Unit': 'Percent'
        }
    ]
)

# Metric with statistics
cloudwatch.put_metric_data(
    Namespace='MyApp',
    MetricData=[
        {
            'MetricName': 'RequestLatency',
            'StatisticValues': {
                'SampleCount': 1000,
                'Sum': 50000,
                'Minimum': 10,
                'Maximum': 500
            },
            'Unit': 'Milliseconds'
        }
    ]
)

import json
import boto3

cloudwatch = boto3.client('cloudwatch')

dashboard_body = {
    "widgets": [
        {
            "type": "metric",
            "properties": {
                "metrics": [
                    ["AWS/EC2", "CPUUtilization", {"stat": "Average"}],
                    [".", ".", {"stat": "Maximum"}]
                ],
                "period": 300,
                "stat": "Average",
                "region": "us-east-1",
                "title": "EC2 CPU Utilization",
                "yAxis": {
                    "left": {"min": 0, "max": 100}
                }
            }
        },
        {
            "type": "log",
            "properties": {
                "query": """
                    SOURCE '/aws/lambda/my-function'
                    | fields @timestamp, @message
                    | filter @message like /ERROR/
                    | sort @timestamp desc
                    | limit 50
                """,
                "region": "us-east-1",
                "title": "Recent Errors"
            }
        },
        {
            "type": "metric",
            "properties": {
                "metrics": [
                    ["AWS/ApplicationELB", "TargetResponseTime",
                     {"stat": "Average", "label": "Avg Response Time"}],
                    ["...", {"stat": "p99", "label": "P99 Response Time"}]
                ],
                "view": "timeSeries",
                "stacked": false,
                "region": "us-east-1",
                "title": "ALB Response Times"
            }
        }
    ]
}

cloudwatch.put_dashboard(
    DashboardName='Production-Overview',
    DashboardBody=json.dumps(dashboard_body)
)

# Python Flask application
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.ext.flask.middleware import XRayMiddleware
from flask import Flask

app = Flask(__name__)

# Configure X-Ray
xray_recorder.configure(
    service='my-api',
    context_missing='LOG_ERROR',
    plugins=('EC2Plugin', 'ECSPlugin'),
    sampling_rules='sampling-rules.json'
)

XRayMiddleware(app, xray_recorder)

@app.route('/api/orders')
@xray_recorder.capture('process_order')
def process_order():
    # Custom subsegment
    subsegment = xray_recorder.begin_subsegment('database_query')
    try:
        result = database.query('SELECT * FROM orders')
        subsegment.put_metadata('row_count', len(result))
        subsegment.put_annotation('query_type', 'select')
        return result
    except Exception as e:
        subsegment.add_exception(e)
        raise
    finally:
        xray_recorder.end_subsegment()

# Async operations
@xray_recorder.capture_async('async_operation')
async def fetch_user_data(user_id):
    async with aiohttp.ClientSession() as session:
        async with session.get(f'https://api.example.com/users/{user_id}') as response:
            return await response.json()

import boto3
import json
from botocore.exceptions import ClientError

secrets = boto3.client('secretsmanager')

def get_secret(secret_name):
    """Retrieve secret from Secrets Manager with caching"""
    try:
        response = secrets.get_secret_value(SecretId=secret_name)

        if 'SecretString' in response:
            return json.loads(response['SecretString'])
        else:
            return base64.b64decode(response['SecretBinary'])
    except ClientError as e:
        if e.response['Error']['Code'] == 'ResourceNotFoundException':
            print(f"Secret {secret_name} not found")
        raise e

def rotate_secret(secret_name):
    """Trigger secret rotation"""
    response = secrets.rotate_secret(
        SecretId=secret_name,
        RotationLambdaARN='arn:aws:lambda:region:account:function:rotate-secret',
        RotationRules={
            'AutomaticallyAfterDays': 30
        }
    )
    return response

# Usage
db_credentials = get_secret('prod/database/credentials')
database_url = f"postgresql://{db_credentials['username']}:{db_credentials['password']}@{db_credentials['host']}:5432/mydb"

import boto3

ssm = boto3.client('ssm')

# Store parameter
ssm.put_parameter(
    Name='/app/prod/db_password',
    Value='secure-password',
    Type='SecureString',
    KeyId='alias/aws/ssm',
    Tier='Standard',
    Tags=[
        {'Key': 'Environment', 'Value': 'Production'},
        {'Key': 'Application', 'Value': 'MyApp'}
    ]
)

# Get parameter
response = ssm.get_parameter(
    Name='/app/prod/db_password',
    WithDecryption=True
)
password = response['Parameter']['Value']

# Get parameters by path
response = ssm.get_parameters_by_path(
    Path='/app/prod/',
    Recursive=True,
    WithDecryption=True
)

config = {param['Name'].split('/')[-1]: param['Value']
          for param in response['Parameters']}

import boto3
import json

ec2 = boto3.client('ec2')
sns = boto3.client('sns')

def lambda_handler(event, context):
    """Respond to GuardDuty findings"""

    finding = event['detail']
    severity = finding['severity']
    finding_type = finding['type']

    if 'recon' in finding_type.lower():
        # Port scanning detected
        instance_id = finding['resource']['instanceDetails']['instanceId']

        # Isolate instance
        isolate_instance(instance_id)

        # Create snapshot for forensics
        create_forensic_snapshot(instance_id)

        # Notify security team
        notify_security_team(finding)

    elif 'trojan' in finding_type.lower() or 'backdoor' in finding_type.lower():
        # Malware detected
        instance_id = finding['resource']['instanceDetails']['instanceId']

        # Immediate isolation
        isolate_instance(instance_id)

        # Revoke IAM credentials
        instance_profile = get_instance_profile(instance_id)
        if instance_profile:
            disable_iam_credentials(instance_profile)

        # Escalate to critical
        escalate_incident(finding, 'CRITICAL')

def isolate_instance(instance_id):
    """Apply restrictive security group"""
    # Create forensics security group
    forensic_sg = ec2.create_security_group(
        GroupName=f'forensic-isolation-{instance_id}',
        Description='Isolation for incident response',
        VpcId=get_instance_vpc(instance_id)
    )

    # No inbound, only outbound to logging
    ec2.authorize_security_group_egress(
        GroupId=forensic_sg['GroupId'],
        IpPermissions=[
            {
                'IpProtocol': 'tcp',
                'FromPort': 443,
                'ToPort': 443,
                'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
            }
        ]
    )

    # Apply to instance
    ec2.modify_instance_attribute(
        InstanceId=instance_id,
        Groups=[forensic_sg['GroupId']]
    )

def create_forensic_snapshot(instance_id):
    """Create EBS snapshots for forensic analysis"""
    volumes = ec2.describe_volumes(
        Filters=[{'Name': 'attachment.instance-id', 'Values': [instance_id]}]
    )

    for volume in volumes['Volumes']:
        ec2.create_snapshot(
            VolumeId=volume['VolumeId'],
            Description=f'Forensic snapshot - GuardDuty finding - {instance_id}',
            TagSpecifications=[
                {
                    'ResourceType': 'snapshot',
                    'Tags': [
                        {'Key': 'Forensic', 'Value': 'true'},
                        {'Key': 'InstanceId', 'Value': instance_id}
                    ]
                }
            ]
        )

def notify_security_team(finding):
    """Send detailed notification"""
    message = {
        'severity': finding['severity'],
        'type': finding['type'],
        'description': finding['description'],
        'resource': finding['resource'],
        'time': finding['time']
    }

    sns.publish(
        TopicArn='arn:aws:sns:region:account:security-incidents',
        Subject=f"GuardDuty Finding: {finding['type']}",
        Message=json.dumps(message, indent=2)
    )