soc_ai/agent/hunt_gates.py
Rama main · réplica de lectura del repositorio original.
"""Deterministic post-hunt citation gate for HuntReport findings.
Investigations validate citations deterministically (soc_ai.agent.gates); hunt
findings did not — a finding could cite an ES ``_id`` the hunt never actually
pulled. This module closes that gap: after the agent lands a
:class:`~soc_ai.agent.hunt.HuntReport`, each finding's citations are resolved
against the evidence the hunt ACTUALLY gathered (the ``tool_result`` payloads
that streamed this run). A finding whose citations resolve to nothing has its
non-resolving citations stripped and its severity capped; a high/critical
finding that cites nothing at all is capped too. Finding and chart ``title``s
lose their surrounding whitespace (:func:`_clean_title`) and are never cut. A
long title is stored whole and the console wraps it. No LLM
calls — this is the trust layer, one layer down from the investigation citation
gate.
Resolver choice — two-tier (M2, 2026-08-25 audit). Hunt citations are
overwhelmingly bare ES ``_id`` strings, and an id-shaped citation is a claim
about RETRIEVED EVIDENCE — so it resolves ONLY by membership in the set of
document identifiers and decisive typed values the hunt actually retrieved
(``_id``/``uid``/``sid`` plus hash/JA3/rule-metadata evidence-key leaves,
collected from the real STRUCTURE of the gathered payloads, see
:func:`_gathered_evidence_tokens`). The previous substring form was forgeable:
a fabricated id-shaped token planted in attacker-controllable field content (a
DNS query name, TLS SNI, URI, User-Agent) of a genuinely-retrieved doc
resolved and defeated the whole gate. Non-id citations (values, IPs, domains,
free text) keep the distinctive-token machinery (``_FUZZY_TOKEN_RE`` +
``_CITATION_STOP_WORDS``) from :mod:`soc_ai.agent.gates` against the JSON dump
of the gathered tool-result payloads — the fallback the E1.3 spec explicitly
sanctions for value-shaped citations.
"""
from __future__ import annotations
import ipaddress
import json
import logging
import re
from datetime import UTC, datetime, timedelta
from typing import Any
from soc_ai.agent.evidence import _classify_citation, _collect_evidence_values
from soc_ai.agent.gates import _CITATION_STOP_WORDS, _FUZZY_TOKEN_RE
from soc_ai.agent.narrative_grounding import (
claims_no_host_telemetry,
coverage_replacer,
coverage_subject,
rewrite_host_gap_claims,
)
from soc_ai.dossier.coverage import (
HostCoverage,
HostIdentity,
host_coverage,
host_identity,
plane_phrase,
)
from soc_ai.dossier.coverage import describe as describe_coverage
from soc_ai.so_client import fields
from soc_ai.tools._synth_scope import SynthScope
_LOGGER = logging.getLogger(__name__)
def _is_address(value: str) -> bool:
try:
ipaddress.ip_address(value)
except ValueError:
return False
return True
# Severity ordinal — "cap at X" == min(current, X). Only ever LOWERS a severity.
_SEV_ORDER: tuple[str, ...] = ("info", "low", "medium", "high", "critical")
_SEV_RANK: dict[str, int] = {s: i for i, s in enumerate(_SEV_ORDER)}
_UNRESOLVED_NOTE = (
"The citations did not resolve to gathered evidence, so soc-ai lowered the severity to low."
)
_HIGH_NO_CITE_NOTE = (
"This high-severity finding has no citations, so soc-ai lowered the severity to medium."
)
# Three sentences in the analyst's terms (dogfood 2026-10-06 C5). The stored
# rows of older builds carry two other phrasings, and the hunt page rewords
# both (soc_ai.hunting.hunt_words.validator_note).
_ALERT_ONLY_NOTE = (
"Severity held at medium. The finding cites alerts only. An investigation can raise it."
)
# ── Corroboration gate: alert-query vs corroborating-evidence tools ───────────
# The trust-erosion failure this closes (two real prod hunts, 2026-07): a
# wall-clock-truncated hunt gets written up from a transcript DOMINATED by
# Suricata alert documents, and the synthesizer asserts compromise at HIGH
# severity citing the very alert that IS the claim. The E1.3 gate above passes
# such a finding because the cited id genuinely EXISTS in the transcript — it
# checks existence, not corroboration. This gate adds the missing check: a
# high/critical THREAT finding must cite at least one item that resolves into a
# NON-alert (corroborating) tool result, or its severity is capped.
#
# ``_ALERT_QUERY_TOOLS`` is a DENY-set — the tools whose results ARE the
# detector's claim (Suricata alert documents, rule text, rule base-rates), which
# therefore CANNOT corroborate a threat asserted from those same alerts:
# * t_query_events_oql — the primary lens; returns raw event docs INCLUDING
# suricata.alert documents (the loud alert titles that dominated the
# truncated prod transcripts). Its results are partitioned per-document by
# ``_oql_telemetry_docs`` BEFORE the membership test below — docs positively
# identified as non-alert telemetry corroborate; alert docs and
# unidentifiable docs do not (2026-07-20 telemetry-latitude design).
# * t_query_detections — searches detection RULES (the claim source itself).
# * t_get_rule_content — returns the signature's own definition (the claim).
# * t_rule_prevalence — a rule's base-rate/noisiness (rule metadata).
# * t_suggest_rule_tuning — rule disposition trend (rule metadata).
# ``t_get_event_raw`` is NOT a blanket corroborator: it fetches a single event by
# _id, which may be the underlying flow record OR the very alert doc that raised
# the claim. It is partitioned by ``event.dataset``/``event.kind`` (same test as
# ``t_query_events_oql``) — a fetched telemetry doc corroborates, a refetched
# suricata.alert doc does not.
# Every OTHER tool is corroborating evidence BEYOND the alert document —
# t_get_pcap / t_decode_payload (decoded packets), t_enrich_* (blocklist/MISP/
# ASN), t_prevalence (host novelty), t_host_summary (host artifact/OS),
# t_query_zeek_logs (the underlying flow records), the online
# quartet (greynoise/shodan/cve) and the web tools. A finding grounded in ANY of
# those has looked past the alert title, which is exactly what the gate requires.
_ALERT_QUERY_TOOLS: frozenset[str] = frozenset(
{
"t_query_events_oql",
"t_query_detections",
"t_get_rule_content",
"t_rule_prevalence",
"t_suggest_rule_tuning",
}
)
# Canonical non-threat finding categories. The gate must treat ANY other category
# value the way the read path does (routes_hunts._finding_category maps anything
# outside the canonical triple back to "threat" for display): a non-canonical
# spelling ("suspicious activity") must NOT dodge the corroboration cap while still
# rendering the red threat badge. Only these two are genuinely non-threat.
_NON_THREAT_CATEGORIES: frozenset[str] = frozenset({"visibility_gap", "observation"})
# Chart budget — a model that emits plausible-but-uncited charts freely is the whole
# risk, so beyond this ceiling extras are dropped even if they'd otherwise resolve.
_MAX_CHARTS = 4
# Title cleanup. The HuntReport schema and the prompt ask for short headlines,
# and the console wraps a long one. A title was once cut at 90 characters with
# an ellipsis, and later at 500. The full text was then nowhere, because the
# cut happened before the store. The owner banned the ellipsis cut
# (2026-10-07), so the title is stored whole.
def _clean_title(title: str) -> str:
"""``title`` with the surrounding whitespace removed, never cut."""
return (title or "").strip()
def _severity_rank(severity: str) -> int:
"""``severity``'s rank on the ladder; an unknown value ranks as ``critical``.
The one answer this module gives about a severity it does not recognise, and
it is the loudest one on purpose. Every read of the ladder here is asking a
safety question — is this claim loud enough to need a cap, and where does
the cap land — so an unrecognised value has to be treated as the loudest
thing it could be, or a malformed severity walks past the gate.
It used to be two answers in one module. :func:`_cap_severity` resolved an
unknown to ``critical``, which is the safe direction: the cap applies and
the finding is lowered to the ceiling. The two gate tests resolved it to
rank 0 — ``info`` — so a finding carrying anything the ladder does not hold
(``"sev:high"``, ``"urgent"``, ``"HIGH!"``) ranked below ``high``, skipped
the high/critical branch, and kept both its severity string and its
citations. The permissive default decided whether a cap ran at all and the
conservative one only decided where it landed, so the conservative one never
got the chance. The schema is a free-form ``str`` with the ladder in its
description, which is a request to the model, not a constraint on it.
"""
return _SEV_RANK.get((severity or "").strip().lower(), _SEV_RANK["critical"])
def _cap_severity(severity: str, ceiling: str) -> str:
"""Lower ``severity`` to at most ``ceiling`` (never raises). Unknown severities
are treated as their loudest rank so a malformed value can't dodge the cap.
"""
cap = _SEV_RANK[ceiling]
return _SEV_ORDER[min(_severity_rank(severity), cap)]
# Tool-call short-circuits the toolset returns INSTEAD of running the tool: an
# error (``_tool_error``, which echoes the rejected query ``fragment``), a
# duplicate-call stop (``_dedup_result``, which echoes the model's own ``args``)
# and the prefetch hint. None of them retrieved anything — every value they
# carry is the model's own words handed back — so they must never enter the
# citation corpus. Otherwise a finding could cite a domain the model merely
# ASKED about twice and pass both the citation gate and the corroboration gate.
_SHORT_CIRCUIT_KEYS: tuple[str, ...] = ("error", "duplicate_call", "prefetch_already_has_this")
def _is_short_circuit(result: Any) -> bool:
"""True iff ``result`` is a toolset short-circuit payload (see above).
Deliberately narrower than :func:`soc_ai.agent.evidence.tool_return_is_evidence`:
that predicate also rejects inference tools (decoded payloads corroborate a
hunt finding) and zero-hit search shapes (an aggregation-only OQL rollup is
``total=0, hits=[]`` and is exactly the measured evidence the corroboration
gate wants). Only the echo shapes are dropped here.
"""
return isinstance(result, dict) and any(result.get(k) for k in _SHORT_CIRCUIT_KEYS)
def _retrieved_results(tool_results: list[Any]) -> list[Any]:
"""``tool_results`` minus the short-circuit payloads.
Applied ONCE, up front, so the evidence text, the retrieved-id set and the
corroboration subset are all built from the same corpus. Labeled
``{tool_name, result}`` items are judged on their ``result``; a bare legacy
item is judged on itself.
"""
kept: list[Any] = []
for item in tool_results:
payload = item.get("result") if isinstance(item, dict) and "result" in item else item
if _is_short_circuit(payload):
continue
kept.append(item)
return kept
def _gathered_evidence_text(tool_results: list[Any]) -> str:
"""Lower-cased JSON dump of every gathered tool-result payload.
``tool_results`` are the ``result`` values from the run's ``tool_result``
events (the data the hunt actually pulled). Mirrors
:func:`soc_ai.agent.evidence._bundle_dump_text` — one flat, lower-cased JSON
blob the citation tokens are substring/word-boundary matched against.
"""
if not tool_results:
return ""
try:
return json.dumps(tool_results, default=str).lower()
except Exception:
return ""
def _gathered_evidence_tokens(tool_results: list[Any]) -> frozenset[str]:
"""Document ids + typed-evidence values the hunt actually RETRIEVED (lowercased).
M2 (2026-08-25 audit): collected from the real STRUCTURE of the gathered
payloads — ``_id`` leaves on ES hits, ``uid`` leaves on Zeek rows / raw
fetches (``log.id.uid``; zeek rows carry no ``_id``), ``sid``/``uuid`` rule
identifiers, plus the decisive typed-evidence leaves (``file.hash.*``,
``hash.ja3``/``ja3s``, detector rule metadata — see
:data:`soc_ai.agent.evidence._EVIDENCE_KEYS`) — via
:func:`soc_ai.agent.evidence._collect_evidence_values`.
Attacker-plantable CONTENT fields (a DNS query name, an SNI, a URI) are
never identity or typed-evidence keys, so a token planted there cannot
enter this set. Works over both labeled ``{tool_name, result}`` items and
bare legacy results (the walk only harvests evidence-keyed leaves either
way).
"""
ids: set[str] = set()
for item in tool_results:
_collect_evidence_values(item, ids)
return frozenset(ids)
def _citation_resolves(citation: str, evidence_text: str, retrieved_ids: frozenset[str]) -> bool:
"""True iff ``citation`` resolves against the gathered evidence.
Two-tier (M2): an ID-SHAPED citation (per the shared classifier —
``(id X)`` or a bare 12+-char alphanumeric) is a claim that specific
evidence was retrieved (a document, or a decisive typed value like a file
hash or JA3), so it resolves ONLY by membership in
``retrieved_ids`` — never by substring, which an attacker could satisfy by
planting the token in any content field of a doc the hunt legitimately
pulled. Every other shape keeps the investigation gate's distinctive-token
discipline (GATE C): a stop-word or a short generic fragment never resolves
a citation on its own; a long token (>= 8 chars — a hash, a full IP)
resolves on a substring match, a medium token (>= 5 chars — a domain label,
a hyphenated host) resolves only on a word boundary. This kills hollow
matches while resolving real values the hunt pulled.
"""
kind, target = _classify_citation(citation)
if kind == "id":
return target is not None and target.lower() in retrieved_ids
if not evidence_text:
return False
for tok in _FUZZY_TOKEN_RE.findall(citation):
low = tok.lower()
if low in _CITATION_STOP_WORDS:
continue
if len(tok) >= 8 and low in evidence_text:
return True
if len(tok) >= 5 and re.search(rf"\b{re.escape(low)}\b", evidence_text):
return True
return False
def _oql_telemetry_docs(result: Any) -> list[Any]:
"""The subset of an OQL result's hit docs POSITIVELY identified as telemetry.
``t_query_events_oql`` returns ``EsSearchResult.model_dump()`` —
``{total, hits: [<ES hit>], aggregations, ...}`` where each hit carries the
doc under ``_source`` (or ``fields`` with list-wrapped values). A doc is
telemetry iff its ``event.dataset`` is present and is NOT ``suricata.alert``
and its ``event.kind`` is not ``alert``. Everything else — alert docs, docs
with no dataset, aggregation-only results, malformed shapes — is excluded,
so this helper can only ever ADD corroboration for docs that are provably
not the detector's own claim. Never raises on shape surprises.
"""
if not isinstance(result, dict):
return []
hits = result.get("hits")
if not isinstance(hits, list):
return []
def _one(container: dict[str, Any], dotted: str) -> Any:
# _source form nests ({"event": {"dataset": ...}}); fields form is flat
# dotted keys with list-wrapped values ({"event.dataset": ["zeek.dns"]}).
head, _, tail = dotted.partition(".")
nested = container.get(head)
value = nested.get(tail) if isinstance(nested, dict) else container.get(dotted)
if isinstance(value, list):
return value[0] if value else None
return value
telemetry: list[Any] = []
for hit in hits:
if not isinstance(hit, dict):
continue
src = hit.get("_source")
if not isinstance(src, dict):
src = hit.get("fields")
if not isinstance(src, dict):
continue
# ECS-first with a fallback, because a plane that lives only in
# ``data_stream.dataset`` is still positively identified. Reading
# ``event.dataset`` alone dropped 632,523 documents on the development
# grid — every network_traffic plane, from the one sensor watching the
# live VLANs — as "unidentifiable", so a hunt that found real traffic
# there had its evidence discarded by this gate.
dataset = fields.dataset_of(src)
kind = _one(src, "event.kind")
if not dataset:
continue # genuinely no identity → still not corroborating
if dataset == "suricata.alert" or kind == "alert":
continue # the detector's claim cannot corroborate itself
telemetry.append(hit)
return telemetry
def _oql_aggregations(result: Any) -> Any:
"""The aggregation rollup of an AGGREGATION-ONLY OQL response, else ``None``.
A ``groupby``/``count`` OQL query runs with ``size=0`` -- its ``hits`` list is
empty and every measured value lives in ``aggregations`` (top talkers, rare
destinations, NXDOMAIN churn per host -- most of the primer's worked hunts).
:func:`_oql_telemetry_docs` only walks ``hits``, so such a response would
contribute NOTHING to corroboration even though a computed rollup over
telemetry is exactly the measured evidence that looks past a single detector
alert. Returns the ``aggregations`` payload when the response carries
aggregations and NO hits (aggregation-only); ``None`` otherwise. Never raises
on shape surprises.
"""
if not isinstance(result, dict):
return None
hits = result.get("hits")
if isinstance(hits, list) and hits:
return None # not aggregation-only -- the per-doc partition governs
aggs = result.get("aggregations")
if isinstance(aggs, dict) and aggs:
return aggs
return None
def _corroborating_results(tool_results: list[Any]) -> list[Any]:
"""The corroborating (non-alert) subset of the gathered tool results.
``tool_results`` are labeled ``{tool_name, result}`` items (the shape
:func:`soc_ai.api.hunt_runner._stream_node` now gathers). Items whose
``tool_name`` is in :data:`_ALERT_QUERY_TOOLS` are the detector's own claim
(Suricata alert documents, rule text, rule base-rates) and are EXCLUDED, so a
citation that resolves only into an alert document does NOT resolve here.
``t_query_events_oql`` is special-cased first: its result is partitioned
per-document by :func:`_oql_telemetry_docs`, so a zeek/host doc found via
the broad lens corroborates while the alert docs in the same result don't.
``t_get_event_raw`` gets the same partition (its single ``_source`` is wrapped
as a synthetic hit), so refetching the raw form of the detector alert doc does
NOT corroborate. Everything else — decoded payloads, enrichment,
host/prevalence, zeek flow records — is corroboration and its ``result`` is
included.
Defensive: a bare (un-labeled) item — a legacy shape, or a tool_result that
somehow arrived without a ``tool_name`` — is treated as corroborating (it is
not a KNOWN alert-query tool), so this can only ever ADD trust, never
silently strip a real corroboration on a shape surprise.
"""
corroborating: list[Any] = []
for item in tool_results:
if isinstance(item, dict) and "tool_name" in item and "result" in item:
tool = item.get("tool_name")
if tool == "t_query_events_oql":
# Partition at the DOC level: zeek/host docs found through the
# broad lens are real corroboration; alert docs and anything
# not positively identified stay out.
docs = _oql_telemetry_docs(item["result"])
if docs:
corroborating.append(docs)
else:
# An aggregation-only response (groupby/count -- the primer's
# top-talkers / rare-destination / NXDOMAIN-churn hunts) returns
# size=0: `hits` is empty and the measured data lives entirely in
# `aggregations`. Those are computed rollups over telemetry, not a
# detector's claim, so they corroborate -- include them.
aggs = _oql_aggregations(item["result"])
if aggs is not None:
corroborating.append(aggs)
continue
if tool == "t_get_event_raw":
# A raw fetch returns a single event's _source. It corroborates
# ONLY when that doc is positively identified as non-alert
# telemetry — refetching the raw form of the SAME detector alert
# (event.dataset=suricata.alert / kind=alert) is the alert
# re-stated, not corroboration. Reuse the OQL doc partition by
# wrapping the _source as a synthetic hit.
src = item["result"]
if isinstance(src, dict) and _oql_telemetry_docs({"hits": [{"_source": src}]}):
corroborating.append(src)
continue
if tool in _ALERT_QUERY_TOOLS:
continue # a detector-claim result cannot corroborate itself
corroborating.append(item["result"])
else:
# Un-labeled / legacy item: not a known alert-query tool → corroborating.
corroborating.append(item)
return corroborating
def _corroborating_evidence_text(tool_results: list[Any]) -> str:
"""Lower-cased JSON dump of ONLY the corroborating (non-alert) tool results
(see :func:`_corroborating_results` for the partition rules)."""
return _gathered_evidence_text(_corroborating_results(tool_results))
def _has_corroborating_citation(
citations: list[str], corroborating_text: str, corroborating_ids: frozenset[str]
) -> bool:
"""True iff ANY citation resolves into a NON-alert (corroborating) tool result.
Uses the same two-tier resolver as citation resolution, but against the
corroborating subset only — ``corroborating_text`` (the JSON dump of the
non-alert tool results) for value-shaped citations, ``corroborating_ids``
(the identity set collected from those same results) for id-shaped ones, so
an alert doc's id can resolve without corroborating. This is what a
high/critical THREAT finding must satisfy: at least one piece of support
that looked BEYOND the detector alert that raised the claim.
"""
if not corroborating_text and not corroborating_ids:
return False
return any(_citation_resolves(c, corroborating_text, corroborating_ids) for c in citations)
def _resolve_finding_citations(
citations: list[str], evidence_text: str, retrieved_ids: frozenset[str]
) -> tuple[list[str], float]:
"""Return (resolved_citations, coverage_ratio) for one finding's citations.
``coverage_ratio`` = resolved / total; an empty citation list is vacuously
1.0 (no missing evidence to penalize — same convention as the investigation
gate), and its resolved list is empty.
"""
if not citations:
return [], 1.0
resolved = [c for c in citations if _citation_resolves(c, evidence_text, retrieved_ids)]
return resolved, len(resolved) / len(citations)
def _validate_hunt_findings(
findings: list[Any],
tool_results: list[Any],
) -> tuple[list[Any], dict[str, int]]:
"""Deterministic citation gate over a HuntReport's findings — PURE.
For each finding (a :class:`~soc_ai.agent.hunt.HuntFinding` or any object
supporting ``model_copy(update=...)``):
* Resolve its ``citations`` against ``tool_results`` (the JSON payloads the
hunt actually gathered) via the distinctive-token resolver.
* ``coverage_ratio == 0`` with a NON-empty citation list (NONE resolve):
strip the non-resolving citations (keep any that resolved — here none),
cap severity at ``"low"``, set ``validator_note``.
* EMPTY citation list: left alone UNLESS severity is high/critical, in which
case cap to ``"medium"`` with the "lacks citations" note.
* Otherwise (at least one citation resolves): keep only the resolving
citations if some didn't resolve. The finding is GROUNDED — but a
``category == "threat"`` finding at high/critical severity gets one more
check (the CORROBORATION gate): at least one of its resolving citations
must resolve into a NON-alert tool result (a tool NOT in
:data:`_ALERT_QUERY_TOOLS`). If EVERY resolving citation resolves only into
detector-alert documents (Suricata alerts / rule text / rule base-rates),
the claim is asserted purely from the alert that raised it — severity is
capped to ``"medium"`` with :data:`_ALERT_ONLY_NOTE`. A non-threat finding,
or a threat below high, keeps its severity (only the loud high/critical
alert-title write-up is the trust failure this closes).
Independently of citations, every finding's ``title`` goes through
:func:`_clean_title`. The title is never cut.
Returns ``(validated_findings, counts)`` where ``counts`` carries per-hunt
tallies for the ``citation_validation`` audit event:
``{findings, findings_capped, citations_total, citations_stripped}``.
"""
# Drop the toolset's short-circuit payloads first: an error, a duplicate-call
# stop or a prefetch hint carries only the model's own arguments echoed
# back, and a citation must never resolve against those.
tool_results = _retrieved_results(tool_results)
evidence_text = _gathered_evidence_text(tool_results)
# Evidence set for id-shaped citations (M2): the document ids and typed
# values the hunt actually retrieved, collected from payload STRUCTURE —
# never dumped text.
retrieved_ids = _gathered_evidence_tokens(tool_results)
# The corroboration subset: only NON-alert tool results. A high/critical
# threat finding must cite at least one item resolving into THIS, or the
# claim rests solely on the detector alert that raised it. Computed once,
# with its own (narrower) identity set for id-shaped citations.
corroborating = _corroborating_results(tool_results)
corroborating_text = _gathered_evidence_text(corroborating)
corroborating_ids = _gathered_evidence_tokens(corroborating)
validated: list[Any] = []
counts = {
"findings": len(findings),
"findings_capped": 0,
"citations_total": 0,
"citations_stripped": 0,
}
for raw_finding in findings:
# Deterministic title cleanup (independent of citations). The title is
# stored whole.
title = str(getattr(raw_finding, "title", None) or "")
clean_title = _clean_title(title)
finding = (
raw_finding
if clean_title == title
else raw_finding.model_copy(update={"title": clean_title})
)
citations = list(getattr(finding, "citations", None) or [])
severity = str(getattr(finding, "severity", None) or "info")
counts["citations_total"] += len(citations)
if not citations:
# No citations: fine for an observation, but a high/critical claim
# with nothing behind it is capped to medium.
if _severity_rank(severity) >= _SEV_RANK["high"]:
validated.append(
finding.model_copy(
update={
"severity": _cap_severity(severity, "medium"),
"validator_note": _HIGH_NO_CITE_NOTE,
}
)
)
counts["findings_capped"] += 1
else:
validated.append(finding)
continue
resolved, coverage = _resolve_finding_citations(citations, evidence_text, retrieved_ids)
if coverage == 0.0:
# None of the finding's citations resolve to gathered evidence:
# strip them and cap the severity at low.
counts["citations_stripped"] += len(citations)
validated.append(
finding.model_copy(
update={
"citations": resolved, # == [] here (nothing resolved)
"severity": _cap_severity(severity, "low"),
"validator_note": _UNRESOLVED_NOTE,
}
)
)
counts["findings_capped"] += 1
elif len(resolved) < len(citations):
# Partial coverage: keep only the resolving citations (drop the
# fabricated ones). The finding is grounded, but a high/critical
# threat still faces the corroboration gate on its RESOLVING cites.
counts["citations_stripped"] += len(citations) - len(resolved)
grounded = finding.model_copy(update={"citations": resolved})
validated.append(
_apply_corroboration_cap(
grounded, resolved, corroborating_text, corroborating_ids, counts
)
)
else:
# Fully grounded — but a high/critical threat must still corroborate
# beyond the detector alert (the E1.3 pass above only proved the
# cited ids EXIST, not that they look past the alert).
validated.append(
_apply_corroboration_cap(
finding, resolved, corroborating_text, corroborating_ids, counts
)
)
return validated, counts
def _apply_corroboration_cap(
finding: Any,
resolved_citations: list[str],
corroborating_text: str,
corroborating_ids: frozenset[str],
counts: dict[str, int],
) -> Any:
"""Cap a high/critical THREAT finding that cites only detector alerts.
Given a finding that already PASSED citation resolution (its
``resolved_citations`` all exist in the gathered evidence), enforce the extra
corroboration bar: if it is a ``category == "threat"`` finding at high/critical
severity and NONE of its resolving citations resolve into a non-alert
(corroborating) tool result, its claim rests solely on the detector alert that
raised it — cap severity to ``"medium"`` and set :data:`_ALERT_ONLY_NOTE`
(incrementing ``findings_capped``). Anything else passes through untouched:
a non-threat finding, a threat below high, or a threat with real
corroboration keeps its severity. PURE aside from the ``counts`` tally.
"""
category = str(getattr(finding, "category", None) or "").strip().lower()
severity = str(getattr(finding, "severity", None) or "info")
if category in _NON_THREAT_CATEGORIES or _severity_rank(severity) < _SEV_RANK["high"]:
return finding
if _has_corroborating_citation(resolved_citations, corroborating_text, corroborating_ids):
return finding # grounded in evidence beyond the detector alert
counts["findings_capped"] += 1
return finding.model_copy(
update={
"severity": _cap_severity(severity, "medium"),
"validator_note": _ALERT_ONLY_NOTE,
}
)
# ── The lead-link gate ────────────────────────────────────────────────────────
#
# A link between leads is not activity on a host, and it is not a finding
# (change E, 2026-10-30). Hunt 01M5QVP4DZ46HNF7WM3J8FTHZK wrote "Leads 47 and
# 48 are one campaign" as a high threat finding: the hunt opened an
# investigation for it, and the run ended in error at the 900 s limit. The hunt
# of lead 49 wrote "Lead 47 and 49 are one campaign" as its F2. The prompt asks
# for the link in the narrative. This gate catches the model that writes it as
# a threat anyway: the finding becomes an observation with severity info, so
# no investigation opens for it. The text stays.
#
# The title decides, and the detail does not: a real threat may name a lead in
# its detail. A read-only scan of 980 prod findings matched only the 10
# lead-link titles. The gate only lowers a finding and writes a note, and the
# lead_link hold (soc_ai.store.leads) keeps such a lead from a clean close.
_LEAD_LINK_TITLE_RE = re.compile(
r"\bleads?\s+#?\d+\b|\brelated\s+leads?\b|\b(?:two|both|these|all)\s+leads\b",
re.IGNORECASE,
)
_LEAD_LINK_NOTE = (
"This finding links leads. soc-ai changed its type to observation and its "
"severity to info. The hunt opens no investigation for a link between leads."
)
def is_lead_link(finding: Any) -> bool:
"""True when the title of the finding states a link between leads."""
title = finding.get("title") if isinstance(finding, dict) else getattr(finding, "title", None)
return bool(_LEAD_LINK_TITLE_RE.search(str(title or "")))
def gate_lead_links(findings: list[Any]) -> tuple[list[Any], int]:
"""Lower each threat finding whose title links leads. PURE.
Returns the findings and how many the gate lowered. The note is set
outright: the type change replaces any earlier severity cap, and
``hunt_words.validator_note`` rewrites a note that combines two.
"""
out: list[Any] = []
lowered = 0
for finding in findings:
category = str(getattr(finding, "category", None) or "threat").strip().lower()
if category in _NON_THREAT_CATEGORIES or not is_lead_link(finding):
out.append(finding)
continue
lowered += 1
out.append(
finding.model_copy(
update={
"category": "observation",
"severity": "info",
"validator_note": _LEAD_LINK_NOTE,
}
)
)
return out, lowered
def _validate_hunt_charts(
charts: list[Any],
tool_results: list[Any],
) -> tuple[list[Any], dict[str, int]]:
"""Deterministic chart gate over a HuntReport's charts — PURE.
The model may author charts (a beacon-interval histogram, bytes-over-time)
that a generic chart can't guess — but an INVENTED series must never render.
Each chart is held to the SAME trust bar as findings (E3.3): its
``source_citations`` are resolved against ``tool_results`` (the JSON payloads
the hunt actually gathered) with the SAME distinctive-token resolver
:func:`_validate_hunt_findings` uses. A chart is DROPPED when it has:
* NO ``source_citations`` at all (nothing ties its numbers to gathered data), or
* source_citations of which NONE resolve to gathered evidence (an invented
series citing ids the hunt never pulled), or
* an EMPTY ``series`` (nothing to plot).
Surviving charts are capped at :data:`_MAX_CHARTS` (extras beyond the ceiling
are dropped) — bias is toward DROPPING; a chart that can't be traced to
evidence is never rendered. Kept charts get their ``title`` cleaned via
:func:`_clean_title` (the same cleanup as findings, never a cut).
Returns ``(kept_charts, counts)`` where ``counts`` carries per-hunt tallies
for the ``citation_validation`` audit event:
``{charts, charts_dropped}``.
"""
# Same corpus as findings: short-circuit echoes are not evidence a chart
# can be sourced from either.
tool_results = _retrieved_results(tool_results)
evidence_text = _gathered_evidence_text(tool_results)
# Same two-tier resolver as findings (M2): id-shaped source_citations must
# name evidence the hunt actually retrieved, not a token in dumped text.
retrieved_ids = _gathered_evidence_tokens(tool_results)
kept: list[Any] = []
counts = {"charts": len(charts), "charts_dropped": 0}
for chart in charts:
citations = list(getattr(chart, "source_citations", None) or [])
series = list(getattr(chart, "series", None) or [])
# DROP: nothing to plot, no citations, or none of the citations resolve.
if not series or not citations or len(kept) >= _MAX_CHARTS:
counts["charts_dropped"] += 1
continue
if not any(_citation_resolves(c, evidence_text, retrieved_ids) for c in citations):
counts["charts_dropped"] += 1
continue
# Same deterministic title cleanup as findings, only for kept charts.
title = str(getattr(chart, "title", None) or "")
clean_title = _clean_title(title)
kept.append(
chart if clean_title == title else chart.model_copy(update={"title": clean_title})
)
return kept, counts
# ── Host gap gate ────────────────────────────────────────────────────────────
# A production hunt reported "No host telemetry on <host> for attribution" as a
# visibility gap. Its evidence was two Elastic Defend queries that returned
# zero. The host shipped system.syslog, system.auth and osquery through
# Elastic Agent. The citation gate above exempts a visibility gap from every
# check, so nothing compared the claim with the host. This gate does: a gap
# that names a host is read against the host's coverage, and a gap on a host
# that ships any plane is rewritten to the plane the host lacks.
# How far back the gate reads a host's coverage. A hunt objective covers the
# past 72 hours unless it says otherwise.
HOST_GAP_WINDOW_HOURS = 72
# The most hosts one finding sends to the grid. A gap names one host or two.
_MAX_GAP_HOSTS = 3
_HOST_GAP_REWRITE_NOTE = (
"soc-ai read the host's coverage. The finding now names the plane the host does not ship."
)
_HOST_GAP_PRESENT_NOTE = (
"soc-ai read the host's coverage. The host ships every core plane, so this is no gap."
)
_HOST_GAP_UNREAD_NOTE = "soc-ai could not read the host's coverage."
# The planes a finding's text names. Order is the order a title names them in.
_PLANE_MENTIONS: tuple[tuple[str, re.Pattern[str]], ...] = (
("process", re.compile(r"\bprocess(?:es)?\b|\bsysmon\b|endpoint\.events\.process", re.I)),
(
"endpoint_network",
re.compile(
r"endpoint\.events\.network|\bendpoint\s+(?:process\s*(?:/|or|and)\s*)?network", re.I
),
),
("windows_security", re.compile(r"\bwindows\b|\bwinlog\b|\b46\d\d\b", re.I)),
(
"host_logs",
re.compile(r"\bsyslog\b|system\.auth|\bauth(?:entication)? logs?\b|\bhost logs?\b", re.I),
),
("osquery", re.compile(r"osquery", re.I)),
)
def _planes_named(text: str) -> list[str]:
return [plane for plane, pattern in _PLANE_MENTIONS if pattern.search(text)]
def _append_note(current: str | None, note: str) -> str:
if not current:
return note
if note in current:
return current
return f"{current} {note}"
def _rewrite_host_gap(finding: Any, covered: list[tuple[str, HostCoverage]]) -> Any:
"""The finding restated with the planes its host ships and the plane it lacks."""
host, cov = covered[0]
agent = cov.agent
label = agent.name if agent is not None and agent.name else host
title = str(getattr(finding, "title", "") or "")
detail = str(getattr(finding, "detail", "") or "")
absent = cov.absent()
named = _planes_named(f"{title} {detail}")
lacking = [p for p in named if p in absent] or absent
facts: list[str] = []
for other_host, other_cov in covered:
other_agent = other_cov.agent
subject = other_agent.name if other_agent is not None and other_agent.name else other_host
facts.extend(describe_coverage(other_cov, subject=subject))
kept, _n = rewrite_host_gap_claims(detail, lambda _sentence: "")
new_detail = " ".join([*facts, kept]).strip()
if lacking:
planes = " or ".join(plane_phrase(p) for p in lacking[:2])
return finding.model_copy(
update={
"title": _clean_title(f"No {planes} telemetry on {label}"),
"detail": new_detail,
"validator_note": _append_note(
getattr(finding, "validator_note", None), _HOST_GAP_REWRITE_NOTE
),
}
)
return finding.model_copy(
update={
"title": _clean_title(f"Host telemetry present on {label}"),
"detail": new_detail,
"category": "observation",
"validator_note": _append_note(
getattr(finding, "validator_note", None), _HOST_GAP_PRESENT_NOTE
),
}
)
async def _host_gap_coverage(
host: str,
*,
elastic: Any,
settings: Any,
sessionmaker: Any | None,
since: datetime,
until: datetime,
include_synth: SynthScope,
) -> HostCoverage:
"""One host's coverage. The identity comes from the dossier when there is a store."""
identity = HostIdentity(addresses=[host]) if _is_address(host) else HostIdentity(names=[host])
if sessionmaker is not None:
try:
async with sessionmaker() as db:
identity = await host_identity(db, host)
except Exception: # the address alone still reads the host
_LOGGER.warning("host gap gate: identity read failed for one host", exc_info=True)
return await host_coverage(
elastic,
settings,
addresses=identity.addresses,
names=identity.names,
agent_ids=identity.agent_ids,
since=since,
until=until,
include_synth=include_synth,
)
async def gate_host_gaps(
report: Any,
*,
elastic: Any,
settings: Any,
sessionmaker: Any | None = None,
include_synth: SynthScope = False,
now: datetime | None = None,
window_hours: int = HOST_GAP_WINDOW_HOURS,
) -> tuple[Any, dict[str, int]]:
"""Check each host gap of a HuntReport against the host's coverage.
A ``visibility_gap`` finding that names a host gets one coverage read per
host, cached across findings. When a host ships any plane, the finding is
rewritten: the detail states the planes the host ships and the planes it
does not ship, and the title names the plane, "No process telemetry on
X". When the host ships every core plane, the finding is no gap and
becomes an observation. When the read fails, the finding stays and its
note says that soc-ai could not read the host's coverage. A host that
ships nothing keeps its gap. Every other finding passes through.
The narrative gets the same rule. A narrative sentence that says a host
has no host telemetry is replaced by the coverage facts when the host
ships a plane. The hosts are the ones the gaps named, else the report's
affected hosts.
Returns ``(report, counts)``; counts are ``host_gaps``,
``host_gaps_rewritten``, ``host_gaps_unread`` and, when the narrative
changed, ``narrative_sentences_rewritten``.
"""
counts = {"host_gaps": 0, "host_gaps_rewritten": 0, "host_gaps_unread": 0}
findings = list(getattr(report, "findings", None) or [])
current = now or datetime.now(UTC)
since = current - timedelta(hours=max(1, window_hours))
cache: dict[str, HostCoverage] = {}
out: list[Any] = []
for finding in findings:
category = str(getattr(finding, "category", None) or "").strip().lower()
hosts = [str(h).strip() for h in (getattr(finding, "hosts", None) or []) if str(h).strip()]
if category != "visibility_gap" or not hosts:
out.append(finding)
continue
counts["host_gaps"] += 1
reads: list[tuple[str, HostCoverage]] = []
for host in hosts[:_MAX_GAP_HOSTS]:
if host not in cache:
cache[host] = await _host_gap_coverage(
host,
elastic=elastic,
settings=settings,
sessionmaker=sessionmaker,
since=since,
until=current,
include_synth=include_synth,
)
reads.append((host, cache[host]))
covered = [(h, c) for h, c in reads if c.covered]
if covered:
out.append(_rewrite_host_gap(finding, covered))
counts["host_gaps_rewritten"] += 1
elif any(not c.read_ok for _h, c in reads):
out.append(
finding.model_copy(
update={
"validator_note": _append_note(
getattr(finding, "validator_note", None), _HOST_GAP_UNREAD_NOTE
)
}
)
)
counts["host_gaps_unread"] += 1
else:
out.append(finding)
update: dict[str, Any] = {}
if counts["host_gaps"]:
update["findings"] = out
narrative = str(getattr(report, "narrative", "") or "")
if any(claims_no_host_telemetry(line) for line in re.split(r"(?<=[.!?])\s+", narrative)):
if not cache:
for host in [str(h).strip() for h in getattr(report, "affected_hosts", None) or []][
:_MAX_GAP_HOSTS
]:
if host:
cache[host] = await _host_gap_coverage(
host,
elastic=elastic,
settings=settings,
sessionmaker=sessionmaker,
since=since,
until=current,
include_synth=include_synth,
)
subjects = [coverage_subject(host, cov) for host, cov in cache.items()]
new_narrative, n = rewrite_host_gap_claims(narrative, coverage_replacer(subjects))
if n:
update["narrative"] = new_narrative
counts["narrative_sentences_rewritten"] = n
if not update:
return report, counts
return report.model_copy(update=update), counts