Lab 09 — SOC Alert Triage Automation

Mission Information

Difficulty: Advanced
Estimated Time: 180–240 minutes
Primary Language: Python
Security Domain: SOC / Detection Engineering / Incident Response / Security Automation
Environment: Local controlled lab
Automation Type: Defensive SOC Alert Triage

Mission

You are working as a SOC automation engineer.

Your SOC receives alerts from multiple security technologies:

SIEM

EDR / XDR

IDENTITY PLATFORM

FIREWALL

CLOUD SECURITY

VULNERABILITY MANAGEMENT

THREAT INTELLIGENCE

The challenge is not simply receiving alerts.

The challenge is determining:

WHICH ALERTS
REQUIRE ATTENTION FIRST?

Your mission is to build a Python automation pipeline that transforms raw alerts into enriched, prioritized analyst work items.

The final workflow will be:

RAW ALERTS
     ↓
VALIDATE
     ↓
NORMALIZE
     ↓
DEDUPLICATE
     ↓
ENRICH
 ┌───┼──────────────┐
 ↓   ↓      ↓       ↓
USER ASSET   IOC   VULNERABILITY
 ↓   ↓      ↓       ↓
 └───┼──────────────┘
     ↓
CORRELATE
     ↓
RISK SCORE
     ↓
TRIAGE RECOMMENDATION
     ↓
ANALYST QUEUE
     ↓
HUMAN REVIEW

Why This Lab Matters

A SOC can receive:

100

1,000

10,000+

ALERTS PER DAY

Treating every alert equally creates:

ALERT FATIGUE

SLOW INVESTIGATION

MISSED CRITICAL EVENTS

INCONSISTENT TRIAGE

ANALYST BURNOUT

Security automation helps analysts answer:

WHO IS INVOLVED?

WHAT ASSET IS INVOLVED?

HOW CRITICAL IS THE ASSET?

IS THE USER PRIVILEGED?

IS THE SOURCE KNOWN?

ARE THERE VULNERABILITIES?

IS THE ASSET INTERNET-FACING?

ARE THERE RELATED ALERTS?

WHAT SHOULD BE INVESTIGATED FIRST?

Important Principle

This lab automates:

COLLECTION

NORMALIZATION

ENRICHMENT

CORRELATION

PRIORITIZATION

REPORTING

It does not automatically:

DISABLE ACCOUNTS

ISOLATE ENDPOINTS

BLOCK IP ADDRESSES

DELETE CLOUD RESOURCES

RESET PASSWORDS

MODIFY FIREWALL RULES

High-impact response remains:

ANALYST / INCIDENT RESPONDER
          ↓
VALIDATION
          ↓
APPROVAL
          ↓
CONTROLLED RESPONSE

Learning Objectives

By completing this lab, you should be able to:

PROCESS SOC ALERTS

VALIDATE SECURITY EVENTS

NORMALIZE MULTIPLE ALERT TYPES

DEDUPLICATE ALERTS

ENRICH IDENTITY CONTEXT

ENRICH ASSET CONTEXT

ENRICH IOC CONTEXT

ADD VULNERABILITY CONTEXT

ADD CLOUD CONTEXT

CORRELATE RELATED ALERTS

BUILD EXPLAINABLE RISK SCORING

ASSIGN TRIAGE PRIORITIES

GENERATE ANALYST RECOMMENDATIONS

BUILD INVESTIGATION QUEUES

EXPORT CSV

EXPORT JSON

GENERATE MARKDOWN REPORTS

PRESERVE HUMAN APPROVAL

Final Architecture

                    SECURITY ALERTS
                          ↓
                    ALERT INGESTION
                          ↓
                     VALIDATION
                          ↓
                    NORMALIZATION
                          ↓
                    DEDUPLICATION
                          ↓
                     ENRICHMENT
        ┌─────────────────┼─────────────────┐
        ↓                 ↓                 ↓
     IDENTITY            ASSET             IOC
        ↓                 ↓                 ↓
   PRIVILEGE          CRITICALITY      REPUTATION
        ↓                 ↓                 ↓
        └────────────┬────┴────┬────────────┘
                     ↓         ↓
               VULNERABILITY  CLOUD
                     ↓         ↓
                     └────┬────┘
                          ↓
                     CORRELATION
                          ↓
                    RISK SCORING
                          ↓
                 TRIAGE RECOMMENDATION
                          ↓
                    ANALYST QUEUE
                          ↓
                     HUMAN REVIEW

01 — Create the Lab Workspace

Create:

soc-alert-triage/
|
+-- data/
|
+-- reports/
|
+-- src/
|
+-- tests/
|
+-- logs/
|
+-- README.md

Linux/macOS:

mkdir -p soc-alert-triage/{data,reports,src,tests,logs}
cd soc-alert-triage

PowerShell:

mkdir soc-alert-triage

cd soc-alert-triage

mkdir data
mkdir reports
mkdir src
mkdir tests
mkdir logs

02 — Verify Python

Run:

python --version

Recommended:

Python 3.10+

03 — Understand SOC Triage

SOC triage is the process of determining:

WHAT HAPPENED?

WHO IS INVOLVED?

WHAT SYSTEM IS INVOLVED?

HOW IMPORTANT IS IT?

WHAT CONTEXT EXISTS?

IS THERE RELATED ACTIVITY?

WHAT SHOULD THE ANALYST DO NEXT?

04 — Alert vs Incident

An:

ALERT

is a signal that requires evaluation.

An:

INCIDENT

is a security event or group of events that has been validated and managed according to the organization’s incident process.

Therefore:

ALERT
≠
CONFIRMED INCIDENT

05 — Create Synthetic Alerts

Create:

data/alerts.json

Add:

[
  {
    "alert_id": "ALT-1001",
    "timestamp": "2026-08-29T08:15:00Z",
    "source": "identity-monitor",
    "alert_type": "multiple_failed_logins",
    "severity": "medium",
    "user": "admin01",
    "asset": "JUMP01",
    "source_ip": "203.0.113.25",
    "description": "Multiple failed authentication attempts detected."
  },
  {
    "alert_id": "ALT-1002",
    "timestamp": "2026-08-29T08:18:00Z",
    "source": "identity-monitor",
    "alert_type": "successful_login_after_failures",
    "severity": "high",
    "user": "admin01",
    "asset": "JUMP01",
    "source_ip": "203.0.113.25",
    "description": "Successful authentication observed after repeated failures."
  },
  {
    "alert_id": "ALT-1003",
    "timestamp": "2026-08-29T08:22:00Z",
    "source": "endpoint-monitor",
    "alert_type": "suspicious_process",
    "severity": "high",
    "user": "user01",
    "asset": "WEB01",
    "source_ip": "198.51.100.50",
    "description": "Endpoint monitoring generated a suspicious process alert."
  },
  {
    "alert_id": "ALT-1004",
    "timestamp": "2026-08-29T08:30:00Z",
    "source": "cloud-security",
    "alert_type": "public_storage",
    "severity": "medium",
    "user": "",
    "asset": "legacy-public-files",
    "source_ip": "",
    "description": "Cloud storage resource appears publicly accessible."
  },
  {
    "alert_id": "ALT-1005",
    "timestamp": "2026-08-29T08:35:00Z",
    "source": "vulnerability-monitor",
    "alert_type": "critical_vulnerability",
    "severity": "critical",
    "user": "",
    "asset": "WEB01",
    "source_ip": "",
    "description": "Critical vulnerability identified on WEB01."
  },
  {
    "alert_id": "ALT-1006",
    "timestamp": "2026-08-29T08:40:00Z",
    "source": "network-monitor",
    "alert_type": "unusual_connection",
    "severity": "medium",
    "user": "service-web",
    "asset": "WEB01",
    "source_ip": "192.0.2.75",
    "description": "Unusual network connection requires analyst review."
  },
  {
    "alert_id": "ALT-1006",
    "timestamp": "2026-08-29T08:40:00Z",
    "source": "network-monitor",
    "alert_type": "unusual_connection",
    "severity": "medium",
    "user": "service-web",
    "asset": "WEB01",
    "source_ip": "192.0.2.75",
    "description": "Duplicate training alert."
  }
]

06 — Why Include a Duplicate?

Real pipelines can receive duplicate alerts because of:

RETRIES

MULTIPLE COLLECTORS

FORWARDER ISSUES

API PAGINATION

MESSAGE REDELIVERY

Your automation should identify duplicates before creating analyst work.

07 — Create Identity Context

Create:

data/identities.json

Add:

[
  {
    "user": "admin01",
    "department": "Cloud Operations",
    "privileged": true,
    "account_type": "human",
    "enabled": true,
    "risk_level": "high"
  },
  {
    "user": "user01",
    "department": "Application Engineering",
    "privileged": false,
    "account_type": "human",
    "enabled": true,
    "risk_level": "normal"
  },
  {
    "user": "service-web",
    "department": "Application Platform",
    "privileged": false,
    "account_type": "service",
    "enabled": true,
    "risk_level": "normal"
  }
]

08 — Why Identity Context Matters

Compare:

FAILED LOGIN

for:

TRAINING-USER

versus:

PRIVILEGED CLOUD ADMINISTRATOR

The raw detection may be similar.

The security context is not.

09 — Create Asset Context

Create:

data/assets.json

Add:

