Why multiply the severity factors instead of adding weighted terms?

Multiplication makes the factors compound the way real risk does — a credential-dumping technique on a tier-0 asset with a failed posture reinforces rather than merely sums. Addition lets one large term dominate and flattens context. The unbounded growth that multiplication introduces is contained by a logistic clamp that saturates the score toward the ceiling instead of overflowing.

Where does weighted severity scoring sit relative to correlation and routing?

Downstream of normalization, enrichment, and correlation, and upstream of routing and SOAR dispatch. All enrichment such as asset tier, identity posture, and ATT&CK mapping must be resolved before scoring because scoring is synchronous and CPU-bound. The correlated-events count comes from the cross-source linking layer, not from the scorer.

How do I keep weighted scoring weight tables honest over time?

Feed analyst disposition back into the weights through periodic Bayesian updating rather than ad-hoc edits, and version every change. Entities that score high but never confirm as incidents get an explicit dampener via asset-tier exclusions, not raw signature blocks, so the audit trail stays intact.

Implementing Weighted Severity Scoring for Al…

Flat, vendor-supplied severity tags cannot tell a failed login on a sandbox runner apart from the same event on a domain controller, so analysts drown in undifferentiated mediums while real intrusions sit unranked. This guide implements the exact multiplier math that fixes that, as one precise stage of dynamic severity scoring within the broader Alert Correlation & Rule Engines pipeline.

Root-Cause Context

Weighted scoring problems surface because severity is decided at the wrong layer. SIEM detection content and raw CVSS both assign severity before the event is enriched, so the number that reaches triage reflects the signature, not the situation. A single alert carries the same raw_severity whether the principal is a service account or a tier-0 admin, whether the host is ephemeral or a production database, and whether one or three telemetry sources agree.

Three forces make this worse in production SOC pipelines. First, format drift: every detection vendor emits severity on its own scale (1–4, 1–10, Low/Medium/High), so any naive merge mixes incomparable units. Second, context starvation: the enrichment that would contextualize the score — asset tier, identity posture, MITRE ATT&CK technique mapping — is resolved downstream of where severity was set, and nothing re-computes it. Third, single-source bias: the same compromised entity appearing across EDR, IAM, and network telemetry should be one high-confidence incident, but without cross-source event linking feeding a convergence factor, the model emits three independent mediums.

Weighted severity scoring resolves all three by re-deriving a single bounded score at the correlation layer from a lightweight baseline and a set of deterministic, auditable multipliers. The math must be explainable (every escalation produces a decision record), bounded (no overflow during a log storm), and stateless (so it scales by stream partition).

Prerequisites

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 a heavy framework.

python3 -m venv .venv
source .venv/bin/activate
pip install "pydantic>=2.6"

Assumptions: Python 3.11+ (for StrEnum and faster startup); upstream enrichment has already attached asset tier, identity posture, and any ATT&CK technique IDs to the event; and weight tables live in version control as config rather than being hardcoded per detection.

Production-Ready Implementation

The module below is self-contained. It validates the enriched event at the boundary with Pydantic, computes the four contextual multipliers, passes the product through a logistic clamp so scores saturate toward the ceiling instead of overflowing, and exposes an async score_stream so the same core drops into an asyncio consumer. Validation failures raise a typed error that the caller routes to a dead-letter path rather than crashing the worker.

from __future__ import annotations

import asyncio
import math
from enum import StrEnum
from typing import AsyncIterator, Iterable

from pydantic import BaseModel, Field, ValidationError, field_validator


class AssetTier(StrEnum):
    CRITICAL = "critical"
    STANDARD = "standard"
    EPHEMERAL = "ephemeral"
    SANDBOX = "sandbox"


class Posture(StrEnum):
    VERIFIED = "verified"
    DEGRADED = "degraded"
    FAILED = "failed"


class ScoringError(Exception):
    """Carries a machine-readable code for DLQ routing."""

    def __init__(self, code: str, detail: str) -> None:
        super().__init__(f"{code}: {detail}")
        self.code = code
        self.detail = detail


class EnrichedAlert(BaseModel):
    alert_id: str
    raw_severity: float = Field(ge=0.0, le=10.0)
    asset_tier: AssetTier
    posture: Posture
    entity_id: str
    mitre_technique: str | None = None
    source_systems: list[str] = Field(default_factory=list)
    correlated_events: int = Field(default=1, ge=1)

    @field_validator("mitre_technique")
    @classmethod
    def _normalize_technique(cls, v: str | None) -> str | None:
        return v.upper().strip() if v else None


