Security Incident Response

Incident Response Lifecycle

Preparation → Detection → Analysis → Containment → Eradication → Recovery → Post-Incident

Security Monitoring with SIEM

from elasticsearch import Elasticsearch
import json
from datetime import datetime, timedelta

es = Elasticsearch([os.getenv('ELASTICSEARCH_URL')])

class SIEMAnalyzer:
    def detect_brute_force(self, timeframe_minutes: int = 5) -> list:
        query = {
            "query": {
                "bool": {
                    "filter": [
                        {"term": {"event_type": "auth_failure"}},
                        {"range": {"@timestamp": {"gte": f"now-{timeframe_minutes}m"}}},
                    ]
                }
            },
            "aggs": {
                "by_user": {
                    "terms": {"field": "username.keyword", "size": 100},
                    "aggs": {
                        "by_ip": {"terms": {"field": "source_ip.keyword", "size": 100}},
                        "failure_count": {"value_count": {"field": "_id"}},
                    }
                }
            }
        }
        
        result = es.search(index="auth-logs-*", body=query)
        incidents = []
        
        for bucket in result['aggregations']['by_user']['buckets']:
            if bucket['failure_count']['value'] >= 10:
                incidents.append({
                    'type': 'brute_force',
                    'username': bucket['key'],
                    'attempts': bucket['failure_count']['value'],
                    'source_ips': [b['key'] for b in bucket['by_ip']['buckets']],
                    'severity': 'high',
                })
        
        return incidents

    def detect_data_exfiltration(self) -> list:
        query = {
            "query": {
                "bool": {
                    "filter": [
                        {"range": {"@timestamp": {"gte": "now-1h"}}},
                        {"range": {"bytes_out": {"gte": 104857600}}},  # > 100MB
                    ]
                }
            }
        }
        
        result = es.search(index="network-logs-*", body=query, size=50)
        return [hit['_source'] for hit in result['hits']['hits']]

    def create_incident(self, detection: dict):
        incident = {
            "@timestamp": datetime.utcnow().isoformat(),
            "incident_id": f"INC-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}",
            "type": detection['type'],
            "severity": detection['severity'],
            "status": "open",
            "details": detection,
        }
        es.index(index="incidents", body=incident)
        self.alert_soc(incident)

    def alert_soc(self, incident: dict):
        import requests
        requests.post(os.getenv('PAGERDUTY_WEBHOOK'), json={
            "payload": {
                "summary": f"{incident['severity'].upper()}: {incident['type']}",
                "severity": incident['severity'],
                "source": "SIEM",
                "custom_details": incident,
            },
            "routing_key": os.getenv('PAGERDUTY_KEY'),
            "event_action": "trigger",
        })

Automated Containment

import boto3
from typing import Optional

class IncidentContainment:
    def __init__(self):
        self.ec2 = boto3.client('ec2')
        self.iam = boto3.client('iam')

    def isolate_instance(self, instance_id: str) -> bool:
        """Quarantine compromised EC2 instance."""
        # Create isolation security group
        sg_response = self.ec2.create_security_group(
            GroupName=f'quarantine-{instance_id}',
            Description=f'Isolation SG for incident response - {instance_id}',
            VpcId=self._get_instance_vpc(instance_id),
        )
        isolation_sg_id = sg_response['GroupId']
        
        # No inbound or outbound rules = fully isolated
        
        # Apply isolation SG
        self.ec2.modify_instance_attribute(
            InstanceId=instance_id,
            Groups=[isolation_sg_id],
        )
        
        # Tag for tracking
        self.ec2.create_tags(
            Resources=[instance_id],
            Tags=[
                {'Key': 'security:quarantined', 'Value': 'true'},
                {'Key': 'security:quarantine_time', 'Value': datetime.utcnow().isoformat()},
            ]
        )
        
        print(f"Instance {instance_id} isolated in quarantine SG {isolation_sg_id}")
        return True

    def disable_user(self, username: str) -> bool:
        """Disable IAM user access during incident."""
        # Deactivate access keys
        paginator = self.iam.get_paginator('list_access_keys')
        for page in paginator.paginate(UserName=username):
            for key in page['AccessKeyMetadata']:
                self.iam.update_access_key(
                    UserName=username,
                    AccessKeyId=key['AccessKeyId'],
                    Status='Inactive',
                )
        
        # Detach all policies
        for policy in self.iam.list_attached_user_policies(UserName=username)['AttachedPolicies']:
            self.iam.detach_user_policy(
                UserName=username,
                PolicyArn=policy['PolicyArn'],
            )
        
        # Add explicit deny policy
        deny_policy = json.dumps({
            "Version": "2012-10-17",
            "Statement": [{"Effect": "Deny", "Action": "*", "Resource": "*"}]
        })
        self.iam.put_user_policy(
            UserName=username,
            PolicyName='IncidentResponseDeny',
            PolicyDocument=deny_policy,
        )
        
        print(f"User {username} disabled during incident")
        return True

    def block_ip(self, ip_address: str, reason: str):
        """Add IP to AWS WAF block list."""
        waf = boto3.client('wafv2')
        waf.update_ip_set(
            Name='threat-block-list',
            Scope='REGIONAL',
            Id=os.getenv('WAF_IP_SET_ID'),
            LockToken=self._get_waf_lock_token(),
            Addresses=[f"{ip_address}/32"],
        )

Evidence Collection

#!/bin/bash
# incident_collect.sh - Collect forensic evidence

INCIDENT_ID=$1
OUTPUT_DIR="/forensics/${INCIDENT_ID}_$(date +%Y%m%d_%H%M%S)"
mkdir -p "$OUTPUT_DIR"

echo "[*] Collecting system state..."

# Running processes
ps auxf > "$OUTPUT_DIR/processes.txt"

# Network connections
netstat -tulpn > "$OUTPUT_DIR/network_connections.txt"
ss -tulpn >> "$OUTPUT_DIR/network_connections.txt"

# Recent auth events
last -F > "$OUTPUT_DIR/last_logins.txt"
lastb -F > "$OUTPUT_DIR/failed_logins.txt"

# Recent files modified
find /var /tmp /home -newer /proc/1 -type f 2>/dev/null > "$OUTPUT_DIR/recent_files.txt"

# Crontabs
for user in $(cut -d: -f1 /etc/passwd); do
  crontab -l -u "$user" 2>/dev/null >> "$OUTPUT_DIR/crontabs.txt"
done

# Memory dump (if volatility available)
# sudo avml "$OUTPUT_DIR/memory.lime"

# Hash all collected files
sha256sum "$OUTPUT_DIR"/* > "$OUTPUT_DIR/checksums.sha256"

echo "[*] Evidence collected in $OUTPUT_DIR"

Incident Response Playbook