Dynamic Severity Scoring in SOC Automation Pipelines

Static severity labels collapse under modern attack velocity. When a single event triggers a P1 escalation regardless of asset criticality, identity context, or concurrent telemetry, analysts drown in noise and real intrusions get buried. Dynamic severity scoring replaces fixed priority tiers with context-aware, continuously recalculated risk values computed at the correlation layer. As part of the broader Alert Correlation & Rule Engines pipeline, this scoring stage sits downstream of normalization and correlation and upstream of routing — it is the decision function that converts an enriched event into an actionable priority. This guide walks through a production-grade Python implementation: the data model, the weighting math, score saturation, failure handling, and the observability you need to trust the output in a live Security Operations Center.

Problem Framing

Consider a mid-size enterprise SOC ingesting roughly 28,000 events per second across EDR, cloud audit logs (AWS CloudTrail, Azure Activity), identity providers, and network flow records. The detection content emits a flat severity drawn from vendor defaults or raw CVSS. The result is predictable: tens of thousands of “medium” alerts per shift, no two of which carry the same real-world risk, and an analyst pool that can meaningfully triage perhaps a few dozen.

The concrete failure is severity that ignores context. A failed authentication on an ephemeral CI runner and a failed authentication on a domain controller arrive with identical baseline scores. A PowerShell execution that maps to T1059.001 PowerShell looks the same as a benign admin script. Worse, when the same compromised principal appears in three telemetry sources within a 15-minute window, the static model still emits three independent mediums instead of one high-confidence critical. Dynamic severity scoring resolves this by applying deterministic, auditable multipliers — asset criticality, ATT&CK weighting, identity risk, and cross-source confidence — to a lightweight baseline, producing a single bounded score that reflects organizational risk rather than vendor taxonomy.

Prerequisites & Environment

The reference engine targets Python 3.11+ for StrEnum, the typing improvements, and faster startup. The scoring core is intentionally dependency-light so it can run inside a Kafka consumer, a SIEM webhook handler, or a SOAR step function without pulling a heavy framework.

python3 -m venv .venv
source .venv/bin/activate
pip install "pydantic>=2.6" "structlog>=24.1"
# Optional, for stream integration shown later:
pip install "aiokafka>=0.10"

Infrastructure assumptions:

• A normalization stage already emits canonical events. If yours does not, build it first via the schema validation pipelines under the ingestion layer — scoring quality is bounded by field hygiene upstream.
• Enrichment lookups (asset CMDB tier, identity risk, threat-intel verdict) are resolved before the event reaches the scorer. Scoring must stay synchronous and CPU-bound; network calls belong in the enrichment gate.
• A dead-letter topic or table exists for events that fail validation, so nothing is silently dropped.

Architecture Overview

The scorer is a pure function wrapped in a thin service: a validated NormalizedEvent goes in, a bounded integer severity and a structured decision record come out. State (sliding windows, entity tracking) lives in the correlation layer, not here, which keeps scoring idempotent and horizontally scalable.

Step-by-Step Implementation

Step 1 — Model the normalized event

Use Pydantic so the scorer rejects malformed input at the boundary instead of producing a garbage score. This model is the contract between the correlation layer and the scoring layer.

from __future__ import annotations

from enum import Enum
from typing import Annotated

from pydantic import BaseModel, Field, field_validator


class AssetCriticality(float, Enum):
    """Multiplier applied to the baseline based on asset tier."""
    SANDBOX = 0.5      # decommissioned / ephemeral
    LOW = 1.0
    MEDIUM = 1.5
    HIGH = 2.0
    CRITICAL = 3.0     # domain controllers, prod databases


Severity = Annotated[int, Field(ge=1, le=10)]


class NormalizedEvent(BaseModel):
    event_id: str
    baseline_severity: Severity
    asset_criticality: AssetCriticality = AssetCriticality.LOW
    attck_techniques: list[str] = Field(default_factory=list)   # e.g. ["T1059.001"]
    identity_risk_score: float = Field(default=0.0, ge=0.0, le=1.0)
    cross_source_confidence: float = Field(default=1.0, ge=0.0, le=1.0)
    metadata: dict[str, str] = Field(default_factory=dict)

    @field_validator("attck_techniques")
    @classmethod
    def _strip_blanks(cls, v: list[str]) -> list[str]:
        return [t.strip().upper() for t in v if t and t.strip()]

Step 2 — Declare the weight tables and config

Keep every tunable knob in one configuration object so platform teams can hot-reload it through config-as-code without redeploying the service. Technique weights anchor to the MITRE ATT&CK framework so prioritization tracks kill-chain progression rather than isolated anomalies.