[
  {
    "asset": "JUMP01",
    "type": "jump-host",
    "environment": "production",
    "criticality": "critical",
    "internet_facing": false,
    "owner": "Cloud Operations"
  },
  {
    "asset": "WEB01",
    "type": "web-server",
    "environment": "production",
    "criticality": "high",
    "internet_facing": true,
    "owner": "Application Engineering"
  },
  {
    "asset": "legacy-public-files",
    "type": "cloud-storage",
    "environment": "production",
    "criticality": "high",
    "internet_facing": true,
    "owner": "Cloud Platform Team"
  }
]

10 — Asset Context Questions

For every alert ask:

IS IT PRODUCTION?

IS IT CRITICAL?

IS IT INTERNET-FACING?

WHO OWNS IT?

WHAT SERVICE DOES IT SUPPORT?

11 — Create IOC Context

Create:

data/ioc_context.json

Add:

[
  {
    "indicator": "203.0.113.25",
    "type": "ip",
    "source": "synthetic-training-feed",
    "confidence": 70,
    "classification": "review",
    "last_seen": "2026-08-28T18:00:00Z"
  },
  {
    "indicator": "198.51.100.50",
    "type": "ip",
    "source": "synthetic-training-feed",
    "confidence": 40,
    "classification": "informational",
    "last_seen": "2026-08-27T10:00:00Z"
  }
]

12 — IOC Context Warning

Remember:

IOC MATCH
≠
CONFIRMED COMPROMISE

IOC data contributes context.

It does not automatically determine the outcome of an investigation.

13 — Create Vulnerability Context

Create:

data/vulnerabilities.json

Add:

[
  {
    "asset": "WEB01",
    "finding_id": "VULN-2001",
    "severity": "critical",
    "score": 9.8,
    "status": "open",
    "internet_exploitable": true
  },
  {
    "asset": "JUMP01",
    "finding_id": "VULN-2002",
    "severity": "medium",
    "score": 5.4,
    "status": "open",
    "internet_exploitable": false
  }
]

14 — Why Vulnerability Context Matters

Compare:

SUSPICIOUS ACTIVITY
+
FULLY PATCHED LOW-RISK ASSET

with:

SUSPICIOUS ACTIVITY
+
INTERNET-FACING ASSET
+
OPEN CRITICAL VULNERABILITY

The second deserves greater attention.

15 — Create Cloud Context

Create:

data/cloud_context.json

Add:

[
  {
    "asset": "legacy-public-files",
    "provider": "aws",
    "scope": "production",
    "public": true,
    "encrypted": false,
    "security_findings": [
      "storage-public-access",
      "storage-encryption"
    ]
  },
  {
    "asset": "WEB01",
    "provider": "aws",
    "scope": "production",
    "public": true,
    "encrypted": true,
    "security_findings": []
  }
]

16 — Create the Triage Engine

Create:

src/soc_triage.py

Start with:

from pathlib import Path
from collections import Counter, defaultdict
from datetime import datetime, timezone
import csv
import json
import logging

17 — Configure Paths

Add:

BASE_DIR = Path(__file__).resolve().parent.parent

DATA_DIR = BASE_DIR / "data"

REPORT_DIR = BASE_DIR / "reports"

LOG_DIR = BASE_DIR / "logs"

REPORT_DIR.mkdir(
    parents=True,
    exist_ok=True
)

LOG_DIR.mkdir(
    parents=True,
    exist_ok=True
)

18 — Configure Logging

Add:

logging.basicConfig(
    level=logging.INFO,
    format=(
        "%(asctime)s "
        "%(levelname)s "
        "%(message)s"
    )
)

19 — Create JSON Loader

Add:

def load_json(
    filename
):
    path = (
        DATA_DIR /
        filename
    )

    try:
        with path.open(
            "r",
            encoding="utf-8"
        ) as file:

            return json.load(
                file
            )

    except FileNotFoundError:
        logging.error(
            "Missing file: %s",
            path
        )

        return []

    except json.JSONDecodeError as error:
        logging.error(
            "Invalid JSON in %s: %s",
            path,
            error
        )

        return []

20 — Load Lab Data

Create:

def load_lab_data():
    return {
        "alerts":
            load_json(
                "alerts.json"
            ),

        "identities":
            load_json(
                "identities.json"
            ),

        "assets":
            load_json(
                "assets.json"
            ),

        "iocs":
            load_json(
                "ioc_context.json"
            ),

        "vulnerabilities":
            load_json(
                "vulnerabilities.json"
            ),

        "cloud":
            load_json(
                "cloud_context.json"
            )
    }

21 — Define Required Alert Fields

Create:

REQUIRED_ALERT_FIELDS = {
    "alert_id",
    "timestamp",
    "source",
    "alert_type",
    "severity",
    "description"
}

22 — Validate Alert

Create:

def validate_alert(
    alert
):
    missing = (
        REQUIRED_ALERT_FIELDS
        -
        set(
            alert.keys()
        )
    )

    if missing:
        return (
            False,
            "Missing fields: "
            +
            ", ".join(
                sorted(
                    missing
                )
            )
        )

    if not str(
        alert.get(
            "alert_id",
            ""
        )
    ).strip():

        return (
            False,
            "Empty alert_id"
        )

    return (
        True,
        None
    )

23 — Validate Timestamp

Create:

def parse_timestamp(
    value
):
    try:
        return datetime.fromisoformat(
            value.replace(
                "Z",
                "+00:00"
            )
        )

    except (
        ValueError,
        AttributeError
    ):
        return None

24 — Add Timestamp Validation

Inside alert validation:

timestamp = parse_timestamp(
    alert.get(
        "timestamp"
    )
)

if timestamp is None:
    return (
        False,
        "Invalid timestamp"
    )

25 — Define Valid Severities

Create:

VALID_SEVERITIES = {
    "critical",
    "high",
    "medium",
    "low",
    "informational"
}

26 — Validate Severity

Add:

severity = str(
    alert.get(
        "severity",
        ""
    )
).strip().lower()

if severity not in VALID_SEVERITIES:
    return (
        False,
        "Invalid severity"
    )

27 — Normalize Alerts

Create:

def normalize_alert(
    alert
):
    return {
        "alert_id":
            str(
                alert[
                    "alert_id"
                ]
            ).strip(),

        "timestamp":
            str(
                alert[
                    "timestamp"
                ]
            ).strip(),

        "source":
            str(
                alert[
                    "source"
                ]
            ).strip().lower(),

        "alert_type":
            str(
                alert[
                    "alert_type"
                ]
            ).strip().lower(),

        "severity":
            str(
                alert[
                    "severity"
                ]
            ).strip().lower(),

        "user":
            str(
                alert.get(
                    "user",
                    ""
                )
            ).strip().lower(),

        "asset":
            str(
                alert.get(
                    "asset",
                    ""
                )
            ).strip().upper(),

        "source_ip":
            str(
                alert.get(
                    "source_ip",
                    ""
                )
            ).strip(),

        "description":
            str(
                alert[
                    "description"
                ]
            ).strip()
    }

28 — Process Validation

Create:

def validate_and_normalize(
    alerts
):
    valid = []

    invalid = []

    for alert in alerts:
        is_valid, reason = (
            validate_alert(
                alert
            )
        )

        if not is_valid:
            invalid.append({
                "alert":
                    alert,

                "reason":
                    reason
            })

            continue

        valid.append(
            normalize_alert(
                alert
            )
        )

    return valid, invalid

29 — Deduplicate Alerts

Create:

def deduplicate_alerts(
    alerts
):
    unique = []

    duplicates = []

    seen = set()

    for alert in alerts:
        alert_id = alert[
            "alert_id"
        ]

        if alert_id in seen:
            duplicates.append(
                alert
            )

            continue

        seen.add(
            alert_id
        )

        unique.append(
            alert
        )

    return unique, duplicates

30 — Why Deduplication Comes Early

Without deduplication:

ONE SECURITY EVENT
      ↓
FIVE DUPLICATE ALERTS
      ↓
FIVE ANALYST CASES

This creates unnecessary workload.

31 — Build Identity Index

Create:

def build_identity_index(
    identities
):
    return {
        str(
            item[
                "user"
            ]
        ).strip().lower():
            item

        for item
        in identities
    }

32 — Build Asset Index

Create:

def build_asset_index(
    assets
):
    return {
        str(
            item[
                "asset"
            ]
        ).strip().upper():
            item

        for item
        in assets
    }

33 — Build IOC Index

Create:

def build_ioc_index(
    iocs
):
    return {
        str(
            item[
                "indicator"
            ]
        ).strip():
            item

        for item
        in iocs
    }

34 — Build Vulnerability Index

One asset may have multiple vulnerabilities.

Create:

def build_vulnerability_index(
    vulnerabilities
):
    index = defaultdict(
        list
    )

    for item in vulnerabilities:
        asset = str(
            item[
                "asset"
            ]
        ).strip().upper()

        index[
            asset
        ].append(
            item
        )

    return index

35 — Build Cloud Index

Create:

def build_cloud_index(
    cloud_records
):
    return {
        str(
            item[
                "asset"
            ]
        ).strip().upper():
            item

        for item
        in cloud_records
    }

36 — Build Enrichment Indexes

Create:

def build_indexes(
    data
):
    return {
        "identity":
            build_identity_index(
                data[
                    "identities"
                ]
            ),

        "asset":
            build_asset_index(
                data[
                    "assets"
                ]
            ),

        "ioc":
            build_ioc_index(
                data[
                    "iocs"
                ]
            ),

        "vulnerability":
            build_vulnerability_index(
                data[
                    "vulnerabilities"
                ]
            ),

        "cloud":
            build_cloud_index(
                data[
                    "cloud"
                ]
            )
    }

