Security Operations (SOC & SIEM) — Complete Beginner's Guide

# simple_siem.py — A minimal SIEM implementation
import json
import re
from collections import defaultdict
from datetime import datetime, timedelta

class SimpleSIEM:
    """A minimal SIEM engine for demonstration."""

    def __init__(self):
        self.events = []
        self.correlation_rules = []
        self.alerts = []

    def ingest_log(self, log_entry: dict):
        """Ingest a normalized log entry."""
        # Add timestamp if not present
        if 'timestamp' not in log_entry:
            log_entry['timestamp'] = datetime.now().isoformat()
        self.events.append(log_entry)

    def ingest_syslog(self, raw_log: str):
        """Parse and ingest a syslog-style log."""
        # Simple syslog parser
        pattern = r'(\w{3}\s+\d+\s+\d+:\d+:\d+)\s+(\S+)\s+(\S+)\[(?\d+)?\]:?\s*(.*)'
        match = re.match(pattern, raw_log)

        if match:
            self.ingest_log({
                'raw': raw_log,
                'source': 'syslog',
                'hostname': match.group(2),
                'process': match.group(3),
                'message': match.group(5)
            })

    def add_correlation_rule(self, name: str, condition: callable, severity: str = "MEDIUM"):
        """Add a correlation rule — condition is a function that takes events."""
        self.correlation_rules.append({
            'name': name,
            'condition': condition,
            'severity': severity
        })

    def run_correlation(self):
        """Run all correlation rules against recent events."""
        recent = [e for e in self.events
                  if datetime.fromisoformat(e['timestamp']) > datetime.now() - timedelta(hours=1)]

        for rule in self.correlation_rules:
            matches = rule['condition'](recent)
            if matches:
                self.alerts.append({
                    'rule': rule['name'],
                    'severity': rule['severity'],
                    'timestamp': datetime.now().isoformat(),
                    'matches': len(matches),
                    'details': matches[:3]  # First 3 matches
                })
        return self.alerts

    def get_dashboard(self) -> dict:
        """Generate a SOC dashboard summary."""
        total = len(self.events)
        alerts_last_hour = len([a for a in self.alerts
                               if datetime.fromisoformat(a['timestamp']) > datetime.now() - timedelta(hours=1)])

        return {
            "total_events": total,
            "events_last_hour": len([e for e in self.events
                                    if datetime.fromisoformat(e['timestamp']) > datetime.now() - timedelta(hours=1)]),
            "active_alerts": alerts_last_hour,
            "top_sources": dict(Counter(e.get('source', 'unknown') for e in self.events[-1000:]).most_common(5))
        }

from collections import Counter

# Example usage
siem = SimpleSIEM()

# Add correlation rules
def brute_force_detection(events):
    """Detect brute force attempts: 5+ failed logins from same IP in 1 minute."""
    ip_attempts = defaultdict(list)
    for e in events:
        if 'failed login' in e.get('message', '').lower():
            ip = e.get('source_ip', 'unknown')
            ip_attempts[ip].append(e['timestamp'])

    suspicious = []
    for ip, timestamps in ip_attempts.items():
        if len(timestamps) >= 5:
            suspicious.append({"ip": ip, "attempts": len(timestamps)})
    return suspicious

def port_scan_detection(events):
    """Detect port scans: connections to 10+ different ports from same IP."""
    ip_ports = defaultdict(set)
    for e in events:
        if 'connection' in e.get('message', '').lower():
            ip = e.get('source_ip', 'unknown')
            port = e.get('dest_port', 0)
            ip_ports[ip].add(port)

    suspicious = []
    for ip, ports in ip_ports.items():
        if len(ports) >= 10:
            suspicious.append({"ip": ip, "ports_scanned": len(ports)})
    return suspicious

siem.add_correlation_rule("Brute Force Detection", brute_force_detection, "HIGH")
siem.add_correlation_rule("Port Scan Detection", port_scan_detection, "MEDIUM")

# Simulate some logs
for i in range(10):
    siem.ingest_log({
        'source': 'ssh',
        'source_ip': '192.168.1.100',
        'message': f'Failed login attempt #{i} from 192.168.1.100'
    })

# Run correlation
alerts = siem.run_correlation()
print("=== SOC Dashboard ===")
dashboard = siem.get_dashboard()
for k, v in dashboard.items():
    print(f"{k}: {v}")

print("\n=== Active Alerts ===")
for alert in alerts:
    print(f"[{alert['severity']}] {alert['rule']} — {alert['matches']} matches")

# docker-compose.yml — Minimal ELK SIEM stack
version: '3.8'
services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.12.0
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
    ports:
      - "9200:9200"

  logstash:
    image: docker.elastic.co/logstash/logstash:8.12.0
    volumes:
      - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf
    ports:
      - "5000:5000"
    depends_on:
      - elasticsearch

  kibana:
    image: docker.elastic.co/kibana/kibana:8.12.0
    ports:
      - "5601:5601"
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    depends_on:
      - elasticsearch

# alert_triage.py — Alert triage decision framework

