Case Enrichment Pipelines: Concurrent, Fail-Open SOC Context

An alert without context is a question; an enriched case is an answer. The difference between an analyst spending fifteen minutes pivoting across five consoles and opening a case that already knows the source IP’s threat-intel verdict, the host’s business owner, and the user’s recent login history is the enrichment pipeline. This page is part of the broader SOAR playbook & response automation discipline, and it builds the stage that decorates a case with the context every downstream gate depends on — routing needs asset criticality, the decision gate needs a threat verdict, ticketing needs a human-readable summary. The defining constraint is that enrichment must be fail-open: a slow or broken intel feed may never block a case from being routed and acted upon, because a delayed critical response is far more dangerous than a case that is merely under-enriched.

Problem Framing

Consider a concrete failure scenario. A case fires on a suspicious outbound connection and enters enrichment, which calls five sources in sequence: a threat-intel platform, a GeoIP service, the CMDB, the identity provider, and a sandbox detonation API. The sandbox call routinely takes 30 seconds, and today the threat-intel platform is timing out entirely. Because the calls are sequential and un-bounded, the case sits in enrichment for over a minute, and when the intel feed hangs, the whole case stalls behind it — so a genuine ransomware indicator waits, un-routed, while the pipeline blocks on a single unresponsive dependency. Worse, every identical case re-queries the same five sources, hammering the intel platform’s rate limit and making the outage worse.

The job of an enrichment pipeline is to make that scenario safe: fan the sources out concurrently, bound each with its own timeout, cache by indicator so a storm queries each source once, and treat every enrichment as advisory so a missing or slow source degrades the case’s context rather than blocking its progress. That requires three things working together: concurrent fan-out with per-source timeouts, an indicator-keyed cache, and a fail-open merge that attaches whatever returned in time. The rest of this page builds exactly that.

Prerequisites & Environment

The reference implementation targets Python 3.11+ (it uses asyncio.TaskGroup semantics and modern typing). Dependencies are aiohttp for concurrent source calls and pydantic for the case contract; the rest is standard library.

python3 -m venv .venv
source .venv/bin/activate
pip install "aiohttp>=3.9,<4.0" "pydantic>=2.6,<3.0"

Infrastructure assumptions:

• A set of enrichment sources reachable over HTTP: a threat-intel API, an asset/CMDB API, an identity directory, and optionally a sandbox. Each is modeled as an injectable async callable so the pipeline is testable against fixtures.
• A shared cache (Redis or an in-process TTL cache) keyed by indicator value, so repeated indicators within a window hit the cache rather than the upstream source.
• A dead-letter queue is not used to block cases here; instead, per-source failures are recorded on the case as partial results. The DLQ is reserved only for cases that fail their own typed contract.

Architecture Overview

Enrichment is a fan-out/fan-in stage. The case’s indicators are dispatched to all sources concurrently, each guarded by a timeout; results that return in time are merged onto the case, and sources that time out or error are recorded as partial-result markers without blocking the merge.

The pivotal design decision is that the merge never waits for the slowest source past its timeout. Each source has its own deadline sized to its normal latency, and the merge collects results as they arrive; a source that misses its deadline contributes a recorded gap, not a stall. This is what makes enrichment fail-open rather than fail-stuck.

Step-by-Step Implementation

Step 1 — Model the case and its enrichment result

The case carries indicators; enrichment appends an advisory context block plus a per-source coverage record so downstream gates know what is known and what is missing.

from __future__ import annotations

from pydantic import BaseModel, Field


class EnrichableCase(BaseModel):
    case_id: str = Field(min_length=1)
    src_ip: str | None = None
    host_id: str | None = None
    user_name: str | None = None
    file_hash: str | None = None


class Enrichment(BaseModel):
    """Advisory context; never blocks a downstream gate."""

    threat_verdict: str | None = None      # "malicious" | "suspicious" | "clean"
    owner_team: str | None = None
    criticality: int | None = None
    recent_auth_country: str | None = None
    coverage: dict[str, str] = Field(default_factory=dict)   # source -> ok|timeout|error

Step 2 — Wrap each source with a per-source timeout and cache

A single helper enforces the deadline and records coverage. It returns a partial marker on timeout or error rather than raising, so one bad source cannot abort the fan-out.

import asyncio
import logging
from typing import Awaitable, Callable, Optional

logger = logging.getLogger("soc.soar.enrich")

SourceFn = Callable[[EnrichableCase], Awaitable[dict]]


async def guarded_source(
    name: str, fn: SourceFn, case: EnrichableCase, timeout: float,
    cache: dict[str, dict],
) -> tuple[str, str, dict]:
    """Return (source_name, status, data); never raises."""
    cache_key = f"{name}:{case.src_ip or case.file_hash or case.host_id or ''}"
    if cache_key in cache:
        return name, "ok", cache[cache_key]
    try:
        data = await asyncio.wait_for(fn(case), timeout)
        cache[cache_key] = data
        return name, "ok", data
    except asyncio.TimeoutError:
        logger.warning("ERR_ENRICH_004 source=%s timeout case=%s", name, case.case_id)
        return name, "timeout", {}
    except Exception as exc:                     # ERR_ENRICH_005
        logger.warning("ERR_ENRICH_005 source=%s err=%s", name, exc)
        return name, "error", {}

Step 3 — Fan out concurrently and merge fail-open

All sources run at once under asyncio.gather; the merge folds whatever returned into the Enrichment, recording coverage for the rest.

