diff --git a/extractor/llm_extract.py b/extractor/llm_extract.py index 0b6be964..b9421b4f 100644 --- a/extractor/llm_extract.py +++ b/extractor/llm_extract.py @@ -311,6 +311,28 @@ async def extract_with_llm( for key in merged: merged[key] = list(set(merged[key])) + # belief-gate verification pass. The LLM has *declared* which + # format-constrained IOCs it found; a deterministic parser now + # computes which of those are actually well-formed AND present in the + # source text (required - present). Truncated hashes, hallucinated + # CVEs and invented MITRE techniques are dropped here — before they + # reach the result dict — so they never clear the downstream + # confidence floor on method prior alone. Only the LLM-added, + # format-constrained keys are touched; regex/NER values already in + # `result` and non-gated LLM keys pass through untouched. Never + # raises; disable with VOIDACCESS_VERIFY_LLM_IOCS=false. + try: + from extractor.verify_gate import gate_llm_merged, is_enabled + if is_enabled(): + merged, _verdicts = gate_llm_merged(merged, text) + for _v in _verdicts: + logger.info( + "verify-gate: rejected %d/%d unverifiable %s from LLM", + len(_v.missing), len(_v.required), _v.entity_type, + ) + except Exception as _gate_exc: + logger.debug("verify-gate skipped (non-fatal): %s", _gate_exc) + # Merge LLM results into result dict (keyed by internal entity type) for llm_key, entity_type in _LLM_KEY_TO_TYPE.items(): new_vals = merged[llm_key] diff --git a/extractor/verify_gate.py b/extractor/verify_gate.py new file mode 100644 index 00000000..3dbfa49f --- /dev/null +++ b/extractor/verify_gate.py @@ -0,0 +1,215 @@ +""" +extractor/verify_gate.py — Deterministic verification gate for LLM-extracted IOCs. + +belief-gate discipline applied to entity extraction. The LLM *declares* which +format-constrained indicators it found (the "required" set); a deterministic +parser computes which of those are actually well-formed **and** present in the +source page text (the "present" set). Anything the LLM claims that the parser +cannot confirm (``required - present``) is rejected before it enters the entity +store — the same set-difference guarantee described in belief-gate/gate-REPL. + +Why this exists +--------------- +``extractor/llm_extract.py`` merges the LLM's ``file_hashes_*``, +``cve_identifiers`` and ``mitre_techniques`` verbatim into the result dict. +Only wallets (``_classify_wallet``) and URLs (``.onion`` split) are +re-verified. A truncated hash, a hallucinated ``CVE-2024-99999`` or an +invented ``T9999`` therefore flows straight into normalisation and can clear +the 0.80 confidence floor on method prior alone. This module closes that leak. + +Design +------ +Only *format-constrained* types are gated, and only the values the LLM added +(regex/NER values are already deterministic and are never routed here): + + FILE_HASH_MD5, FILE_HASH_SHA1, FILE_HASH_SHA256, CVE_NUMBER, MITRE_TECHNIQUE + +Two independent checks per value — **both** must pass: + 1. shape — ``fullmatch`` against the strict format regex (catches + truncated / malformed values). + 2. grounding — the exact token appears in the source page text as a + well-formed match (catches hallucinations the LLM invented + that are syntactically valid but never appeared). + +The regex patterns are imported from ``extractor.regex_patterns`` so there is a +single source of truth for what a valid hash / CVE / MITRE technique looks +like. Zero runtime dependencies; every function is total and never raises. + +Disable with ``VOIDACCESS_VERIFY_LLM_IOCS=false``. + +Public interface +---------------- +GateVerdict — dataclass: required / present / missing / ok +verify_type(entity_type, values, text) — (kept_values, GateVerdict) +gate_llm_merged(merged, page_text) — gate an llm_extract ``merged`` dict +is_enabled() — read the feature flag +""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass, field +from typing import Optional + +from extractor.regex_patterns import ( + _CVE_RE, + _FILE_HASH_MD5_RE, + _FILE_HASH_SHA1_RE, + _FILE_HASH_SHA256_RE, + _MITRE_TECHNIQUE_RE, +) + +logger = logging.getLogger(__name__) + +# Strict format patterns per gated entity type — single source of truth is +# extractor.regex_patterns (these are the same compiled objects the regex tier +# uses), so the gate and the regex extractor can never drift apart. +_FORMAT_RE = { + "FILE_HASH_MD5": _FILE_HASH_MD5_RE, + "FILE_HASH_SHA1": _FILE_HASH_SHA1_RE, + "FILE_HASH_SHA256": _FILE_HASH_SHA256_RE, + "CVE_NUMBER": _CVE_RE, + "MITRE_TECHNIQUE": _MITRE_TECHNIQUE_RE, +} + +# Hex indicators compare case-insensitively; CVE / MITRE canonicalise to upper +# (matching _extract_cve / _extract_mitre which upper-case their output). +_HEX_TYPES = frozenset({"FILE_HASH_MD5", "FILE_HASH_SHA1", "FILE_HASH_SHA256"}) +_UPPER_TYPES = frozenset({"CVE_NUMBER", "MITRE_TECHNIQUE"}) + +# llm_extract.py output keys → internal entity type, for the gated subset only. +# Keys absent here (crypto_wallets, urls, dates, threat_actor_handles, +# malware_names) are handled elsewhere and pass through this gate untouched. +_LLM_KEY_TO_GATED_TYPE = { + "cve_identifiers": "CVE_NUMBER", + "mitre_techniques": "MITRE_TECHNIQUE", + "file_hashes_md5": "FILE_HASH_MD5", + "file_hashes_sha1": "FILE_HASH_SHA1", + "file_hashes_sha256": "FILE_HASH_SHA256", +} + + +@dataclass +class GateVerdict: + """Outcome of verifying one entity type's LLM-claimed values. + + ``ok`` is True only when nothing was rejected — mirroring belief-gate's + "never certify an answer it can't prove": if ``missing`` is non-empty the + caller knows exactly which claims failed and why. + """ + + entity_type: str + required: list[str] = field(default_factory=list) # what the LLM claimed + present: list[str] = field(default_factory=list) # parser-verified, kept + missing: list[str] = field(default_factory=list) # required - present + ok: bool = True + + +def is_enabled() -> bool: + """True unless VOIDACCESS_VERIFY_LLM_IOCS is explicitly falsey.""" + return os.getenv("VOIDACCESS_VERIFY_LLM_IOCS", "true").strip().lower() not in ( + "0", + "false", + "no", + "off", + ) + + +def _canon(entity_type: str, value: str) -> str: + """Canonical form for grounding comparison.""" + v = value.strip() + if entity_type in _HEX_TYPES: + return v.lower() + if entity_type in _UPPER_TYPES: + return v.upper() + return v + + +def _present_tokens(entity_type: str, page_text: str) -> set[str]: + """Canonical set of well-formed tokens of *entity_type* actually in *text*. + + Uses the format regex directly (no surrounding-context requirement) so a + genuine hash the LLM found where the regex tier's context gate missed it is + still credited as grounded — while a value the LLM invented (absent from the + text) is not. + """ + rex = _FORMAT_RE.get(entity_type) + if rex is None or not page_text: + return set() + return {_canon(entity_type, m.group(0)) for m in rex.finditer(page_text)} + + +def verify_type( + entity_type: str, + values, + page_text: str, + *, + grounded: bool = True, +) -> tuple[list[str], GateVerdict]: + """Verify *values* claimed for *entity_type* against the deterministic parser. + + Returns ``(kept_values, verdict)``. A value is kept only if it (1) matches + the strict format regex and (2) — when *grounded* — appears in *page_text*. + Non-gated types pass through unchanged with an all-ok verdict. Never raises. + """ + required = [str(v) for v in (values or [])] + rex = _FORMAT_RE.get(entity_type) + if rex is None: + return required, GateVerdict(entity_type, required, list(required), [], True) + + try: + present_in_text = _present_tokens(entity_type, page_text) if grounded else None + kept: list[str] = [] + present: list[str] = [] + missing: list[str] = [] + seen: set[str] = set() + + for raw in required: + val = (raw or "").strip() + if not val: + continue + canon = _canon(entity_type, val) + shape_ok = bool(rex.fullmatch(val)) + ground_ok = True if present_in_text is None else (canon in present_in_text) + if shape_ok and ground_ok: + if canon not in seen: + seen.add(canon) + kept.append(val) + present.append(val) + else: + missing.append(val) + + return kept, GateVerdict(entity_type, required, present, missing, not missing) + except Exception as exc: # noqa: BLE001 — gate must never break extraction + logger.debug("verify_type failed for %s (non-fatal): %s", entity_type, exc) + # Fail-open on an internal gate error: return the input unchanged rather + # than silently dropping legitimate entities on a bug in the gate. + return required, GateVerdict(entity_type, required, list(required), [], True) + + +def gate_llm_merged( + merged: dict[str, list], + page_text: str, + *, + grounded: bool = True, +) -> tuple[dict[str, list], list[GateVerdict]]: + """Verify the format-constrained subset of an ``llm_extract`` merged dict. + + *merged* is keyed by LLM output key (e.g. ``file_hashes_sha256``). Only the + gated keys are touched; everything else is copied through untouched. Returns + ``(gated_merged, verdicts)`` where *verdicts* holds only the types that had + at least one rejection (empty when nothing was rejected). + """ + out = dict(merged) + verdicts: list[GateVerdict] = [] + for llm_key, entity_type in _LLM_KEY_TO_GATED_TYPE.items(): + if llm_key not in out: + continue + kept, verdict = verify_type( + entity_type, out[llm_key], page_text, grounded=grounded + ) + out[llm_key] = kept + if verdict.missing: + verdicts.append(verdict) + return out, verdicts diff --git a/tests/test_verify_gate.py b/tests/test_verify_gate.py new file mode 100644 index 00000000..b7f47a98 --- /dev/null +++ b/tests/test_verify_gate.py @@ -0,0 +1,171 @@ +""" +tests/test_verify_gate.py — belief-gate verification of LLM-extracted IOCs. + +The gate must be leak-proof: it may never keep a value that is malformed or +absent from the source text, and it must never drop a value that is both +well-formed and genuinely present. These tests assert both directions plus +the fail-open / feature-flag behaviour. +""" + +from __future__ import annotations + +import pytest + +from extractor.verify_gate import ( + GateVerdict, + gate_llm_merged, + is_enabled, + verify_type, +) + +# A realistic, fully-formed set of indicators. +GOOD_MD5 = "d41d8cd98f00b204e9800998ecf8427e" # 32 hex +GOOD_SHA1 = "da39a3ee5e6b4b0d3255bfef95601890afd80709" # 40 hex +GOOD_SHA256 = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" # 64 hex +GOOD_CVE = "CVE-2024-3094" +GOOD_MITRE = "T1486" + +PAGE_TEXT = ( + "Malware sample md5 " + f"{GOOD_MD5} was dropped; the loader sha256 {GOOD_SHA256} beacons out. " + f"Exploits {GOOD_CVE} and maps to ATT&CK {GOOD_MITRE}. " + f"File hash sha1 {GOOD_SHA1} observed." +) + + +# --------------------------------------------------------------------------- # +# Shape rejection — malformed values never pass, regardless of grounding. +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize( + "entity_type,bad_value", + [ + ("FILE_HASH_SHA256", "e3b0c44298fc1c149afbf4c8"), # truncated + ("FILE_HASH_MD5", "d41d8cd98f00b204e9800998ecf8427"), # 31 chars + ("FILE_HASH_SHA1", "zz39a3ee5e6b4b0d3255bfef95601890afd80709"), # non-hex + ("CVE_NUMBER", "CVE-24-3094"), # 2-digit year + ("CVE_NUMBER", "CVE-2024-99"), # id too short + ("MITRE_TECHNIQUE", "T999"), # only 3 digits + ("MITRE_TECHNIQUE", "T1486.1"), # bad sub-technique + ], +) +def test_malformed_values_are_rejected(entity_type, bad_value): + # Put the bad value in the text too, to prove shape (not grounding) rejects it. + kept, verdict = verify_type(entity_type, [bad_value], f"context {bad_value}") + assert kept == [] + assert verdict.missing == [bad_value] + assert verdict.ok is False + + +# --------------------------------------------------------------------------- # +# Grounding rejection — well-formed but hallucinated values never pass. +# --------------------------------------------------------------------------- # +def test_hallucinated_but_wellformed_hash_is_rejected(): + hallucinated = "a" * 64 # valid SHA256 shape, absent from PAGE_TEXT + kept, verdict = verify_type("FILE_HASH_SHA256", [hallucinated], PAGE_TEXT) + assert kept == [] + assert verdict.missing == [hallucinated] + assert verdict.ok is False + + +def test_hallucinated_cve_is_rejected(): + kept, verdict = verify_type("CVE_NUMBER", ["CVE-2024-99999"], PAGE_TEXT) + assert kept == [] + assert not verdict.ok + + +# --------------------------------------------------------------------------- # +# True positives — well-formed AND present values are kept. +# --------------------------------------------------------------------------- # +def test_genuine_values_are_kept(): + for entity_type, value in ( + ("FILE_HASH_MD5", GOOD_MD5), + ("FILE_HASH_SHA1", GOOD_SHA1), + ("FILE_HASH_SHA256", GOOD_SHA256), + ("CVE_NUMBER", GOOD_CVE), + ("MITRE_TECHNIQUE", GOOD_MITRE), + ): + kept, verdict = verify_type(entity_type, [value], PAGE_TEXT) + assert kept == [value], f"{entity_type} dropped a genuine value" + assert verdict.ok is True + assert verdict.missing == [] + + +def test_grounding_is_case_insensitive_for_hex(): + kept, verdict = verify_type("FILE_HASH_SHA256", [GOOD_SHA256.upper()], PAGE_TEXT) + assert kept == [GOOD_SHA256.upper()] + assert verdict.ok + + +def test_mixed_batch_keeps_only_verified(): + values = [GOOD_SHA256, "a" * 64, "e3b0c44298fc"] # good, hallucinated, truncated + kept, verdict = verify_type("FILE_HASH_SHA256", values, PAGE_TEXT) + assert kept == [GOOD_SHA256] + assert set(verdict.missing) == {"a" * 64, "e3b0c44298fc"} + assert verdict.ok is False + + +# --------------------------------------------------------------------------- # +# gate_llm_merged — only gated keys touched; non-gated keys pass through. +# --------------------------------------------------------------------------- # +def test_gate_llm_merged_filters_only_gated_keys(): + merged = { + "file_hashes_sha256": [GOOD_SHA256, "a" * 64], + "cve_identifiers": [GOOD_CVE, "CVE-2024-99999"], + "mitre_techniques": [GOOD_MITRE], + # non-gated keys must be returned untouched + "threat_actor_handles": ["@evilcorp"], + "malware_names": ["LockBit"], + "crypto_wallets": ["bc1qxy"], + } + out, verdicts = gate_llm_merged(merged, PAGE_TEXT) + + assert out["file_hashes_sha256"] == [GOOD_SHA256] + assert out["cve_identifiers"] == [GOOD_CVE] + assert out["mitre_techniques"] == [GOOD_MITRE] + # untouched + assert out["threat_actor_handles"] == ["@evilcorp"] + assert out["malware_names"] == ["LockBit"] + assert out["crypto_wallets"] == ["bc1qxy"] + # verdicts recorded only for keys that had rejections + rejected_types = {v.entity_type for v in verdicts} + assert rejected_types == {"FILE_HASH_SHA256", "CVE_NUMBER"} + + +def test_ungated_type_passes_through_verify_type(): + kept, verdict = verify_type("MALWARE_FAMILY", ["LockBit", "Emotet"], PAGE_TEXT) + assert kept == ["LockBit", "Emotet"] + assert verdict.ok is True + + +# --------------------------------------------------------------------------- # +# Grounding can be disabled independently of shape. +# --------------------------------------------------------------------------- # +def test_grounding_disabled_keeps_wellformed_absent_value(): + hallucinated = "b" * 64 + kept, verdict = verify_type( + "FILE_HASH_SHA256", [hallucinated], "no hashes here", grounded=False + ) + assert kept == [hallucinated] # shape passes, grounding not checked + assert verdict.ok + + +# --------------------------------------------------------------------------- # +# Feature flag. +# --------------------------------------------------------------------------- # +def test_is_enabled_defaults_true(monkeypatch): + monkeypatch.delenv("VOIDACCESS_VERIFY_LLM_IOCS", raising=False) + assert is_enabled() is True + + +@pytest.mark.parametrize("val", ["false", "0", "no", "off", "FALSE", "Off"]) +def test_is_enabled_falsey_values(monkeypatch, val): + monkeypatch.setenv("VOIDACCESS_VERIFY_LLM_IOCS", val) + assert is_enabled() is False + + +def test_empty_and_blank_values_are_dropped_quietly(): + kept, verdict = verify_type("CVE_NUMBER", ["", " ", GOOD_CVE], PAGE_TEXT) + assert kept == [GOOD_CVE] + # blank strings are skipped, not counted as missing + assert verdict.missing == [] + assert verdict.ok is True