37 — Enrich Identity Context

Create:

def enrich_identity(
    alert,
    identity_index
):
    user = alert[
        "user"
    ]

    if not user:
        return {
            "identity_found":
                False
        }

    identity = (
        identity_index.get(
            user
        )
    )

    if not identity:
        return {
            "identity_found":
                False
        }

    return {
        "identity_found":
            True,

        "department":
            identity.get(
                "department"
            ),

        "privileged":
            identity.get(
                "privileged",
                False
            ),

        "account_type":
            identity.get(
                "account_type"
            ),

        "identity_enabled":
            identity.get(
                "enabled"
            ),

        "identity_risk":
            identity.get(
                "risk_level"
            )
    }

38 — Enrich Asset Context

Create:

def enrich_asset(
    alert,
    asset_index
):
    asset = alert[
        "asset"
    ]

    if not asset:
        return {
            "asset_found":
                False
        }

    record = (
        asset_index.get(
            asset
        )
    )

    if not record:
        return {
            "asset_found":
                False
        }

    return {
        "asset_found":
            True,

        "asset_type":
            record.get(
                "type"
            ),

        "environment":
            record.get(
                "environment"
            ),

        "asset_criticality":
            record.get(
                "criticality"
            ),

        "internet_facing":
            record.get(
                "internet_facing"
            ),

        "asset_owner":
            record.get(
                "owner"
            )
    }

39 — Enrich IOC Context

Create:

def enrich_ioc(
    alert,
    ioc_index
):
    source_ip = alert[
        "source_ip"
    ]

    if not source_ip:
        return {
            "ioc_found":
                False
        }

    record = (
        ioc_index.get(
            source_ip
        )
    )

    if not record:
        return {
            "ioc_found":
                False
        }

    return {
        "ioc_found":
            True,

        "ioc_source":
            record.get(
                "source"
            ),

        "ioc_confidence":
            record.get(
                "confidence"
            ),

        "ioc_classification":
            record.get(
                "classification"
            ),

        "ioc_last_seen":
            record.get(
                "last_seen"
            )
    }

40 — Enrich Vulnerability Context

Create:

def enrich_vulnerabilities(
    alert,
    vulnerability_index
):
    asset = alert[
        "asset"
    ]

    records = (
        vulnerability_index.get(
            asset,
            []
        )
    )

    open_records = [
        item
        for item
        in records
        if item.get(
            "status"
        ) == "open"
    ]

    critical = [
        item
        for item
        in open_records
        if item.get(
            "severity"
        ) == "critical"
    ]

    exploitable = [
        item
        for item
        in open_records
        if item.get(
            "internet_exploitable"
        )
    ]

    return {
        "open_vulnerability_count":
            len(
                open_records
            ),

        "critical_vulnerability_count":
            len(
                critical
            ),

        "internet_exploitable_vulnerability_count":
            len(
                exploitable
            )
    }

41 — Enrich Cloud Context

Create:

def enrich_cloud(
    alert,
    cloud_index
):
    asset = alert[
        "asset"
    ]

    record = (
        cloud_index.get(
            asset
        )
    )

    if not record:
        return {
            "cloud_asset":
                False
        }

    return {
        "cloud_asset":
            True,

        "cloud_provider":
            record.get(
                "provider"
            ),

        "cloud_scope":
            record.get(
                "scope"
            ),

        "cloud_public":
            record.get(
                "public"
            ),

        "cloud_encrypted":
            record.get(
                "encrypted"
            ),

        "cloud_security_findings":
            record.get(
                "security_findings",
                []
            )
    }

42 — Combine Enrichment

Create:

def enrich_alert(
    alert,
    indexes
):
    enriched = (
        alert.copy()
    )

    enriched.update(
        enrich_identity(
            alert,
            indexes[
                "identity"
            ]
        )
    )

    enriched.update(
        enrich_asset(
            alert,
            indexes[
                "asset"
            ]
        )
    )

    enriched.update(
        enrich_ioc(
            alert,
            indexes[
                "ioc"
            ]
        )
    )

    enriched.update(
        enrich_vulnerabilities(
            alert,
            indexes[
                "vulnerability"
            ]
        )
    )

    enriched.update(
        enrich_cloud(
            alert,
            indexes[
                "cloud"
            ]
        )
    )

    return enriched

43 — Test Enrichment

For:

ALT-1002

you should obtain context similar to:

USER:
admin01

PRIVILEGED:
true

ASSET:
JUMP01

CRITICALITY:
critical

IOC:
203.0.113.25

IOC CONFIDENCE:
70

44 — Missing Context Is Important

If:

identity_found = false

do not automatically conclude:

USER IS SAFE

It may mean:

IDENTITY INVENTORY GAP

UNKNOWN ACCOUNT

STALE CMDB

NORMALIZATION PROBLEM

45 — Build Risk Scoring

Risk scoring should be:

EXPLAINABLE

REPEATABLE

TUNABLE

not mysterious.

46 — Base Severity Scores

Create:

BASE_SEVERITY_SCORE = {
    "critical": 50,
    "high": 40,
    "medium": 25,
    "low": 10,
    "informational": 0
}

47 — Context Scores

For this lab:

PRIVILEGED IDENTITY
+15

CRITICAL ASSET
+15

HIGH ASSET
+10

INTERNET-FACING ASSET
+10

IOC CONFIDENCE >= 70
+10

CRITICAL OPEN VULNERABILITY
+15

INTERNET-EXPLOITABLE VULNERABILITY
+10

CLOUD SECURITY FINDING
+10

These are:

TRAINING WEIGHTS

not universal SOC scoring standards.

48 — Create Risk Score Function

Add:

def calculate_risk_score(
    alert
):
    score = (
        BASE_SEVERITY_SCORE.get(
            alert[
                "severity"
            ],
            0
        )
    )

    reasons = [
        (
            "Base alert severity: "
            f"{alert['severity']}"
        )
    ]

    if alert.get(
        "privileged"
    ):
        score += 15

        reasons.append(
            "Privileged identity involved"
        )

    criticality = alert.get(
        "asset_criticality"
    )

    if criticality == "critical":
        score += 15

        reasons.append(
            "Critical asset involved"
        )

    elif criticality == "high":
        score += 10

        reasons.append(
            "High-criticality asset involved"
        )

    if alert.get(
        "internet_facing"
    ):
        score += 10

        reasons.append(
            "Internet-facing asset"
        )

    confidence = alert.get(
        "ioc_confidence"
    )

    if (
        isinstance(
            confidence,
            (int, float)
        )
        and
        confidence >= 70
    ):
        score += 10

        reasons.append(
            "Higher-confidence IOC context"
        )

    if alert.get(
        "critical_vulnerability_count",
        0
    ) > 0:

        score += 15

        reasons.append(
            "Critical open vulnerability present"
        )

    if alert.get(
        "internet_exploitable_vulnerability_count",
        0
    ) > 0:

        score += 10

        reasons.append(
            "Internet-exploitable vulnerability present"
        )

    if alert.get(
        "cloud_security_findings",
        []
    ):
        score += 10

        reasons.append(
            "Cloud security findings present"
        )

    return (
        min(
            score,
            100
        ),
        reasons
    )

49 — Why Cap the Score?

Use:

0–100

for predictable interpretation.

50 — Risk Bands

Create:

def risk_band(
    score
):
    if score >= 80:
        return "critical"

    if score >= 60:
        return "high"

    if score >= 40:
        return "medium"

    return "low"

51 — Risk Score vs Alert Severity

Do not confuse:

VENDOR ALERT SEVERITY

with:

SOC TRIAGE PRIORITY

Example:

MEDIUM ALERT
+
PRIVILEGED ADMIN
+
CRITICAL ASSET
+
HIGH-CONFIDENCE IOC
=
HIGH TRIAGE PRIORITY

52 — Add Risk to Alerts

Create:

def score_alert(
    alert
):
    enriched = (
        alert.copy()
    )

    score, reasons = (
        calculate_risk_score(
            alert
        )
    )

    enriched[
        "risk_score"
    ] = score

    enriched[
        "risk_band"
    ] = risk_band(
        score
    )

    enriched[
        "risk_reasons"
    ] = reasons

    return enriched

53 — Build Triage Recommendations

Automation should recommend:

REVIEW PRIORITY

rather than declare:

COMPROMISED

54 — Recommendation Function

Create:

def triage_recommendation(
    alert
):
    score = alert[
        "risk_score"
    ]

    if score >= 80:
        return (
            "Immediate analyst review. "
            "Validate the alert, correlate related "
            "activity, confirm asset and identity context, "
            "and follow the approved incident-response "
            "process if escalation criteria are met."
        )

    if score >= 60:
        return (
            "High-priority analyst review. "
            "Validate detection evidence and investigate "
            "related identity, endpoint, network, and "
            "vulnerability context."
        )

    if score >= 40:
        return (
            "Standard analyst review. "
            "Validate context and determine whether "
            "additional investigation is required."
        )

    return (
        "Low-priority review. "
        "Retain for normal SOC triage and correlation."
    )

55 — Add Recommendation

Create:

def add_recommendation(
    alert
):
    result = alert.copy()

    result[
        "triage_recommendation"
    ] = (
        triage_recommendation(
            alert
        )
    )

    return result

56 — Correlation

A single alert provides:

ONE SIGNAL

Correlation asks:

WHAT ELSE HAPPENED
AROUND THE SAME USER,
ASSET,
SOURCE,
OR TIME?

57 — Correlation Keys

Useful keys include:

USER

ASSET

SOURCE IP

ALERT TYPE

TIME WINDOW

58 — Count Alerts by User

Create:

def alerts_by_user(
    alerts
):
    counts = Counter()

    for alert in alerts:
        user = alert.get(
            "user"
        )

        if user:
            counts[
                user
            ] += 1

    return counts

59 — Count Alerts by Asset

Create:

def alerts_by_asset(
    alerts
):
    counts = Counter()

    for alert in alerts:
        asset = alert.get(
            "asset"
        )

        if asset:
            counts[
                asset
            ] += 1

    return counts

60 — Count Alerts by Source IP

Create:

def alerts_by_source_ip(
    alerts
):
    counts = Counter()

    for alert in alerts:
        source_ip = alert.get(
            "source_ip"
        )

        if source_ip:
            counts[
                source_ip
            ] += 1

    return counts

61 — Add Correlation Counts

Create:

def add_correlation_context(
    alerts
):
    user_counts = (
        alerts_by_user(
            alerts
        )
    )

    asset_counts = (
        alerts_by_asset(
            alerts
        )
    )

    ip_counts = (
        alerts_by_source_ip(
            alerts
        )
    )

    results = []

    for alert in alerts:
        item = alert.copy()

        item[
            "related_user_alerts"
        ] = user_counts.get(
            alert.get(
                "user"
            ),
            0
        )

        item[
            "related_asset_alerts"
        ] = asset_counts.get(
            alert.get(
                "asset"
            ),
            0
        )

        item[
            "related_source_ip_alerts"
        ] = ip_counts.get(
            alert.get(
                "source_ip"
            ),
            0
        )

        results.append(
            item
        )

    return results

62 — Review `admin01`

You should see:

ALT-1001
MULTIPLE FAILED LOGINS

ALT-1002
SUCCESS AFTER FAILURES

with the same:

USER

ASSET

SOURCE IP

This is stronger context than reviewing each alert completely independently.

63 — Review `WEB01`

You should see multiple signals involving:

WEB01

including:

SUSPICIOUS PROCESS

CRITICAL VULNERABILITY

UNUSUAL CONNECTION

64 — Correlation Does Not Prove Causation

Remember:

SAME ASSET
+
SIMILAR TIME

does not automatically prove:

SAME ATTACK

It tells the analyst:

INVESTIGATE RELATIONSHIP

65 — Add Correlation Bonus

For this training lab, you may optionally add:

3+ ALERTS ON SAME ASSET
+10

and:

2+ ALERTS FOR SAME USER
+5

66 — Correlation Score Function

Create:

def correlation_bonus(
    alert
):
    score = 0

    reasons = []

    if alert.get(
        "related_asset_alerts",
        0
    ) >= 3:

        score += 10

        reasons.append(
            "Multiple alerts involve the same asset"
        )

    if alert.get(
        "related_user_alerts",
        0
    ) >= 2:

        score += 5

        reasons.append(
            "Multiple alerts involve the same user"
        )

    return score, reasons

67 — Recalculate Final Risk

Create:

def calculate_final_risk(
    alert
):
    base_score, reasons = (
        calculate_risk_score(
            alert
        )
    )

    bonus, correlation_reasons = (
        correlation_bonus(
            alert
        )
    )

    final_score = min(
        base_score + bonus,
        100
    )

    return (
        final_score,
        reasons
        +
        correlation_reasons
    )

68 — Processing Order Matters

Your workflow should now be:

LOAD
 ↓
VALIDATE
 ↓
NORMALIZE
 ↓
DEDUPLICATE
 ↓
ENRICH
 ↓
CORRELATE
 ↓
SCORE
 ↓
RECOMMEND
 ↓
REPORT

69 — Build the Complete Processing Pipeline

Create:

def process_alerts(
    data
):
    valid, invalid = (
        validate_and_normalize(
            data[
                "alerts"
            ]
        )
    )

    unique, duplicates = (
        deduplicate_alerts(
            valid
        )
    )

    indexes = build_indexes(
        data
    )

    enriched = [
        enrich_alert(
            alert,
            indexes
        )
        for alert
        in unique
    ]

    correlated = (
        add_correlation_context(
            enriched
        )
    )

    final_alerts = []

    for alert in correlated:
        item = alert.copy()

        score, reasons = (
            calculate_final_risk(
                item
            )
        )

        item[
            "risk_score"
        ] = score

        item[
            "risk_band"
        ] = risk_band(
            score
        )

        item[
            "risk_reasons"
        ] = reasons

        item[
            "triage_recommendation"
        ] = (
            triage_recommendation(
                item
            )
        )

        final_alerts.append(
            item
        )

    return (
        final_alerts,
        invalid,
        duplicates
    )

70 — Sort the Analyst Queue

Create:

def build_analyst_queue(
    alerts
):
    return sorted(
        alerts,
        key=lambda item: (
            item[
                "risk_score"
            ],
            item[
                "timestamp"
            ]
        ),
        reverse=True
    )

71 — Analyst Queue Mental Model

Instead of:

ALERT 1

ALERT 2

ALERT 3

ALERT 4

the SOC receives:

HIGHEST CONTEXTUAL RISK
          ↓
NEXT HIGHEST
          ↓
STANDARD REVIEW
          ↓
LOWER PRIORITY

72 — Generate Alert Summary

Create:

def build_summary(
    alerts,
    invalid,
    duplicates
):
    return {
        "raw_alerts":
            (
                len(alerts)
                +
                len(duplicates)
                +
                len(invalid)
            ),

        "valid_unique_alerts":
            len(
                alerts
            ),

        "invalid_alerts":
            len(
                invalid
            ),

        "duplicates":
            len(
                duplicates
            ),

        "risk_bands":
            dict(
                Counter(
                    item[
                        "risk_band"
                    ]
                    for item
                    in alerts
                )
            ),

        "sources":
            dict(
                Counter(
                    item[
                        "source"
                    ]
                    for item
                    in alerts
                )
            )
    }

73 — Export Analyst Queue

Create:

def export_analyst_queue(
    alerts
):
    output = (
        REPORT_DIR /
        "analyst-triage-queue.csv"
    )

    fields = [
        "alert_id",
        "timestamp",
        "risk_score",
        "risk_band",
        "severity",
        "source",
        "alert_type",
        "user",
        "asset",
        "source_ip",
        "privileged",
        "asset_criticality",
        "internet_facing",
        "ioc_confidence",
        "critical_vulnerability_count",
        "related_user_alerts",
        "related_asset_alerts",
        "triage_recommendation"
    ]

    with output.open(
        "w",
        encoding="utf-8",
        newline=""
    ) as file:

        writer = csv.DictWriter(
            file,
            fieldnames=fields,
            extrasaction="ignore"
        )

        writer.writeheader()

        writer.writerows(
            build_analyst_queue(
                alerts
            )
        )

74 — Export Full JSON

Create:

def export_json(
    alerts,
    summary,
    invalid,
    duplicates
):
    output = (
        REPORT_DIR /
        "soc-triage-results.json"
    )

    with output.open(
        "w",
        encoding="utf-8"
    ) as file:

        json.dump(
            {
                "summary":
                    summary,

                "alerts":
                    alerts,

                "invalid_alerts":
                    invalid,

                "duplicates":
                    duplicates
            },
            file,
            indent=2
        )

75 — Export Invalid Alerts

Create:

def export_invalid_alerts(
    invalid
):
    output = (
        REPORT_DIR /
        "invalid-alerts.json"
    )

    output.write_text(
        json.dumps(
            invalid,
            indent=2
        ),
        encoding="utf-8"
    )

76 — Export Duplicates

Create:

def export_duplicates(
    duplicates
):
    output = (
        REPORT_DIR /
        "duplicate-alerts.json"
    )

    output.write_text(
        json.dumps(
            duplicates,
            indent=2
        ),
        encoding="utf-8"
    )

77 — Generate Markdown Investigation Report

Create:

def generate_markdown_report(
    alerts,
    summary
):
    queue = build_analyst_queue(
        alerts
    )

    lines = []

    lines.append(
        "# SOC Alert Triage Report"
    )

    lines.append("")

    lines.append(
        "## Executive Summary"
    )

    lines.append("")

    lines.append(
        f"- Valid unique alerts: "
        f"{summary['valid_unique_alerts']}"
    )

    lines.append(
        f"- Invalid alerts: "
        f"{summary['invalid_alerts']}"
    )

    lines.append(
        f"- Duplicate alerts: "
        f"{summary['duplicates']}"
    )

    lines.append("")

    lines.append(
        "## Prioritized Analyst Queue"
    )

    lines.append("")

    for alert in queue:
        lines.append(
            f"### {alert['risk_band'].upper()} — "
            f"{alert['alert_id']} — "
            f"{alert['alert_type']}"
        )

        lines.append("")

        lines.append(
            f"- Risk score: "
            f"{alert['risk_score']}"
        )

        lines.append(
            f"- Original severity: "
            f"{alert['severity']}"
        )

        lines.append(
            f"- User: "
            f"{alert['user'] or 'N/A'}"
        )

        lines.append(
            f"- Asset: "
            f"{alert['asset'] or 'N/A'}"
        )

        lines.append(
            f"- Source IP: "
            f"{alert['source_ip'] or 'N/A'}"
        )

        lines.append(
            f"- Recommendation: "
            f"{alert['triage_recommendation']}"
        )

        lines.append("")

        lines.append(
            "**Risk Factors**"
        )

        lines.append("")

        for reason in alert[
            "risk_reasons"
        ]:
            lines.append(
                f"- {reason}"
            )

        lines.append("")

    lines.append(
        "## Analyst Guidance"
    )

    lines.append("")

    lines.append(
        "Risk scores and recommendations are triage aids. "
        "They do not establish compromise. Analysts should "
        "validate source evidence, identity and asset context, "
        "related activity, known business behavior, and approved "
        "incident criteria before escalating or taking response actions."
    )

    output = (
        REPORT_DIR /
        "soc-triage-report.md"
    )

    output.write_text(
        "\n".join(
            lines
        ),
        encoding="utf-8"
    )