async def enrich(
    case: EnrichableCase,
    sources: dict[str, tuple[SourceFn, float]],
    cache: dict[str, dict],
) -> Enrichment:
    results = await asyncio.gather(
        *(guarded_source(name, fn, case, timeout, cache)
          for name, (fn, timeout) in sources.items())
    )
    enr = Enrichment()
    for name, status, data in results:
        enr.coverage[name] = status
        if status != "ok":
            continue
        if name == "intel":
            enr.threat_verdict = data.get("verdict")
        elif name == "asset":
            enr.owner_team = data.get("owner_team")
            enr.criticality = data.get("criticality")
        elif name == "identity":
            enr.recent_auth_country = data.get("country")
    return enr

Step 4 — Attach enrichment and proceed regardless of coverage

The case advances even with partial coverage; a low-coverage case is flagged for an analyst but never blocked. The decision gate reads threat_verdict if present and falls back to the base severity if not.

def coverage_ratio(enr: Enrichment) -> float:
    if not enr.coverage:
        return 0.0
    ok = sum(1 for v in enr.coverage.values() if v == "ok")
    return ok / len(enr.coverage)


def should_flag_low_context(enr: Enrichment, floor: float = 0.5) -> bool:
    # Advisory only: flags for analyst attention, never blocks routing.
    return coverage_ratio(enr) < floor

Schema & Validation Integration

Enrichment reads the case’s ECS-aligned indicators (source.ip, host.id, user.name, file.hash) — normalized upstream by JSON event normalization — and writes to the ECS enrichment namespace (threat.indicator.*, host.risk.score, related.user). Crucially, enrichment fields are the enrichment tier of the case schema: advisory, additive, and never a gate condition, exactly as the SOAR response-automation schema model prescribes. The threat-intel verdict itself comes from feeds mapped through the threat intel feed mapping guides, so the indicator formats the enrichment source returns are already normalized to the site’s schema. The coverage record is the validation signal: persistently low coverage from one source is drift or an outage to triage, not a reason to hold cases.

Error Handling & DLQ Routing

Enrichment failures are recorded, not fatal. Codes follow the ERR_CATEGORY_NNN convention, but note the routing column — almost every code degrades context rather than dead-lettering the case.

The cardinal rule is enrichment is advisory, so fail open. The only case that dead-letters (ERR_ENRICH_011) is one with no usable indicators at all — there is nothing to enrich. Every other failure records a coverage gap and lets the case proceed, because holding a case hostage to a slow intel feed is the failure mode enrichment exists to prevent. Repeated source errors trip a per-source circuit breaker so a dead feed is skipped entirely rather than timing out on every case.

Performance Tuning

Enrichment latency is the max of the per-source timeouts, not their sum — that is the whole point of the fan-out. Tune in this order:

• Size each timeout to the source’s real p95, not a global default. A 1-second CMDB and a 5-second sandbox should not share a timeout; the case’s enrichment latency is the slowest source you actually wait for.
• Cache aggressively by indicator. A storm of cases sharing one malicious IP should query the intel feed once; an indicator-keyed cache with a minutes-long TTL collapses that to a single upstream call and is the largest efficiency win during an incident.
• Circuit-break dead sources so a failing feed is skipped after N consecutive failures rather than costing every case its full timeout.
• Bound total concurrency across cases with a semaphore per source so enrichment respects each vendor’s rate limit, pairing with async log batching principles on the fan-out side.
• Make the sandbox optional and asynchronous — a 5-second detonation should not gate routing; fetch it in the background and attach it to the case when it lands, updating the ticket via automated ticket creation.

Verification & Observability

Confirm correct operation with a coverage metric per source, latency histograms, and tests that assert fail-open behavior when a source hangs.

async def test_enrichment_is_fail_open_on_timeout() -> None:
    async def good_asset(case: EnrichableCase) -> dict:
        return {"owner_team": "fin-ir", "criticality": 3}

    async def hanging_intel(case: EnrichableCase) -> dict:
        await asyncio.sleep(10)          # far exceeds its timeout
        return {"verdict": "malicious"}

    case = EnrichableCase(case_id="C-3", src_ip="203.0.113.5")
    sources = {
        "intel": (hanging_intel, 0.05),   # will time out
        "asset": (good_asset, 1.0),
    }
    enr = await enrich(case, sources, cache={})

    assert enr.coverage["intel"] == "timeout"   # recorded, not raised
    assert enr.owner_team == "fin-ir"           # asset still enriched
    assert enr.threat_verdict is None           # intel gap, case proceeds anyway

Operationally, emit enrichment_source_total{source,status}, enrichment_coverage_ratio (histogram), enrichment_latency_seconds{source}, and enrichment_cache_hit_ratio. A dropping enrichment_coverage_ratio with one source dominating the status="timeout" label is the canary for a degrading feed, and it should trigger a source-level circuit breaker long before it affects case throughput.

Troubleshooting

• Cases stall in enrichment. Root cause: sources are called sequentially or share one oversized timeout. Fix: fan out concurrently and give each source its own deadline sized to its p95.
• An intel outage stops all cases. Root cause: enrichment treats a source failure as fatal instead of advisory. Fix: record the failure as a coverage gap and let the case proceed; only a case with no indicators should ever block.
• A feed’s rate limit is exhausted during a storm. Root cause: identical indicators are queried per case with no cache. Fix: cache by indicator value with a minutes-long TTL so repeats hit the cache.
• Enriched context is stale. Root cause: cache TTL is too long for volatile data like on-call or active-session state. Fix: use a short TTL for volatile fields and a longer one for stable reputation data.
• A malformed source response poisons the case. Root cause: source output is merged without validation. Fix: validate each source’s payload and discard a malformed contribution as ERR_ENRICH_031 while keeping the rest.

FAQ

Related

• SOAR Playbook & Response Automation
• Playbook Orchestration Patterns
• Alert Routing Workflows
• Automated Ticket Creation
• Threat Intel Feed Mapping