from pydantic import BaseModel, Field


class ScoringConfig(BaseModel):
    min_severity: int = 1
    max_severity: int = 10
    # Per-tactic intrinsic weight; credential access outranks reconnaissance.
    technique_weights: dict[str, float] = Field(
        default_factory=lambda: {
            "T1566": 1.4,      # Phishing (Initial Access)
            "T1078": 1.5,      # Valid Accounts (Defense Evasion / Persistence)
            "T1059.001": 1.6,  # PowerShell (Execution)
            "T1003": 1.8,      # OS Credential Dumping (Credential Access)
            "T1595": 0.8,      # Active Scanning (Reconnaissance)
        }
    )
    default_technique_weight: float = 1.2
    identity_risk_weight: float = 1.3
    decay_half_life: float = 5.0   # raw-score units per logistic half-step

Step 3 — Implement the deterministic scoring core

The core multiplies the baseline by each contextual factor, then passes the raw product through a logistic decay so scores saturate toward max_severity instead of overflowing. Every weight applied is captured in a decision record for audit.

import structlog

log = structlog.get_logger("dynamic_severity_scorer")


class DynamicSeverityEngine:
    def __init__(self, config: ScoringConfig | None = None) -> None:
        self.config = config or ScoringConfig()

    def _technique_multiplier(self, techniques: list[str]) -> float:
        if not techniques:
            return 1.0
        # Take the strongest matching technique; chained TTPs compound elsewhere.
        return max(
            self.config.technique_weights.get(t, self.config.default_technique_weight)
            for t in techniques
        )

    def score(self, event: NormalizedEvent) -> dict[str, object]:
        cfg = self.config
        asset_m = float(event.asset_criticality.value)
        tech_m = self._technique_multiplier(event.attck_techniques)
        identity_m = 1.0 + (event.identity_risk_score * cfg.identity_risk_weight)
        confidence_m = max(0.1, min(1.0, event.cross_source_confidence))

        raw = event.baseline_severity * asset_m * tech_m * identity_m * confidence_m

        span = cfg.max_severity - cfg.min_severity
        decayed = 1.0 - 0.5 ** (raw / cfg.decay_half_life)
        dynamic = int(round(cfg.min_severity + span * decayed))
        dynamic = max(cfg.min_severity, min(cfg.max_severity, dynamic))

        decision = {
            "event_id": event.event_id,
            "baseline": event.baseline_severity,
            "dynamic_score": dynamic,
            "weights": {
                "asset": asset_m,
                "attck": round(tech_m, 3),
                "identity": round(identity_m, 3),
                "confidence": round(confidence_m, 3),
            },
            "raw_product": round(raw, 3),
        }
        log.info("severity_calculated", **decision)
        return decision

Step 4 — Run it end to end

if __name__ == "__main__":
    structlog.configure(processors=[structlog.processors.JSONRenderer()])
    engine = DynamicSeverityEngine()

    event = NormalizedEvent(
        event_id="evt-9f8a2c",
        baseline_severity=3,
        asset_criticality=AssetCriticality.CRITICAL,
        attck_techniques=["T1078", "T1059.001"],
        identity_risk_score=0.75,
        cross_source_confidence=0.92,
    )
    result = engine.score(event)
    print(f"Final dynamic severity: {result['dynamic_score']}/10")

A baseline-3 event on a critical asset, mapped to credential-access TTPs with a high-risk principal, escalates to an 8 or 9 — while the same baseline on a sandbox host with no technique mapping stays at 1 or 2. The arithmetic is bounded, repeatable, and fully reconstructable from the decision record.

Schema & Validation Integration

The scorer consumes the canonical event model rather than raw vendor JSON, which is why field naming must align with the site’s normalization model before scoring runs. Map incoming fields to Elastic Common Schema where possible — event.severity for the baseline, host.name plus an enrichment-supplied tier for asset criticality, user.name for the principal keying identity risk, and threat.technique.id for ATT&CK techniques (see the official ECS field reference). Anchoring to ECS keeps the scorer portable across the JSON event normalization patterns defined in the architecture taxonomy.

Validation is enforced by the Pydantic model: an out-of-range baseline_severity, a confidence outside 0.0–1.0, or a missing event_id raises a ValidationError at the boundary. Treat that exception as a routing signal, not a crash — the next section sends it to the dead-letter path so forensic integrity is preserved.

Error Handling & DLQ Routing

Scoring failures fall into three classes, each with a distinct error code and recovery path. Codes follow the ERR_CATEGORY_NNN convention used across the site.