class AlertTriage:
    """Triage security alerts using a structured decision framework."""

    SEVERITY_MAP = {
        "CRITICAL": {"response_time": "5 minutes", "requires_escalation": True},
        "HIGH": {"response_time": "15 minutes", "requires_escalation": True},
        "MEDIUM": {"response_time": "1 hour", "requires_escalation": False},
        "LOW": {"response_time": "24 hours", "requires_escalation": False},
    }

    def __init__(self, alert: dict):
        self.alert = alert
        self.status = "new"
        self.notes = []

    def evaluate_severity(self) -> str:
        """Determine severity based on multiple factors."""
        score = 0

        # Factor 1: Is sensitive data involved?
        if self.alert.get('data_classification') in ['PII', 'PCI', 'PHI']:
            score += 30

        # Factor 2: Is a critical system affected?
        if self.alert.get('system_criticality') == 'critical':
            score += 25

        # Factor 3: Has lateral movement been detected?
        if self.alert.get('lateral_movement'):
            score += 20

        # Factor 4: Is there confirmed data exfiltration?
        if self.alert.get('exfiltration_detected'):
            score += 25

        # Map score to severity
        if score >= 80:
            return "CRITICAL"
        elif score >= 50:
            return "HIGH"
        elif score >= 20:
            return "MEDIUM"
        return "LOW"

    def triage_steps(self) -> list[str]:
        """Generate triage steps based on severity."""
        severity = self.evaluate_severity()
        config = self.SEVERITY_MAP[severity]

        steps = [
            f"1. Acknowledge alert within {config['response_time']}",
            "2. Determine if alert is a true positive or false positive:",
        ]

        if severity in ["CRITICAL", "HIGH"]:
            steps += [
                "   a. Check if any data was accessed/exfiltrated",
                "   b. Identify affected systems and users",
                "   c. Begin containment (isolate systems if needed)",
                "   d. Escalate to Tier 2/3 incident response team",
                f"3. Notify SOC Manager immediately",
                f"4. Create incident ticket with severity {severity}"
            ]
        else:
            steps += [
                "   a. Review alert details in SIEM",
                "   b. Check for related events in last 24 hours",
                "   c. If false positive: document and close",
                "   d. If true positive: investigate during business hours",
                f"3. Create ticket and assign to next available analyst"
            ]

        return steps

    def investigate(self, siem_data: dict) -> dict:
        """Perform initial investigation."""
        finding = {
            "alert_id": self.alert.get('id'),
            "severity": self.evaluate_severity(),
            "status": "investigating",
            "indicators": [],
            "verdict": "pending"
        }

        # Check if IP is in threat intel feeds
        ip = self.alert.get('source_ip')
        if ip and ip in siem_data.get('threat_intel', {}).get('malicious_ips', []):
            finding["indicators"].append(f"Source IP {ip} matches known threat intel")
            finding["verdict"] = "malicious"

        # Check for related events
        related = siem_data.get('related_events', [])
        if len(related) > 5:
            finding["indicators"].append(f"Multiple related events ({len(related)}): pattern suggests active attack")
            if finding["verdict"] != "malicious":
                finding["verdict"] = "suspicious"

        if not finding["indicators"]:
            finding["verdict"] = "false_positive"

        return finding

# Example triage
alert = {
    "id": "ALERT-2026-001",
    "source_ip": "185.220.101.1",
    "data_classification": "PII",
    "system_criticality": "critical",
    "lateral_movement": True,
    "exfiltration_detected": False
}

triage = AlertTriage(alert)
print(f"Severity: {triage.evaluate_severity()}")
print("Triage Steps:")
for step in triage.triage_steps():
    print(f"  {step}")

# soc_dashboard.py
# Simulate a SOC monitoring dashboard
from datetime import datetime, timedelta
import random
import time

class SOCDashboard:
    """Simulate a real-time SOC dashboard."""

    def __init__(self):
        self.metrics = {
            "events_per_second": 0,
            "active_alerts": 0,
            "open_incidents": 0,
            "false_positive_rate": 0.0,
            "avg_response_time": 0
        }
        self.recent_alerts = []

    def simulate_tick(self):
        """Generate simulated SOC metrics."""
        self.metrics["events_per_second"] = random.randint(500, 2000)
        self.metrics["active_alerts"] = random.randint(2, 15)
        self.metrics["open_incidents"] = max(0, self.metrics["active_alerts"] - random.randint(0, 3))
        self.metrics["false_positive_rate"] = round(random.uniform(0.15, 0.40), 2)
        self.metrics["avg_response_time"] = random.randint(5, 30)

        # Random alert generation
        if random.random() < 0.3:
            severity = random.choices(
                ["CRITICAL", "HIGH", "MEDIUM", "LOW"],
                weights=[0.05, 0.15, 0.40, 0.40]
            )[0]
            alert = {
                "timestamp": datetime.now().isoformat(),
                "severity": severity,
                "rule": random.choice([
                    "Brute Force Detected", "Malspam Detected",
                    "C2 Beaconing", "Data Exfiltration",
                    "Ransomware Indicator", "Phishing URL Clicked"
                ])
            }
            self.recent_alerts.append(alert)

    def display(self):
        """Print dashboard to console."""
        print("\033[2J\033[H")  # Clear screen (ANSI)
        print("=" * 60)
        print(f"SOC DASHBOARD — {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("=" * 60)
        print(f"Events/sec:    {self.metrics['events_per_second']}")
        print(f"Active Alerts: {self.metrics['active_alerts']}")
        print(f"Open Incidents: {self.metrics['open_incidents']}")
        print(f"FP Rate:       {self.metrics['false_positive_rate']:.0%}")
        print(f"Avg Response:  {self.metrics['avg_response_time']} min")
        print("-" * 60)
        print("Recent Alerts:")
        for alert in self.recent_alerts[-5:]:
            sev = alert['severity']
            print(f"  [{sev:<8}] {alert['rule']}")

# Run simulation
dashboard = SOCDashboard()
for _ in range(5):
    dashboard.simulate_tick()
    dashboard.display()
    time.sleep(0.5)