Incident Response & Forensics — Complete Guide for Beginners

# ir_detection.py — Simple log anomaly detector
import re
from collections import Counter
from datetime import datetime

class AnomalyDetector:
    """Detect common security anomalies in log data."""

    SUSPICIOUS_PATTERNS = {
        "multiple_auth_failures": r"Failed password.*ssh.*",
        "sql_injection_attempt": r"'.*OR.*'1'='1",
        "port_scan": r"Connection refused.*port",
        "suspicious_user_agent": r"(curl|wget|python-requests|masscan)",
        "large_file_download": r"GET.*\.(zip|tar|gz|sql|bak)\s+200\s+\d{7,}",
    }

    def __init__(self, log_file: str, threshold: int = 5):
        self.log_file = log_file
        self.threshold = threshold
        self.alerts = []

    def analyze(self) -> list[dict]:
        """Analyze a log file for anomalies."""
        try:
            with open(self.log_file, 'r') as f:
                lines = f.readlines()
        except FileNotFoundError:
            return [{"error": f"Log file {self.log_file} not found"}]

        for pattern_name, pattern in self.SUSPICIOUS_PATTERNS.items():
            matches = []
            for line in lines:
                if re.search(pattern, line, re.IGNORECASE):
                    matches.append(line.strip())

            if len(matches) >= self.threshold:
                self.alerts.append({
                    "alert": pattern_name,
                    "count": len(matches),
                    "severity": "HIGH" if len(matches) > 20 else "MEDIUM",
                    "timestamp": datetime.now().isoformat(),
                    "samples": matches[:3]  # First 3 matches
                })

        return self.alerts

# Example
# detector = AnomalyDetector("/var/log/auth.log", threshold=10)
# alerts = detector.analyze()
# for alert in alerts:
#     print(f"[{alert['severity']}] {alert['alert']} ({alert['count']} times)")

# Linux — Block an attacker IP with iptables
sudo iptables -A INPUT -s 192.168.1.100 -j DROP

# Windows — Disable a compromised user account
net user compromised_user /ACTIVE:NO

# Network — Capture live traffic before isolation
sudo tcpdump -i eth0 -w incident_capture.pcap

# Create a forensic image of a disk (Linux)
# Using dd with a hardware write-blocker in production
sudo dd if=/dev/sda of=/mnt/evidence/disk_image.dd bs=4k conv=noerror,sync

# Verify the image with SHA-256
sudo sha256sum /dev/sda
sudo sha256sum /mnt/evidence/disk_image.dd
# Both hashes must match to prove evidence integrity

# persistence_check.py — Check for common persistence mechanisms on Linux

def check_persistence() -> list[str]:
    """Check for common malware persistence mechanisms."""
    findings = []
    import os

    # Check cron jobs for all users
    cron_dirs = [
        "/etc/cron.hourly", "/etc/cron.daily",
        "/etc/cron.weekly", "/etc/cron.monthly",
        "/var/spool/cron/crontabs"
    ]
    for cron_dir in cron_dirs:
        if os.path.isdir(cron_dir):
            suspicious = [f for f in os.listdir(cron_dir)
                         if not f.startswith('.')]
            if suspicious:
                findings.append(f"Unexpected cron files in {cron_dir}: {suspicious}")

    # Check systemd services
    service_dirs = ["/etc/systemd/system", "/usr/lib/systemd/system"]
    for sd_dir in service_dirs:
        if os.path.isdir(sd_dir):
            services = os.listdir(sd_dir)
            # Flag recently added services
            for svc in services:
                path = os.path.join(sd_dir, svc)
                if os.path.getmtime(path) > 0:  # Check modification time
                    findings.append(f"Recently modified systemd service: {svc}")

    # Check SSH authorized_keys for unexpected keys
    ssh_dir = os.path.expanduser("~/.ssh/authorized_keys")
    if os.path.exists(ssh_dir):
        with open(ssh_dir) as f:
            keys = f.read().strip()
            if keys:
                findings.append(f"SSH authorized_keys found — verify each key")

    return findings

# Incident Post-Mortem

**Date**: 2026-06-07
**Incident ID**: IR-2026-042
**Severity**: Critical / High / Medium / Low

## Timeline
- **Detection**: 2026-06-07 14:30 UTC
- **Containment**: 14:52 UTC (22 minutes)
- **Eradication**: 16:10 UTC
- **Recovery**: 18:30 UTC
- **Total duration**: 4 hours

## Root Cause
A phishing email delivered a trojan that established C2 via HTTPS
to a malicious domain. The trojan dumped LSASS credentials.

## What Went Well
- Detection within 15 minutes via EDR alert
- Team activated within 10 minutes via on-call rotation
- Forensic images taken before remediation

## What Could Be Better
- Phishing awareness training needs quarterly refresher
- LSASS should be disabled on non-domain controllers
- C2 domain was not blocked at proxy because HTTPS inspection was disabled

## Action Items
- [ ] Enable HTTPS inspection for all outbound traffic (Owner: Network Team, Due: 1 week)
- [ ] Disable LSASS on all servers (Owner: Security Team, Due: 2 weeks)
- [ ] Conduct phishing simulation this month (Owner: HR, Due: 2 weeks)
- [ ] Update IR playbook for phishing/Trojan scenarios (Owner: IR Lead, Due: 1 week)

# chain_of_custody.py
# Track forensic evidence from collection to court
from datetime import datetime
import json