78 — Build `main()`

Create:

def main():
    logging.info(
        "SOC triage automation started"
    )

    data = load_lab_data()

    alerts, invalid, duplicates = (
        process_alerts(
            data
        )
    )

    summary = build_summary(
        alerts,
        invalid,
        duplicates
    )

    export_analyst_queue(
        alerts
    )

    export_json(
        alerts,
        summary,
        invalid,
        duplicates
    )

    export_invalid_alerts(
        invalid
    )

    export_duplicates(
        duplicates
    )

    generate_markdown_report(
        alerts,
        summary
    )

    logging.info(
        "SOC triage automation completed"
    )

    print(
        f"Reports saved to: "
        f"{REPORT_DIR}"
    )

79 — Add Entry Point

Add:

if __name__ == "__main__":
    main()

80 — Run the Lab

Run:

python src/soc_triage.py

Expected:

SOC triage automation started

SOC triage automation completed

Reports saved to: ...

81 — Expected Reports

You should now have:

reports/
|
+-- analyst-triage-queue.csv
|
+-- soc-triage-results.json
|
+-- invalid-alerts.json
|
+-- duplicate-alerts.json
|
+-- soc-triage-report.md

82 — Review Highest-Priority Alerts

Pay special attention to:

ALT-1002

because it combines:

HIGH ALERT SEVERITY

PRIVILEGED USER

CRITICAL ASSET

IOC CONTEXT

RELATED AUTHENTICATION ALERT

Also review alerts involving:

WEB01

because it combines:

INTERNET-FACING ASSET

HIGH CRITICALITY

CRITICAL OPEN VULNERABILITY

MULTIPLE RELATED ALERTS

83 — Do Not Hard-Code Expected Conclusions

The lab should not automatically say:

ADMIN01 IS COMPROMISED

or:

WEB01 HAS BEEN BREACHED

Instead:

HIGH-PRIORITY INVESTIGATION
RECOMMENDED

84 — Add Investigation Status

Future analyst workflow could use:

NEW

IN_REVIEW

ESCALATED

CLOSED_BENIGN

CLOSED_EXPECTED

INCIDENT_CREATED

85 — Keep Automation and Analyst Decisions Separate

Automation-generated:

risk_score

risk_band

recommendation

Analyst-generated:

disposition

investigation_notes

incident_id

closure_reason

Do not mix these concepts.

86 — Alert Disposition

An analyst may eventually determine:

TRUE POSITIVE

FALSE POSITIVE

BENIGN TRUE POSITIVE

EXPECTED ACTIVITY

INSUFFICIENT EVIDENCE

87 — Preserve Analyst Feedback

Analyst decisions can later improve:

DETECTION RULES

THRESHOLDS

ENRICHMENT

RISK SCORING

SUPPRESSION LOGIC

88 — Feedback Loop

ALERT
 ↓
TRIAGE
 ↓
ANALYST DECISION
 ↓
FEEDBACK
 ↓
DETECTION IMPROVEMENT
 ↓
BETTER ALERT

89 — Time-Window Correlation

Your current correlation uses total counts.

A stronger implementation should correlate within:

5 MINUTES

15 MINUTES

30 MINUTES

1 HOUR

depending on the use case.

90 — Why Time Windows Matter

These alerts:

FAILED LOGIN — JANUARY

SUCCESSFUL LOGIN — AUGUST

should not normally be treated like:

FAILED LOGIN — 08:15

SUCCESSFUL LOGIN — 08:18

91 — Create Time Difference Function

Create:

def minutes_between(
    first,
    second
):
    first_time = parse_timestamp(
        first
    )

    second_time = parse_timestamp(
        second
    )

    if (
        first_time is None
        or
        second_time is None
    ):
        return None

    difference = abs(
        second_time
        -
        first_time
    )

    return (
        difference.total_seconds()
        /
        60
    )

92 — Authentication Sequence Correlation

Search for:

multiple_failed_logins

followed by:

successful_login_after_failures

for the same:

USER

SOURCE IP

within a controlled time window.

93 — Create Authentication Sequence Detector

def find_auth_sequences(
    alerts,
    max_minutes=15
):
    sequences = []

    failures = [
        item
        for item
        in alerts
        if item[
            "alert_type"
        ] == "multiple_failed_logins"
    ]

    successes = [
        item
        for item
        in alerts
        if item[
            "alert_type"
        ] == "successful_login_after_failures"
    ]

    for failure in failures:
        for success in successes:

            same_user = (
                failure.get(
                    "user"
                )
                ==
                success.get(
                    "user"
                )
            )

            same_ip = (
                failure.get(
                    "source_ip"
                )
                ==
                success.get(
                    "source_ip"
                )
            )

            difference = (
                minutes_between(
                    failure[
                        "timestamp"
                    ],
                    success[
                        "timestamp"
                    ]
                )
            )

            if (
                same_user
                and
                same_ip
                and
                difference is not None
                and
                difference <= max_minutes
            ):
                sequences.append({
                    "failure_alert":
                        failure[
                            "alert_id"
                        ],

                    "success_alert":
                        success[
                            "alert_id"
                        ],

                    "user":
                        success[
                            "user"
                        ],

                    "source_ip":
                        success[
                            "source_ip"
                        ],

                    "minutes":
                        round(
                            difference,
                            2
                        )
                })

    return sequences

94 — Expected Sequence

Your synthetic data should identify:

ALT-1001
     ↓
ALT-1002

for:

admin01

203.0.113.25

within approximately:

3 MINUTES

95 — Sequence Detection Is Still Not a Conclusion

A legitimate user could:

TYPE THE WRONG PASSWORD

RETRY

THEN SUCCESSFULLY LOG IN

Therefore the sequence means:

REVIEW

not:

ACCOUNT TAKEOVER CONFIRMED

96 — Build Investigation Groups

A SOC often groups related alerts into:

CASES

STORIES

INCIDENT CANDIDATES

For the lab, group by:

USER

ASSET

SOURCE IP

97 — Group by Asset

Create:

def group_by_asset(
    alerts
):
    groups = defaultdict(
        list
    )

    for alert in alerts:
        asset = alert.get(
            "asset"
        )

        if asset:
            groups[
                asset
            ].append(
                alert[
                    "alert_id"
                ]
            )

    return dict(
        groups
    )

98 — Expected WEB01 Group

You should see:

WEB01
|
+-- ALT-1003
|
+-- ALT-1005
|
+-- ALT-1006

99 — Investigation Story

The analyst can now see:

WEB01
  ↓
SUSPICIOUS PROCESS ALERT
  ↓
CRITICAL VULNERABILITY
  ↓
UNUSUAL NETWORK CONNECTION

The automation should say:

RELATED ACTIVITY REQUIRES REVIEW

not automatically construct an attack conclusion.

100 — Add Data Freshness

Context can become stale.

Examples:

ASSET CRITICALITY

USER ROLE

IOC REPUTATION

VULNERABILITY STATUS

should have timestamps where possible.

101 — IOC Freshness

An IOC observed:

2 YEARS AGO

should not necessarily carry the same weight as one observed:

TODAY

102 — Asset Inventory Freshness

If the CMDB says:

WEB01
=
PRODUCTION

but was last updated:

18 MONTHS AGO

the analyst should know.

103 — Enrichment Confidence

Future enrichment records can include:

source

confidence

last_updated

collection_status

104 — Unknown Context

A mature pipeline should distinguish:

FALSE

from:

UNKNOWN

Example:

internet_facing = false

is different from:

internet_facing = unknown

105 — Data Quality Score

Future alerts could include:

context_completeness

Example:

IDENTITY FOUND
ASSET FOUND
IOC CHECK COMPLETED
VULNERABILITY DATA AVAILABLE

106 — Context Completeness

Conceptually:

AVAILABLE CONTEXT
/
EXPECTED CONTEXT

This helps analysts identify:

LOW-CONFIDENCE TRIAGE

107 — Create Context Completeness

Example:

def context_completeness(
    alert
):
    checks = [
        alert.get(
            "identity_found"
        )
        if alert.get(
            "user"
        )
        else True,

        alert.get(
            "asset_found"
        )
        if alert.get(
            "asset"
        )
        else True,

        (
            "ioc_found"
            in alert
        ),

        (
            "open_vulnerability_count"
            in alert
        )
    ]

    completed = sum(
        bool(item)
        for item
        in checks
    )

    return round(
        (
            completed
            /
            len(checks)
        )
        *
        100,
        2
    )

108 — Why Context Completeness Matters

An alert with:

RISK SCORE = 85

but:

CONTEXT COMPLETENESS = 25%

should be interpreted carefully.

109 — Detection Confidence