class ScoringConfig(BaseModel):
    asset_multipliers: dict[AssetTier, float] = {
        AssetTier.CRITICAL: 3.0,
        AssetTier.STANDARD: 1.0,
        AssetTier.EPHEMERAL: 0.5,
        AssetTier.SANDBOX: 0.3,
    }
    posture_multipliers: dict[Posture, float] = {
        Posture.VERIFIED: 1.0,
        Posture.DEGRADED: 1.8,
        Posture.FAILED: 2.5,
    }
    # ATT&CK technique -> intrinsic baseline (ID + name documented inline below)
    mitre_baselines: dict[str, float] = {
        "T1003": 9.0,  # OS Credential Dumping
        "T1566": 8.5,  # Phishing
        "T1059": 8.0,  # Command and Scripting Interpreter
        "T1071": 7.5,  # Application Layer Protocol
        "T1595": 4.0,  # Active Scanning
    }
    default_mitre_baseline: float = 5.0
    convergence_bonus: float = 2.0   # per additional correlated source
    convergence_cap: float = 3.0     # storm guard on the convergence factor
    ceiling: float = 10.0


class ScoreDecision(BaseModel):
    alert_id: str
    final_score: float
    asset_mult: float
    posture_mult: float
    mitre_weight: float
    convergence_factor: float


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

    def score(self, alert: EnrichedAlert) -> ScoreDecision:
        cfg = self.config

        # Dimension 1 — asset criticality.
        asset_mult = cfg.asset_multipliers[alert.asset_tier]

        # Dimension 2 — identity / device posture.
        posture_mult = cfg.posture_multipliers[alert.posture]

        # Dimension 3 — ATT&CK baseline overrides a weak vendor severity.
        if alert.mitre_technique:
            family = alert.mitre_technique.split(".")[0]
            baseline = cfg.mitre_baselines.get(family, cfg.default_mitre_baseline)
            mitre_weight = max(alert.raw_severity, baseline)
        else:
            mitre_weight = max(alert.raw_severity, 1.0)

        # Dimension 4 — cross-source convergence, capped to survive log storms.
        extra = alert.correlated_events - 1
        convergence_factor = min(
            1.0 + extra * (cfg.convergence_bonus / 10.0),
            cfg.convergence_cap,
        )

        product = mitre_weight * asset_mult * posture_mult * convergence_factor

        # Logistic clamp: saturate toward the ceiling instead of overflowing.
        saturated = cfg.ceiling * (2.0 / (1.0 + math.exp(-product / cfg.ceiling)) - 1.0)
        final = round(max(1.0, min(saturated, cfg.ceiling)), 1)

        return ScoreDecision(
            alert_id=alert.alert_id,
            final_score=final,
            asset_mult=asset_mult,
            posture_mult=posture_mult,
            mitre_weight=mitre_weight,
            convergence_factor=round(convergence_factor, 2),
        )

    def score_payload(self, payload: dict) -> ScoreDecision:
        try:
            alert = EnrichedAlert.model_validate(payload)
        except ValidationError as exc:
            raise ScoringError("ERR_SCORE_001", f"schema validation failed: {exc.errors()}")
        try:
            return self.score(alert)
        except KeyError as exc:
            raise ScoringError("ERR_SCORE_004", f"unknown enum value: {exc}")

    async def score_stream(
        self, payloads: Iterable[dict]
    ) -> AsyncIterator[ScoreDecision | ScoringError]:
        for payload in payloads:
            try:
                yield self.score_payload(payload)
            except ScoringError as err:
                yield err  # caller routes to the DLQ
            await asyncio.sleep(0)  # cooperative yield for the event loop


async def _demo() -> None:
    scorer = WeightedScorer()
    events = [
        {"alert_id": "EVT-9921", "raw_severity": 4.0, "asset_tier": "critical",
         "posture": "failed", "entity_id": "DC-PROD-01",
         "mitre_technique": "T1003", "source_systems": ["EDR", "IAM"],
         "correlated_events": 2},
        {"alert_id": "EVT-9922", "raw_severity": 4.0, "asset_tier": "sandbox",
         "posture": "verified", "entity_id": "ci-runner-7",
         "mitre_technique": "T1595", "correlated_events": 1},
        {"alert_id": "EVT-BAD", "raw_severity": 99, "asset_tier": "critical",
         "posture": "verified", "entity_id": "x"},  # invalid -> ERR_SCORE_001
    ]
    async for result in scorer.score_stream(events):
        if isinstance(result, ScoringError):
            print(f"DLQ <- {result.code}: {result.detail[:60]}")
        else:
            tier = "CRITICAL" if result.final_score >= 8.5 else (
                "HIGH" if result.final_score >= 6.0 else "MEDIUM")
            print(f"{result.alert_id}: {result.final_score} ({tier})")