class ChainOfCustody:
    """Track evidence handling for legal admissibility."""

    def __init__(self, evidence_id: str, description: str):
        self.evidence_id = evidence_id
        self.description = description
        self.handlers = []
        self.add_entry("Evidence collected", "System")

    def add_entry(self, action: str, handler: str, notes: str = ""):
        """Record a custody transfer."""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "action": action,
            "handler": handler,
            "notes": notes
        }
        self.handlers.append(entry)

    def verify_integrity(self, original_hash: str, current_hash: str) -> bool:
        """Verify evidence hasn't been tampered with."""
        match = original_hash == current_hash
        self.add_entry(
            "Hash verification",
            "Forensic Analyst",
            f"Hashes {'match' if match else 'DO NOT MATCH'}"
        )
        return match

    def report(self) -> str:
        """Generate chain of custody report."""
        report = f"=== Chain of Custody: {self.evidence_id} ===\n"
        report += f"Description: {self.description}\n"
        report += f"Total transfers: {len(self.handlers)}\n\n"
        for entry in self.handlers:
            report += f"{entry['timestamp']} — {entry['action']} by {entry['handler']}\n"
            if entry['notes']:
                report += f"  Notes: {entry['notes']}\n"
        return report

# Example
evidence = ChainOfCustody(
    "EVD-2026-001",
    "Disk image — compromised web server #3"
)
evidence.add_entry("Imaged disk", "Analyst A", "SHA-256: a1b2c3...")
evidence.add_entry("Transferred to analysis VM", "Analyst A")
evidence.add_entry("Analyzed with Autopsy", "Analyst B")
evidence.add_entry("Returned to evidence locker", "Analyst C")
print(evidence.report())

# Check file type
file suspicious.exe

# Extract strings
strings suspicious.exe | head -50

# Check entropy (high entropy = packed/encrypted)
python3 -c "
import sys
with open(sys.argv[1], 'rb') as f:
    data = f.read()
    entropy = -sum((data.count(b)/len(data)) * (data.count(b)/len(data)).bit_length() for b in set(data))
    print(f'Entropy: {entropy:.2f}')
" suspicious.exe

# sandbox_monitor.py — Monitor process behavior in a sandbox VM
import subprocess
import json

class SandboxMonitor:
    """Monitor system calls and network connections in a sandbox."""

    def __init__(self, sample_path: str):
        self.sample_path = sample_path

    def capture_network(self, duration: int = 30):
        """Capture network traffic during execution."""
        print(f"Capturing network for {duration}s...")
        # In a real sandbox, you'd use tcpdump or wireshark CLI
        # For this demo, we simulate the output
        return {
            "dns_queries": ["evil-malware.com", "192.168.1.100"],
            "connections": [{"dest": "185.220.101.1", "port": 443}],
            "protocols": ["HTTPS", "DNS"]
        }

    def check_process_creation(self) -> list[dict]:
        """Monitor child processes created."""
        # Simulate detection
        return [
            {"pid": 1234, "name": "explorer.exe", "parent": 567, "cmd": "C:\\Windows\\System32\\cmd.exe /c powershell -enc ..."},
            {"pid": 1235, "name": "powershell.exe", "parent": 1234, "cmd": "powershell -enc ZgByAG8AbQA..."}
        ]

    def check_registry_changes(self) -> list[dict]:
        """Monitor registry modifications."""
        return [
            {"key": "HKLM\\Software\\Microsoft\\Windows\\CurrentVersion\\Run",
             "value": "MaliciousService",
             "action": "CREATE"}
        ]

    def generate_report(self) -> dict:
        """Generate a complete analysis report."""
        return {
            "sample": self.sample_path,
            "network": self.capture_network(),
            "processes": self.check_process_creation(),
            "registry": self.check_registry_changes(),
            "verdict": "MALICIOUS",
            "confidence": "HIGH"
        }

# ir_simulator.py
# Command-line incident response drill
import random
import time

class IRDrill:
    """Run a simulated incident response drill."""

    SCENARIOS = {
        "phishing": {
            "description": "User reports clicking a suspicious link and entering credentials",
            "steps": [
                "Identify affected user and machine",
                "Check email logs for phishing indicators",
                "Reset user credentials",
                "Check for other users who received the same email",
                "Scan affected machine for malware",
                "Block phishing domain at proxy"
            ]
        },
        "ransomware": {
            "description": "Files on file server being renamed with .locked extension",
            "steps": [
                "Isolate file server from network",
                "Identify patient zero (first infected machine)",
                "Take forensic image of affected server",
                "Check if decryption key exists",
                "Restore from clean backup",
                "Patch the vulnerability that allowed initial access"
            ]
        },
        "insider_threat": {
            "description": "Employee downloading large volumes of customer data before resignation",
            "steps": [
                "Disable user account immediately",
                "Escort employee from premises (if in-office)",
                "Take forensic image of workstation",
                "Check what data was accessed/downloaded",
                "Notify legal department",
                "Review access logs for all sensitive data"
            ]
        }
    }

    def run_drill(self, scenario_name: str = None):
        """Run a randomized IR drill."""
        if scenario_name:
            scenario = self.SCENARIOS.get(scenario_name)
            if not scenario:
                print(f"Unknown scenario: {scenario_name}")
                return
        else:
            scenario = random.choice(list(self.SCENARIOS.values()))

        print(f"\n=== INCIDENT RESPONSE DRILL ===")
        print(f"Scenario: {scenario['description']}\n")
        print("Correct steps (in order):")
        for i, step in enumerate(scenario['steps'], 1):
            print(f"  {i}. {step}")

        input("\nPress Enter to reveal the recommended order...")
        print("\n=== Recommended Response ===")
        for step in scenario['steps']:
            time.sleep(0.5)
            print(f"  ✓ {step}")

if __name__ == "__main__":
    drill = IRDrill()
    drill.run_drill("ransomware")