Future alert records may contain:

detection_confidence

Do not confuse:

DETECTION CONFIDENCE

with:

INCIDENT CONFIDENCE

110 — Build Alert Fingerprint

Alert IDs may change across systems.

A fingerprint can combine:

SOURCE

ALERT TYPE

USER

ASSET

SOURCE IP

TIME BUCKET

111 — Why Fingerprints Matter

They can help with:

DEDUPLICATION

CORRELATION

SUPPRESSION

TRENDING

112 — Suppression

Some alerts may be:

KNOWN

EXPECTED

APPROVED

Example:

AUTHORIZED VULNERABILITY SCANNER

A suppression should require:

OWNER

JUSTIFICATION

EXPIRATION

SCOPE

113 — Suppression Warning

Do not create:

IGNORE THIS ALERT FOREVER

because an exception can become stale.

114 — Suppression Workflow

NOISY ALERT
    ↓
ANALYST REVIEW
    ↓
EXPECTED?
    ↓
DOCUMENT
    ↓
APPROVE
    ↓
TIME-LIMITED SUPPRESSION
    ↓
REVIEW

115 — SLA Mapping

Organizations may define:

CRITICAL
→ IMMEDIATE REVIEW

HIGH
→ RAPID REVIEW

MEDIUM
→ STANDARD QUEUE

LOW
→ ROUTINE REVIEW

Use organizational policy rather than inventing universal response times.

116 — Queue Ownership

Future alerts should include:

assigned_team

Examples:

SOC

CLOUD SECURITY

IDENTITY TEAM

ENDPOINT TEAM

APPLICATION SECURITY

117 — Routing Function

Example:

def route_alert(
    alert
):
    source = alert[
        "source"
    ]

    if source == "cloud-security":
        return "Cloud Security"

    if source == "identity-monitor":
        return "SOC / Identity"

    if source == "endpoint-monitor":
        return "SOC / Endpoint"

    if source == "vulnerability-monitor":
        return "Vulnerability Management"

    return "SOC"

118 — Add Routing

Add:

item[
    "assigned_team"
] = route_alert(
    item
)

to the final processing pipeline.

119 — Routing Does Not Replace Escalation

A cloud alert assigned to:

CLOUD SECURITY

may still require:

SOC

INCIDENT RESPONSE

IDENTITY

LEGAL

PRIVACY

depending on the investigation.

120 — Create Case Candidate

For high-context alerts, automation can create a:

CASE CANDIDATE

rather than automatically declaring an incident.

Example:

{
  "case_candidate": true,
  "reason": "Multiple related high-risk alerts"
}

121 — Case Candidate Rule

For the training lab:

def case_candidate(
    alert
):
    return (
        alert[
            "risk_score"
        ] >= 80
        and
        (
            alert.get(
                "related_user_alerts",
                0
            ) >= 2
            or
            alert.get(
                "related_asset_alerts",
                0
            ) >= 2
        )
    )

122 — Why `Candidate`?

Because:

AUTOMATION

can identify:

PATTERN WORTH ESCALATING

but:

ANALYST

should validate whether incident criteria are actually met.

123 — Evidence Preservation

For high-priority alerts, preserve:

RAW ALERT

NORMALIZED ALERT

ENRICHMENT DATA

CORRELATION RESULTS

RISK REASONS

TIMESTAMP

124 — Never Modify Raw Evidence

Keep:

RAW DATA

separate from:

NORMALIZED DATA

125 — Evidence Architecture

RAW ALERT
    ↓
PRESERVE
    ↓
COPY
    ↓
NORMALIZE
    ↓
ENRICH
    ↓
ANALYZE

126 — Export Raw Alert Snapshot

Create:

def export_raw_alerts(
    alerts
):
    output = (
        REPORT_DIR /
        "raw-alert-snapshot.json"
    )

    output.write_text(
        json.dumps(
            alerts,
            indent=2
        ),
        encoding="utf-8"
    )

Call it before transformation.

127 — Hash Raw Evidence

Import:

import hashlib

Create:

def sha256_file(
    path
):
    digest = hashlib.sha256()

    with path.open(
        "rb"
    ) as file:

        for block in iter(
            lambda:
                file.read(
                    8192
                ),
            b""
        ):
            digest.update(
                block
            )

    return digest.hexdigest()

128 — Evidence Hash Purpose

The hash helps answer:

HAS THIS FILE CHANGED
SINCE COLLECTION?

It does not by itself establish the full chain of custody.

129 — Create Investigation Package

Future high-priority case folder:

cases/
|
+-- CASE-001/
    |
    +-- raw-alert.json
    |
    +-- normalized-alert.json
    |
    +-- identity-context.json
    |
    +-- asset-context.json
    |
    +-- vulnerability-context.json
    |
    +-- analyst-notes.md

130 — Integrate Security APIs

In a real authorized environment, enrichment could use:

SIEM API

EDR API

CMDB API

IDENTITY API

THREAT INTELLIGENCE API

VULNERABILITY API

CLOUD SECURITY API

Use the lessons from:

Lab 07 — Security API Integration

131 — API Failure Handling

If threat-intelligence enrichment fails:

DO NOT DROP THE ALERT

Instead:

ALERT
 ↓
IOC ENRICHMENT FAILED
 ↓
MARK CONTEXT UNAVAILABLE
 ↓
CONTINUE TRIAGE

132 — Graceful Degradation

A resilient pipeline should continue when one enrichment source is unavailable.

IDENTITY API
   ✓

ASSET API
   ✓

IOC API
   ✗

VULNERABILITY API
   ✓

Result:

PARTIAL ENRICHMENT

not:

TOTAL PIPELINE FAILURE

133 — Retry Carefully

Retry:

TEMPORARY NETWORK ERROR

429

5XX

Do not repeatedly retry:

401

403

INVALID REQUEST

134 — Secret Management

API tokens must not appear in:

SOURCE CODE

REPORTS

LOGS

GIT

ERROR OUTPUT

135 — SOC Automation Logging

Useful operational logs include:

ALERTS RECEIVED

ALERTS VALIDATED

INVALID ALERTS

DUPLICATES REMOVED

ENRICHMENT FAILURES

ALERTS SCORED

REPORTS GENERATED

PIPELINE DURATION

136 — Metrics

Create SOC automation metrics such as:

RAW ALERT COUNT

VALID ALERT COUNT

DUPLICATE RATE

ENRICHMENT SUCCESS RATE

HIGH-PRIORITY COUNT

AVERAGE RISK SCORE

PROCESSING TIME

137 — Alert Reduction Metric

Example:

RAW ALERTS
1000

DUPLICATES
200

VALID UNIQUE
800

Deduplication reduced analyst workload by:

20%

138 — Be Careful With Automation Metrics

A lower alert count is not automatically:

BETTER SECURITY

You could reduce alerts by:

DISABLING DETECTIONS

which would be harmful.

Measure:

QUALITY

COVERAGE

PRECISION

RESPONSE VALUE

not only volume.

139 — Analyst Feedback Dataset

Future data:

{
  "alert_id": "ALT-1002",
  "analyst_disposition": "true_positive",
  "escalated": true,
  "notes": "Synthetic training result"
}

140 — Detection Tuning

Analyze analyst feedback to identify:

HIGH FALSE-POSITIVE RULES

LOW-VALUE ALERTS

MISSING CONTEXT

BAD THRESHOLDS

USEFUL CORRELATIONS

141 — Never Auto-Tune Blindly

Do not automatically disable a detection because:

MANY ALERTS WERE CLOSED

Investigate:

WHY?

142 — Build Test File

Create:

tests/test_soc_triage.py

143 — Test Alert Validation

def test_valid_alert():
    alert = {
        "alert_id": "ALT-1",
        "timestamp":
            "2026-08-29T08:00:00Z",
        "source":
            "training",
        "alert_type":
            "test",
        "severity":
            "medium",
        "description":
            "Training alert"
    }

    valid, reason = (
        validate_alert(
            alert
        )
    )

    assert valid is True
    assert reason is None

144 — Test Missing Alert ID

def test_missing_alert_id():
    alert = {
        "timestamp":
            "2026-08-29T08:00:00Z",
        "source":
            "training",
        "alert_type":
            "test",
        "severity":
            "medium",
        "description":
            "Training alert"
    }

    valid, reason = (
        validate_alert(
            alert
        )
    )

    assert valid is False

145 — Test Invalid Severity

Use:

severity = emergency-super-critical

Expected:

INVALID

146 — Test Invalid Timestamp

Use:

not-a-date

Expected:

INVALID

147 — Test Deduplication

Input:

ALT-1001

ALT-1001

Expected:

UNIQUE = 1

DUPLICATE = 1

148 — Test Identity Enrichment

Input:

user = admin01

Expected:

privileged = true

149 — Test Asset Enrichment

Input:

asset = WEB01

Expected:

criticality = high

internet_facing = true

150 — Test IOC Enrichment

Input:

203.0.113.25

Expected:

ioc_found = true

confidence = 70

151 — Test Unknown IOC

Input:

192.0.2.200

Expected:

ioc_found = false

Do not mark it:

SAFE

just because your dataset has no record.

152 — Test Vulnerability Context

Input:

WEB01

Expected:

critical_vulnerability_count >= 1

153 — Test Risk Scoring

Build a synthetic alert with:

HIGH SEVERITY

PRIVILEGED USER

CRITICAL ASSET

Confirm the score increases for each documented reason.

154 — Test Risk Cap