if __name__ == "__main__":
    asyncio.run(_demo())

The ATT&CK references above are by ID and name: T1003 OS Credential Dumping, T1566 Phishing, T1059 Command and Scripting Interpreter, T1071 Application Layer Protocol, and T1595 Active Scanning. Splitting on . collapses sub-techniques (T1059.001) onto their parent family so a missing sub-technique weight never silently falls through to the default.

Error-Code Reference

These codes follow the established ERR_CATEGORY_NNN convention and are what the scoring stage raises; route each to the action shown rather than letting it crash the consumer.

Code	Meaning	Action
`ERR_SCORE_001`	Payload failed Pydantic validation (missing field, out-of-range `raw_severity`)	Route raw event to forensic DLQ; do not score
`ERR_SCORE_002`	`correlated_events` < 1 or non-integer from the linking layer	Clamp to 1, score, and flag the correlation source
`ERR_SCORE_003`	ATT&CK technique present but absent from `mitre_baselines`	Apply `default_mitre_baseline`; emit metric for weight-table gap
`ERR_SCORE_004`	Unknown `asset_tier` / `posture` enum value (schema drift)	DLQ and alert config owners; treat as ingestion break
`ERR_SCORE_005`	Computed product overflowed before clamp (config error)	Reject config at load time; logistic clamp prevents at runtime

ERR_SCORE_003 is intentionally a soft failure — an unmapped technique should still score, just without the override — whereas ERR_SCORE_001 and ERR_SCORE_004 are hard failures because they mean the upstream contract broke.

Operational Notes

The scoring core is pure CPU and allocates one small ScoreDecision per event; on a single modern core it sustains well over 100k events/sec, so the realistic ceiling is your serialization and I/O, not the math. Keep the engine stateless and shard by stream partition (entity ID or source) for linear horizontal scale.

A few field-tested quirks. Vendor severity scales are not interchangeable: normalize every source onto the 0–10 baseline at ingestion or the asset multiplier silently amplifies a vendor’s inflated “5”. Cap the convergence factor (convergence_cap) — during a mass-patching window or an outage, one entity can legitimately trip dozens of correlated alerts, and an uncapped bonus turns that into a synthetic critical flood; this is the same storm risk that rate limiting strategies absorb at the gateway. Treat the weight tables as config, not code: load ScoringConfig from version-controlled YAML/JSON so changes are reviewed and auditable, and validate multiplier ranges in CI before merge. Finally, the actionable threshold (the 8.5 in the demo) belongs to threshold tuning strategies and should ride on rolling baselines, not a hardcoded constant.

Verification Checklist

Identical raw events on critical vs sandbox assets produce clearly separated scores (the asset multiplier is actually applied).
A high-weight technique (T1003 OS Credential Dumping) overrides a low raw_severity, while T1595 Active Scanning does not inflate it.
correlated_events=10 does not exceed the convergence_cap, and no single score ever exceeds ceiling (10.0).
A malformed payload yields ERR_SCORE_001 routed to the DLQ rather than an uncaught exception in the consumer.
An unmapped technique yields default_mitre_baseline and an ERR_SCORE_003 metric, not a crash.
Replaying 30–90 days of historical alerts keeps fewer than ~5% above the actionable threshold (no score inflation).

FAQ

Why multiply the factors instead of adding weighted terms?

Multiplication makes the factors interact the way risk actually does: a credential-dumping technique on a tier-0 asset with a failed posture should compound, not merely sum. Addition lets a single large term dominate and flattens context. The trade-off is unbounded growth, which is exactly why the product passes through a logistic clamp that saturates toward the ceiling instead of overflowing.

Where does this stage sit relative to correlation and routing?

Downstream of normalization, enrichment, and correlation; upstream of routing and SOAR dispatch. All network-dependent enrichment (CMDB asset tier, identity posture, ATT&CK mapping) must be resolved before the event reaches the scorer, because scoring is synchronous and CPU-bound. The correlated_events count is supplied by the cross-source linking layer, not computed here.

How do I keep the weight tables honest over time?

Feed analyst disposition (true/false positive) back into the weights through periodic Bayesian updating rather than ad-hoc edits, and version every change. Entities that consistently score high but never confirm as incidents get an explicit dampener via asset-tier exclusions, not a raw signature block, so the audit trail stays intact.

Dynamic Severity Scoring — parent guide
Alert Correlation & Rule Engines — parent architecture
Cross-Source Event Linking
MITRE ATT&CK Integration
Threshold Tuning Strategies

Implementing Weighted Severity Scoring for Alerts