from pydantic import ValidationError


def score_or_dlq(engine: DynamicSeverityEngine, raw: dict, dlq) -> dict | None:
    try:
        event = NormalizedEvent.model_validate(raw)
    except ValidationError as exc:
        dlq.publish({"error_code": "ERR_SCHEMA_001",
                     "event": raw,
                     "detail": exc.errors()})
        log.warning("event_dlq", error_code="ERR_SCHEMA_001",
                    event_id=raw.get("event_id"))
        return None
    return engine.score(event)

The non-fatal classes (ERR_SCORING_002, ERR_SCORING_003) never block the pipeline — they degrade gracefully and emit telemetry so gaps surface as metrics rather than missed detections. Persistent DLQ growth is the canary: it usually means upstream schema drift, which should trip the ingestion layer’s error categorization frameworks.

Performance Tuning

The scoring core is pure CPU and allocates almost nothing, so the practical ceiling is event throughput, not scoring cost. On a single modern core the score() call runs in low single-digit microseconds; the dominant overhead is Pydantic validation, which you can amortize.

• Batch validation. Validate in chunks of 500–2,000 events per consumer poll. Pydantic v2’s Rust core vectorizes well, and larger batches cut per-event Python overhead without inflating tail latency.
• Cache the weight lookup. technique_weights is read-only at steady state; hold it in a plain dict and reload atomically on config change rather than per event.
• Worker pool sizing. Run one consumer process per core with an async poll loop; scoring is synchronous so there is no benefit to threads inside a process (the GIL would serialize them). Scale horizontally by partition.
• Memory ceiling. Keep no per-event state in the scorer. A 2,000-event batch of NormalizedEvent instances is well under 10 MB; set the container limit to 256 MB and alert at 70%.
• Latency target. End-to-end scoring (validate + compute + emit) should hold under 2 ms p99 per event at 28k EPS across the partition set. If you exceed it, the cause is almost always enrichment latency leaking into the synchronous path — move it back to the enrichment gate.

Verification & Observability

Trust in a scoring engine comes from being able to replay any decision. Emit the full decision record (Step 3) as a structured log line and assert on its shape in tests.

def test_critical_asset_escalates() -> None:
    engine = DynamicSeverityEngine()
    high = engine.score(NormalizedEvent(
        event_id="t1", baseline_severity=3,
        asset_criticality=AssetCriticality.CRITICAL,
        attck_techniques=["T1003"], identity_risk_score=0.8,
        cross_source_confidence=0.95,
    ))
    low = engine.score(NormalizedEvent(
        event_id="t2", baseline_severity=3,
        asset_criticality=AssetCriticality.SANDBOX,
    ))
    assert high["dynamic_score"] >= 8
    assert low["dynamic_score"] <= 2
    assert high["dynamic_score"] > low["dynamic_score"]

Operational signals to dashboard:

• Score distribution — a histogram of dynamic_score per hour. A bimodal shape (clear lows and highs) is healthy; a fat middle band means weights need tuning via the threshold tuning strategies that govern actionable cutoffs.
• DLQ rate — ERR_SCHEMA_001 events per minute; a step change signals upstream schema drift.
• Default-weight hit rate — frequency of ERR_SCORING_003; rising values mean the technique table is stale.
• Decision replay — index the JSON decision records so any escalation can be reconstructed during incident review.

Troubleshooting

• Every event scores 9–10 (“runaway scores”). The decay half-life is too large or a weight is miscalibrated. Lower decay_half_life or cap individual multipliers; inspect raw_product in the decision record to find the dominant factor.
• Critical-asset alerts score low. Asset criticality is not resolving — the enrichment gate is returning LOW by default. Confirm CMDB lookups complete before scoring and watch the enrichment_incomplete tag.
• Identical events get different scores across replicas. Non-determinism almost always comes from floating-point ordering or a config skew between pods. Pin ScoringConfig in version control and roll it out atomically; the math itself is deterministic.
• Multi-source incidents still emit duplicates. Scoring is stateless by design — deduplication and convergence belong upstream in cross-source event linking, which resolves entities to a single principal before the event reaches the scorer.
• Scores never change after tuning. The service is holding a stale config in memory. Verify the hot-reload path actually swaps the technique_weights dict and emits a config-version metric on reload.

FAQ

Related

• Alert Correlation & Rule Engines — parent architecture
• Implementing weighted severity scoring for alerts
• Threshold Tuning Strategies
• MITRE ATT&CK Integration
• Cross-Source Event Linking