Create enough contextual factors to exceed:

Expected:

risk_score = 100

155 — Test Correlation

Input multiple alerts using:

WEB01

Confirm:

related_asset_alerts

is correct.

156 — Test Authentication Sequence

Confirm:

ALT-1001

and:

ALT-1002

are correlated within:

15 MINUTES

157 — Test Graceful Missing Identity

Use:

user = unknown-user

Expected:

identity_found = false

with no crash.

158 — Test Missing Asset

Use:

asset = UNKNOWN01

Expected:

asset_found = false

159 — Test Empty IOC Data

Replace:

ioc_context.json

temporarily with:

[]

The triage engine should still process alerts.

160 — Test Missing Vulnerability Data

The pipeline should continue with:

open_vulnerability_count = 0

while ideally recording that collection status separately in a more mature implementation.

161 — Challenge 01 — Add SQLite

Store normalized alerts in:

soc.db

Tables:

alerts

identities

assets

vulnerabilities

triage_results

162 — SQL Investigation

Use skills from:

Lab 05 — SQL Security Analytics

to query:

SELECT
    asset,
    COUNT(*) AS alert_count
FROM alerts
GROUP BY asset
ORDER BY alert_count DESC;

163 — Challenge 02 — Add API Enrichment

Replace one static enrichment file with a:

LOCAL TRAINING API

from Lab 07.

Architecture:

ALERT
 ↓
PYTHON
 ↓
TRAINING API
 ↓
JSON
 ↓
ENRICHMENT

164 — Challenge 03 — Add Cloud Findings

Feed the output of:

Lab 08

directly into the SOC triage pipeline.

CLOUD AUDITOR
      ↓
CLOUD FINDINGS
      ↓
SOC TRIAGE

165 — Challenge 04 — Add Vulnerability Prioritization

Use output from:

Lab 06

to enrich affected assets.

166 — Challenge 05 — Add IOC Pipeline

Use normalized IOC output from:

Lab 02

rather than the simplified IOC context file.

167 — Challenge 06 — Add Log Analyzer Findings

Use:

Lab 01

to create authentication alerts that feed this lab.

Architecture:

AUTH LOG
   ↓
LOG ANALYZER
   ↓
ALERT
   ↓
SOC TRIAGE

168 — Challenge 07 — Build Dashboard Dataset

Export:

dashboard.json

with:

TOTAL ALERTS

CRITICAL QUEUE

HIGH QUEUE

TOP USERS

TOP ASSETS

TOP ALERT TYPES

TOP SOURCES

169 — Challenge 08 — Add JavaScript Dashboard

Use your:

JavaScript Fundamentals

skills to create a simple local dashboard that reads sanitized lab data.

Do not expose:

TOKENS

SECRETS

SENSITIVE RAW EVIDENCE

in browser-side code.

170 — Challenge 09 — Add MITRE ATT&CK Mapping

Add optional fields such as:

technique_id

tactic

to synthetic alerts.

Use mappings for:

CLASSIFICATION

INVESTIGATION CONTEXT

REPORTING

not as proof that an adversary definitely performed a technique.

171 — Challenge 10 — Add Detection Rule ID

Every alert should eventually identify:

rule_id

Example:

AUTH-001

ENDPOINT-004

CLOUD-012

172 — Challenge 11 — Add Rule Version

Track:

rule_version

This helps answer:

WHICH VERSION
GENERATED THE ALERT?

173 — Challenge 12 — Add Detection Source Evidence

Store:

event_id

log_source

sensor

query_reference

without copying unnecessary sensitive raw data into every report.

174 — Challenge 13 — Add Case Candidates

Automatically create:

case-candidates.json

for alerts satisfying your controlled candidate criteria.

Do not automatically mark them:

CONFIRMED INCIDENTS

175 — Challenge 14 — Add SLA Queue

Create:

critical

high

medium

low

queues.

Use organizational response targets rather than arbitrary universal SLAs.

176 — Challenge 15 — Add Analyst Notes Template

Generate:

analyst-investigation-template.md

containing:

ALERT ID

SUMMARY

IDENTITY CONTEXT

ASSET CONTEXT

IOC CONTEXT

VULNERABILITY CONTEXT

RELATED ALERTS

TIMELINE

ANALYST OBSERVATIONS

DISPOSITION

ESCALATION

FOLLOW-UP

177 — Challenge 16 — Add Pipeline Health Report

Generate:

pipeline-health.json

containing:

ALERT SOURCE STATUS

IDENTITY SOURCE STATUS

ASSET SOURCE STATUS

IOC SOURCE STATUS

VULNERABILITY SOURCE STATUS

CLOUD SOURCE STATUS

178 — Challenge 17 — Add Dry Run

Any future response workflow should default to:

DRY RUN

Example:

PROPOSED ACTION:

Request review of account admin01

NO CHANGE PERFORMED

179 — Challenge 18 — Add Approval Gate

Future response architecture:

HIGH-RISK ALERT
      ↓
TRIAGE
      ↓
PROPOSE ACTION
      ↓
ANALYST APPROVAL
      ↓
AUTHORIZED RESPONSE
      ↓
VERIFY

180 — SOAR Relationship

What you are building resembles a simplified:

SOAR

workflow.

SOAR commonly stands for:

SECURITY

ORCHESTRATION

AUTOMATION

RESPONSE

181 — Orchestration vs Automation

Automation:

DO ONE TASK
AUTOMATICALLY

Orchestration:

CONNECT MULTIPLE
SECURITY SYSTEMS
INTO A WORKFLOW

Example:

SIEM
 ↓
IDENTITY
 ↓
CMDB
 ↓
THREAT INTEL
 ↓
VULNERABILITY PLATFORM
 ↓
CASE MANAGEMENT

182 — Why Human-in-the-Loop Matters

A SOC workflow may have permission to:

DISABLE USER

ISOLATE HOST

BLOCK INDICATOR

A false positive could therefore cause:

BUSINESS OUTAGE

USER LOCKOUT

SERVICE DISRUPTION

Use human approval for high-impact decisions unless a carefully governed automation use case has been explicitly approved.

183 — Risk Scoring Governance

Your scoring model must be documented.

Analysts should understand:

WHY DID THIS ALERT
GET SCORE 90?

184 — Bad Risk Model

Avoid:

ALERT SCORE = 93

WHY?

UNKNOWN.

185 — Good Risk Model

Prefer:

BASE HIGH SEVERITY
+40

PRIVILEGED USER
+15

CRITICAL ASSET
+15

HIGH-CONFIDENCE IOC
+10

RELATED USER ALERTS
+5

TOTAL
85

186 — Explainability

Every score should include:

REASONS

because explainability supports:

ANALYST TRUST

TUNING

AUDITING

DEBUGGING

187 — Avoid Double Counting

Be careful when:

ALERT SEVERITY

already includes:

ASSET CRITICALITY

and your risk model adds:

ASSET CRITICALITY AGAIN

You may accidentally inflate risk.

188 — Risk Model Versioning

Add:

risk_model_version

Example:

1.0

This helps explain historical differences.

189 — Detection Quality Metrics

Future metrics:

TRUE POSITIVE RATE

FALSE POSITIVE RATE

BENIGN TRUE POSITIVE RATE

ESCALATION RATE

MEAN TIME TO TRIAGE

ENRICHMENT COVERAGE

190 — Mean Time to Triage

Conceptually:

TRIAGE TIME
-
ALERT CREATED TIME

Track this carefully using consistent timestamps.

191 — Queue Aging

Monitor alerts that remain:

UNREVIEWED

for too long.

ALERT
 ↓
QUEUE
 ↓
AGE
 ↓
ESCALATION POLICY

192 — Auditability

The system should record:

WHEN ALERT ARRIVED

WHEN ENRICHMENT OCCURRED

WHAT DATA SOURCES WERE USED

WHAT SCORE WAS ASSIGNED

WHAT MODEL VERSION WAS USED

WHO REVIEWED IT

WHAT DECISION WAS MADE

193 — Data Retention

SOC data can contain:

USERNAMES

HOSTNAMES

IP ADDRESSES

SECURITY EVENTS

INVESTIGATION NOTES

Follow approved:

RETENTION

ACCESS CONTROL

PRIVACY

EVIDENCE

requirements.

194 — Security of the Automation Platform

The triage platform itself becomes:

SECURITY-CRITICAL INFRASTRUCTURE

Protect:

API TOKENS

DATABASE

LOGS

REPORTS

CONFIGURATION

SCORING RULES

SOURCE CODE

195 — Least Privilege

A triage engine that only:

READS ALERTS

should not automatically receive permission to:

ISOLATE HOSTS

DISABLE ACCOUNTS

DELETE RESOURCES

196 — Separate Read and Response Credentials

Prefer:

TRIAGE SERVICE
      ↓
READ-ONLY TOKEN

and a separately governed:

RESPONSE WORKFLOW
      ↓
APPROVED WRITE PERMISSION

197 — Fail-Safe Behavior

If enrichment fails:

DO NOT
AUTOMATICALLY CLOSE ALERT

If scoring fails:

DO NOT
AUTOMATICALLY MARK LOW RISK

If context is unavailable:

MARK UNKNOWN
AND REQUEST REVIEW

198 — Fail-Safe Mental Model

UNCERTAINTY
    ↓
VISIBILITY
    ↓
HUMAN REVIEW

not:

UNCERTAINTY
    ↓
IGNORE

199 — Common SOC Automation Mistakes

Avoid:

TRUSTING ALERT SEVERITY BLINDLY

NO NORMALIZATION

NO DEDUPLICATION

NO ASSET CONTEXT

NO IDENTITY CONTEXT

IOC MATCH = COMPROMISE

NO VULNERABILITY CONTEXT

NO CORRELATION

NO TIME WINDOWS

NO EXPLAINABLE SCORING

NO DATA FRESHNESS

MISSING DATA = SAFE

AUTOMATICALLY CLOSING ALERTS

AUTOMATICALLY DISABLING USERS

AUTOMATICALLY ISOLATING HOSTS

LOGGING API TOKENS

NO ANALYST FEEDBACK

NO AUDIT TRAIL

200 — SOC Analyst Investigation Questions

For every high-priority alert, ask:

What detection generated the alert?
What evidence supports it?
Is the alert data complete?
Which user is involved?
Is the identity privileged?
Is the account enabled?
Is the behavior normal for this identity?
Which asset is involved?
Is the asset production?
Is it business-critical?
Is it internet-facing?
Are critical vulnerabilities present?
Is the source indicator known?
How current is the IOC information?
Are there related alerts?
Are the alerts close together in time?
Is there an approved exception?
Could this be legitimate activity?
Does it meet incident escalation criteria?
What evidence should be preserved?

201 — Final Project Structure

Your completed project should look like:

soc-alert-triage/
|
+-- data/
|   +-- alerts.json
|   +-- identities.json
|   +-- assets.json
|   +-- ioc_context.json
|   +-- vulnerabilities.json
|   +-- cloud_context.json
|
+-- src/
|   +-- soc_triage.py
|
+-- tests/
|   +-- test_soc_triage.py
|
+-- reports/
|   +-- raw-alert-snapshot.json
|   +-- analyst-triage-queue.csv
|   +-- soc-triage-results.json
|   +-- invalid-alerts.json
|   +-- duplicate-alerts.json
|   +-- soc-triage-report.md
|
+-- logs/
|
+-- README.md
|
+-- architecture.md

202 — README Structure

Document:

PROJECT OVERVIEW

SOC USE CASE

ARCHITECTURE

AUTHORIZED USE

DATA SOURCES

ALERT SCHEMA

VALIDATION

NORMALIZATION

DEDUPLICATION

IDENTITY ENRICHMENT

ASSET ENRICHMENT

IOC ENRICHMENT

VULNERABILITY ENRICHMENT

CLOUD CONTEXT

CORRELATION

RISK MODEL

TRIAGE RECOMMENDATIONS

ANALYST QUEUE

REPORTS

TESTING

SECURITY CONSIDERATIONS

LIMITATIONS

203 — Document Lab Limitations

The lab uses:

SYNTHETIC ALERTS

STATIC JSON ENRICHMENT

SIMPLE CORRELATION

TRAINING RISK WEIGHTS

NO PRODUCTION SIEM

NO PRODUCTION EDR

NO REAL THREAT-INTELLIGENCE FEED

NO AUTOMATIC RESPONSE

204 — Portfolio Deliverables

Your portfolio should demonstrate:

ALERT INGESTION

DATA VALIDATION

NORMALIZATION

DEDUPLICATION

MULTI-SOURCE ENRICHMENT

CORRELATION

EXPLAINABLE RISK SCORING

ANALYST QUEUE GENERATION

REPORTING

TESTING

SAFE AUTOMATION DESIGN

205 — Mission Validation Checklist

Mission Review

You started with:

RAW ALERTS

containing only basic information such as:

ALERT TYPE

SEVERITY

USER

ASSET

SOURCE IP

You transformed them into:

NORMALIZED ALERT
      ↓
IDENTITY CONTEXT
      ↓
ASSET CONTEXT
      ↓
IOC CONTEXT
      ↓
VULNERABILITY CONTEXT
      ↓
CLOUD CONTEXT
      ↓
CORRELATION
      ↓
EXPLAINABLE RISK SCORE
      ↓
TRIAGE RECOMMENDATION
      ↓
PRIORITIZED ANALYST QUEUE

What You Built

You now have the foundation of a defensive SOC triage automation platform.

It can:

INGEST

VALIDATE

NORMALIZE

DEDUPLICATE

ENRICH

CORRELATE

PRIORITIZE

ROUTE

REPORT

security alerts.

Key Security Lesson

The most important lesson is:

ALERT
+
CONTEXT
≠
AUTOMATIC CONCLUSION

Instead:

ALERT
+
IDENTITY
+
ASSET
+
IOC
+
VULNERABILITY
+
CLOUD
+
RELATED ACTIVITY
+
ANALYST JUDGMENT
=
BETTER SECURITY DECISION

Final Mental Model

When an alert arrives, think:

IS THE ALERT VALID?
        ↓
IS IT A DUPLICATE?
        ↓
WHO IS INVOLVED?
        ↓
IS THE USER PRIVILEGED?
        ↓
WHAT ASSET IS INVOLVED?
        ↓
HOW CRITICAL IS IT?
        ↓
IS IT INTERNET-FACING?
        ↓
IS THERE IOC CONTEXT?
        ↓
ARE VULNERABILITIES PRESENT?
        ↓
ARE CLOUD FINDINGS PRESENT?
        ↓
ARE THERE RELATED ALERTS?
        ↓
HOW CLOSE ARE THEY IN TIME?
        ↓
WHAT FACTORS DRIVE PRIORITY?
        ↓
WHAT SHOULD THE ANALYST REVIEW?
        ↓
DOES IT MEET ESCALATION CRITERIA?

The goal is not:

REMOVE THE ANALYST

The goal is:

REMOVE REPETITIVE WORK
SO THE ANALYST
CAN MAKE BETTER DECISIONS

What’s Next?

➡️ Lab 10 — Enterprise Security Automation Capstone

The final Programming Lab brings the entire learning path together.

You will combine:

PYTHON

BASH

POWERSHELL

SQL

JSON

APIs

LINUX SECURITY DATA

WINDOWS SECURITY DATA

CLOUD SECURITY DATA

VULNERABILITY DATA

SOC ALERTS

IOC ENRICHMENT

into one enterprise security automation platform.

The capstone architecture will be:

          ENTERPRISE SECURITY DATA
                    ↓
     ┌──────────────┼──────────────┐
     ↓              ↓              ↓
   LINUX          WINDOWS         CLOUD
     ↓              ↓              ↓
     └──────────────┼──────────────┘
                    ↓
              VULNERABILITIES
                    ↓
                   IOCS
                    ↓
                SOC ALERTS
                    ↓
                INGESTION
                    ↓
               NORMALIZATION
                    ↓
                 DATABASE
                    ↓
               CORRELATION
                    ↓
              RISK SCORING
                    ↓
             ANALYST REPORT
                    ↓
               HUMAN REVIEW

This will be the final lab where you build a portfolio-ready enterprise security automation project from the skills developed throughout the complete Programming learning path.

Lab 09 — SOC Alert Triage Automation

Mission Information

Mission

Why This Lab Matters

Important Principle

Learning Objectives

Final Architecture

01 — Create the Lab Workspace

02 — Verify Python

03 — Understand SOC Triage

04 — Alert vs Incident

05 — Create Synthetic Alerts

06 — Why Include a Duplicate?

07 — Create Identity Context

08 — Why Identity Context Matters

09 — Create Asset Context

10 — Asset Context Questions

11 — Create IOC Context

12 — IOC Context Warning

13 — Create Vulnerability Context

14 — Why Vulnerability Context Matters

15 — Create Cloud Context

16 — Create the Triage Engine

17 — Configure Paths

18 — Configure Logging

19 — Create JSON Loader

20 — Load Lab Data

21 — Define Required Alert Fields

22 — Validate Alert

23 — Validate Timestamp

24 — Add Timestamp Validation

25 — Define Valid Severities

26 — Validate Severity

27 — Normalize Alerts

28 — Process Validation

29 — Deduplicate Alerts

30 — Why Deduplication Comes Early

31 — Build Identity Index

32 — Build Asset Index

33 — Build IOC Index

34 — Build Vulnerability Index

35 — Build Cloud Index

36 — Build Enrichment Indexes

37 — Enrich Identity Context

38 — Enrich Asset Context

39 — Enrich IOC Context

40 — Enrich Vulnerability Context

41 — Enrich Cloud Context

42 — Combine Enrichment

43 — Test Enrichment

44 — Missing Context Is Important

45 — Build Risk Scoring

46 — Base Severity Scores

47 — Context Scores

48 — Create Risk Score Function

49 — Why Cap the Score?

50 — Risk Bands

51 — Risk Score vs Alert Severity

52 — Add Risk to Alerts

53 — Build Triage Recommendations

54 — Recommendation Function

55 — Add Recommendation

56 — Correlation

57 — Correlation Keys

58 — Count Alerts by User

59 — Count Alerts by Asset

60 — Count Alerts by Source IP

61 — Add Correlation Counts

62 — Review admin01

63 — Review WEB01

64 — Correlation Does Not Prove Causation

65 — Add Correlation Bonus

66 — Correlation Score Function

67 — Recalculate Final Risk

68 — Processing Order Matters

69 — Build the Complete Processing Pipeline

70 — Sort the Analyst Queue

71 — Analyst Queue Mental Model

72 — Generate Alert Summary

73 — Export Analyst Queue

62 — Review `admin01`

63 — Review `WEB01`

78 — Build `main()`

122 — Why `Candidate`?