refactor(calibration): consolidate tunable thresholds into calibration.py
Session D: move every calibration constant (entropy floors 5.4/128, 5.1/64, 4.7/40 + shape floors; MAX_SCAN_CHARS; rot13-min; cognitive-load lengths 2000/2500; disposition ranks; active-content severities) into one documented calibration.py, so a parallel Node/TS port can mirror exactly the same numbers. Pure refactor, zero behavior change: calibration is a leaf module (imports only report.Severity) that entropy/lexicon/disposition/active_content now source their thresholds from. MAX_SCAN_CHARS is re-exported from lexicon so output.py and existing callers are unaffected. The 347 pre-existing tests pass unmodified; new test_calibration.py freezes the values and asserts each detector actually reads its threshold from calibration (identity-checked, not a dead copy).
This commit is contained in:
parent
4a9cfd2bbe
commit
ee402e4ea8
6 changed files with 212 additions and 34 deletions
|
|
@ -37,7 +37,8 @@ from __future__ import annotations
|
|||
|
||||
import re
|
||||
|
||||
from .report import Finding, Report, Severity, Source
|
||||
from .calibration import ACTIVE_CONTENT_SEVERITY as _SEVERITY
|
||||
from .report import Finding, Report, Source
|
||||
|
||||
# --- URL defang (shared primitive) -------------------------------------------
|
||||
# Rewrite a URL to a form no renderer will resolve, while keeping it readable.
|
||||
|
|
@ -138,15 +139,9 @@ def _always(url: str) -> bool:
|
|||
return True
|
||||
|
||||
|
||||
_SEVERITY = {
|
||||
# zero-click auto-fetch / auto-execute -> HIGH; click-required -> MEDIUM.
|
||||
"markdown-image": Severity.HIGH,
|
||||
"markdown-link": Severity.MEDIUM,
|
||||
"reference-link": Severity.MEDIUM,
|
||||
"autolink": Severity.MEDIUM,
|
||||
"raw-html": Severity.HIGH,
|
||||
"data-uri": Severity.HIGH,
|
||||
}
|
||||
# Per-construct severities (_SEVERITY, imported above) live in `calibration` —
|
||||
# zero-click auto-fetch/execute -> HIGH, click-required -> MEDIUM — the Node port
|
||||
# shares them.
|
||||
|
||||
|
||||
def scan_active_content(text: str, source: Source = Source.OUTPUT) -> Report:
|
||||
|
|
|
|||
77
src/llm_ingestion_guard/calibration.py
Normal file
77
src/llm_ingestion_guard/calibration.py
Normal file
|
|
@ -0,0 +1,77 @@
|
|||
"""calibration — the one place every tunable threshold lives.
|
||||
|
||||
Every detector in this package is calibrated by a handful of numbers: entropy
|
||||
floors, an input-size cap, minimum lengths, and two small severity/rank tables.
|
||||
Scattered across four modules, those numbers are impossible to audit and — the
|
||||
concrete driver here — impossible for a port to mirror *exactly*. A parallel
|
||||
Node/TypeScript implementation of this gate must classify byte-for-byte the same
|
||||
way, which means it must share the same constants. This module is that shared
|
||||
contract: one documented surface the port copies verbatim.
|
||||
|
||||
**This module holds values, never logic.** It depends only on
|
||||
:mod:`~llm_ingestion_guard.report` (for the :class:`Severity` enum used by the
|
||||
active-content table) and is imported by ``entropy``, ``lexicon``,
|
||||
``disposition`` and ``active_content`` — a leaf in the dependency graph, so no
|
||||
import cycle is possible.
|
||||
|
||||
Changing any number here is a deliberate recalibration, not a refactor:
|
||||
``tests/test_calibration.py`` freezes these values and asserts each detector
|
||||
actually sources its threshold from here.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from .report import Severity
|
||||
|
||||
# --- entropy: length-calibrated Shannon-entropy tiers -----------------------
|
||||
# Bits-per-char floor paired with a minimum length, because the achievable
|
||||
# entropy maximum is length-dependent (a short base64 string cannot reach the
|
||||
# entropy of a long one). Empirically calibrated against the seed scanner:
|
||||
# plaintext prose H ~3.5-4.2; base64 len64 H ~5.2; base64 len128 H ~5.6.
|
||||
ENTROPY_CRITICAL_H, ENTROPY_CRITICAL_LEN = 5.4, 128
|
||||
ENTROPY_HIGH_H, ENTROPY_HIGH_LEN = 5.1, 64
|
||||
ENTROPY_MEDIUM_H, ENTROPY_MEDIUM_LEN = 4.7, 40
|
||||
|
||||
# Shape floors: structured encodings suspicious by size even when their entropy
|
||||
# sits below the classification floor. The hex floor is the *only* path that
|
||||
# classifies a hex blob (a 16-symbol alphabet caps entropy at log2(16)=4.0,
|
||||
# below the 4.7 MEDIUM floor).
|
||||
ENTROPY_BASE64_FLOOR_LEN = 100
|
||||
ENTROPY_HEX_FLOOR_LEN = 64
|
||||
|
||||
# --- lexicon: self-safety + variant thresholds ------------------------------
|
||||
# Input-size cap (OWASP LLM10): large enough for a real ingested document;
|
||||
# beyond it the scanner reads the prefix and flags, so runtime stays bounded
|
||||
# even on a decompression-bomb-sized input.
|
||||
MAX_SCAN_CHARS = 1_000_000
|
||||
|
||||
# Minimum length before the rot13 variant is scanned — shorter strings hit
|
||||
# rot13-look-alike false positives.
|
||||
ROT13_MIN_LEN = 40
|
||||
|
||||
# Cognitive-load trap: a CRITICAL pattern found *only* past the first
|
||||
# COGNITIVE_LOAD_TAIL_START chars of text at least COGNITIVE_LOAD_MIN_LEN long is
|
||||
# a human-in-the-loop trap (an override buried at the tail of verbose output).
|
||||
COGNITIVE_LOAD_MIN_LEN = 2500
|
||||
COGNITIVE_LOAD_TAIL_START = 2000
|
||||
|
||||
# --- disposition: gate-decision ordering ------------------------------------
|
||||
# Rank of each disposition, keyed by its enum *value* string (kept primitive so
|
||||
# this leaf module needs no import from ``disposition``, which would cycle).
|
||||
# ``disposition`` rebuilds the enum-keyed map from this. Higher = more severe.
|
||||
DISPOSITION_RANK = {
|
||||
"warn": 0,
|
||||
"quarantine_review": 1,
|
||||
"fail_secure": 2,
|
||||
}
|
||||
|
||||
# --- active_content: per-construct severities -------------------------------
|
||||
# Zero-click auto-fetch / auto-execute constructs are HIGH; click-required ones
|
||||
# are MEDIUM. Mirrors ``neutralize``'s defang classes.
|
||||
ACTIVE_CONTENT_SEVERITY = {
|
||||
"markdown-image": Severity.HIGH,
|
||||
"markdown-link": Severity.MEDIUM,
|
||||
"reference-link": Severity.MEDIUM,
|
||||
"autolink": Severity.MEDIUM,
|
||||
"raw-html": Severity.HIGH,
|
||||
"data-uri": Severity.HIGH,
|
||||
}
|
||||
|
|
@ -25,6 +25,7 @@ from dataclasses import dataclass
|
|||
from enum import Enum
|
||||
from typing import Callable, Optional
|
||||
|
||||
from .calibration import DISPOSITION_RANK
|
||||
from .report import Report, Severity, severity_rank
|
||||
|
||||
|
||||
|
|
@ -82,11 +83,10 @@ _CARRIER_LABELS = frozenset({
|
|||
"lexicon:unicode-tags-present",
|
||||
})
|
||||
|
||||
_DISPOSITION_RANK = {
|
||||
Disposition.WARN: 0,
|
||||
Disposition.QUARANTINE_REVIEW: 1,
|
||||
Disposition.FAIL_SECURE: 2,
|
||||
}
|
||||
# Enum-keyed rank rebuilt from calibration's value-keyed source of truth
|
||||
# (calibration is a leaf module and cannot import the Disposition enum without a
|
||||
# cycle). Higher = more severe.
|
||||
_DISPOSITION_RANK = {d: DISPOSITION_RANK[d.value] for d in Disposition}
|
||||
|
||||
|
||||
def _more_severe(a: Disposition, b: Disposition) -> Disposition:
|
||||
|
|
|
|||
|
|
@ -39,19 +39,21 @@ import math
|
|||
import re
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from .calibration import (
|
||||
ENTROPY_BASE64_FLOOR_LEN as _BASE64_FLOOR_LEN,
|
||||
ENTROPY_CRITICAL_H as _CRITICAL_H,
|
||||
ENTROPY_CRITICAL_LEN as _CRITICAL_LEN,
|
||||
ENTROPY_HEX_FLOOR_LEN as _HEX_FLOOR_LEN,
|
||||
ENTROPY_HIGH_H as _HIGH_H,
|
||||
ENTROPY_HIGH_LEN as _HIGH_LEN,
|
||||
ENTROPY_MEDIUM_H as _MEDIUM_H,
|
||||
ENTROPY_MEDIUM_LEN as _MEDIUM_LEN,
|
||||
)
|
||||
from .report import Finding, Report, Severity, Source
|
||||
|
||||
# --- length-calibrated entropy thresholds (bits/char, min length) -----------
|
||||
# Empirically calibrated against real distributions in the seed scanner:
|
||||
# plaintext prose H ~3.5-4.2; base64 len64 H ~5.2; base64 len128 H ~5.6.
|
||||
_CRITICAL_H, _CRITICAL_LEN = 5.4, 128
|
||||
_HIGH_H, _HIGH_LEN = 5.1, 64
|
||||
_MEDIUM_H, _MEDIUM_LEN = 4.7, 40
|
||||
|
||||
# Shape-floor lengths: structured encodings that are suspicious by size even
|
||||
# when their entropy sits below the classification floor.
|
||||
_BASE64_FLOOR_LEN = 100
|
||||
_HEX_FLOOR_LEN = 64
|
||||
# Length-calibrated entropy thresholds (bits/char, min length) and shape-floor
|
||||
# lengths now live in `calibration` — the single source of truth the Node port
|
||||
# shares. See that module for the calibration rationale.
|
||||
|
||||
# Candidate extraction: maximal runs of the base64 alphabet (hex is a subset),
|
||||
# with optional trailing padding. Bounded class, no nested quantifier -> linear.
|
||||
|
|
|
|||
|
|
@ -41,13 +41,18 @@ from dataclasses import dataclass
|
|||
from pathlib import Path
|
||||
from urllib.parse import unquote
|
||||
|
||||
from .calibration import (
|
||||
COGNITIVE_LOAD_MIN_LEN,
|
||||
COGNITIVE_LOAD_TAIL_START,
|
||||
MAX_SCAN_CHARS,
|
||||
ROT13_MIN_LEN as _ROT13_MIN_LEN,
|
||||
)
|
||||
from .entropy import try_decode_base64
|
||||
from .report import Finding, Report, Severity, Source
|
||||
|
||||
# --- self-safety: input-size cap (OWASP LLM10) ------------------------------
|
||||
# Large enough for a real ingested document; beyond it we scan the prefix and
|
||||
# flag, so runtime stays bounded even on a decompression-bomb-sized input.
|
||||
MAX_SCAN_CHARS = 1_000_000
|
||||
# Self-safety input-size cap (OWASP LLM10), rot13 variant floor, and the
|
||||
# cognitive-load-trap lengths all live in `calibration` (the Node port shares
|
||||
# them). MAX_SCAN_CHARS is re-exported here for `output` and existing callers.
|
||||
|
||||
_LEXICON_FILE = "injection_lexicon.json"
|
||||
_FLAG_MAP = {"i": re.IGNORECASE, "m": re.MULTILINE, "s": re.DOTALL}
|
||||
|
|
@ -261,9 +266,9 @@ def check_cognitive_load_trap(text: str) -> str | None:
|
|||
chars* of long text (>=2500), else ``None``. Placement is the signal: an
|
||||
override buried at the tail of verbose output is a human-in-the-loop trap.
|
||||
"""
|
||||
if len(text) < 2500:
|
||||
if len(text) < COGNITIVE_LOAD_MIN_LEN:
|
||||
return None
|
||||
tail = text[2000:]
|
||||
tail = text[COGNITIVE_LOAD_TAIL_START:]
|
||||
for pattern in load_lexicon():
|
||||
if pattern.severity is Severity.CRITICAL and pattern.regex.search(tail):
|
||||
return pattern.id
|
||||
|
|
@ -271,8 +276,8 @@ def check_cognitive_load_trap(text: str) -> str | None:
|
|||
|
||||
|
||||
# --- variant set + scan ------------------------------------------------------
|
||||
|
||||
_ROT13_MIN_LEN = 40 # shorter strings hit rot13-look-alike false positives
|
||||
# _ROT13_MIN_LEN (imported from calibration): shorter strings hit
|
||||
# rot13-look-alike false positives.
|
||||
|
||||
|
||||
def _build_variants(text: str) -> list[tuple[str, str]]:
|
||||
|
|
|
|||
99
tests/test_calibration.py
Normal file
99
tests/test_calibration.py
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
"""test_calibration — freeze the shared calibration surface (Session D).
|
||||
|
||||
Session D consolidated every tunable threshold into
|
||||
``llm_ingestion_guard.calibration`` so the Node port can mirror *exactly* the
|
||||
same numbers. These tests are the frozen contract in two halves:
|
||||
|
||||
1. the raw values themselves (the tuple the port shares), and
|
||||
2. the proof that each detector actually *sources* its threshold from here —
|
||||
so the freeze is a live single-source-of-truth, not a dead copy that can
|
||||
silently drift from the value the code uses.
|
||||
|
||||
Changing a calibration number is a deliberate recalibration: it must break a
|
||||
test here first.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from llm_ingestion_guard import calibration as cal
|
||||
from llm_ingestion_guard.report import Severity
|
||||
|
||||
|
||||
# --- frozen raw values ------------------------------------------------------
|
||||
|
||||
def test_entropy_thresholds_frozen():
|
||||
assert (cal.ENTROPY_CRITICAL_H, cal.ENTROPY_CRITICAL_LEN) == (5.4, 128)
|
||||
assert (cal.ENTROPY_HIGH_H, cal.ENTROPY_HIGH_LEN) == (5.1, 64)
|
||||
assert (cal.ENTROPY_MEDIUM_H, cal.ENTROPY_MEDIUM_LEN) == (4.7, 40)
|
||||
|
||||
|
||||
def test_entropy_shape_floors_frozen():
|
||||
assert cal.ENTROPY_BASE64_FLOOR_LEN == 100
|
||||
assert cal.ENTROPY_HEX_FLOOR_LEN == 64
|
||||
|
||||
|
||||
def test_lexicon_selfsafety_frozen():
|
||||
assert cal.MAX_SCAN_CHARS == 1_000_000
|
||||
assert cal.ROT13_MIN_LEN == 40
|
||||
|
||||
|
||||
def test_cognitive_load_lengths_frozen():
|
||||
assert cal.COGNITIVE_LOAD_MIN_LEN == 2500
|
||||
assert cal.COGNITIVE_LOAD_TAIL_START == 2000
|
||||
|
||||
|
||||
def test_disposition_rank_frozen():
|
||||
assert cal.DISPOSITION_RANK == {
|
||||
"warn": 0,
|
||||
"quarantine_review": 1,
|
||||
"fail_secure": 2,
|
||||
}
|
||||
|
||||
|
||||
def test_active_content_severity_frozen():
|
||||
assert cal.ACTIVE_CONTENT_SEVERITY == {
|
||||
"markdown-image": Severity.HIGH,
|
||||
"markdown-link": Severity.MEDIUM,
|
||||
"reference-link": Severity.MEDIUM,
|
||||
"autolink": Severity.MEDIUM,
|
||||
"raw-html": Severity.HIGH,
|
||||
"data-uri": Severity.HIGH,
|
||||
}
|
||||
|
||||
|
||||
# --- binding: each detector reads its threshold from calibration ------------
|
||||
# The freeze is meaningful only if the modules actually READ these values. An
|
||||
# import alias binds the SAME object, so identity (`is`) proves the single
|
||||
# source of truth rather than a coincidental equal copy.
|
||||
|
||||
def test_entropy_module_sources_from_calibration():
|
||||
from llm_ingestion_guard import entropy
|
||||
assert entropy._CRITICAL_H is cal.ENTROPY_CRITICAL_H
|
||||
assert entropy._CRITICAL_LEN is cal.ENTROPY_CRITICAL_LEN
|
||||
assert entropy._HIGH_H is cal.ENTROPY_HIGH_H
|
||||
assert entropy._HIGH_LEN is cal.ENTROPY_HIGH_LEN
|
||||
assert entropy._MEDIUM_H is cal.ENTROPY_MEDIUM_H
|
||||
assert entropy._MEDIUM_LEN is cal.ENTROPY_MEDIUM_LEN
|
||||
assert entropy._BASE64_FLOOR_LEN is cal.ENTROPY_BASE64_FLOOR_LEN
|
||||
assert entropy._HEX_FLOOR_LEN is cal.ENTROPY_HEX_FLOOR_LEN
|
||||
|
||||
|
||||
def test_lexicon_module_sources_from_calibration():
|
||||
from llm_ingestion_guard import lexicon
|
||||
assert lexicon.MAX_SCAN_CHARS is cal.MAX_SCAN_CHARS
|
||||
assert lexicon._ROT13_MIN_LEN is cal.ROT13_MIN_LEN
|
||||
|
||||
|
||||
def test_disposition_module_sources_from_calibration():
|
||||
from llm_ingestion_guard import disposition
|
||||
from llm_ingestion_guard.disposition import Disposition
|
||||
# Enum-keyed rank reconstructed from calibration's value-keyed source.
|
||||
assert disposition._DISPOSITION_RANK == {
|
||||
Disposition.WARN: 0,
|
||||
Disposition.QUARANTINE_REVIEW: 1,
|
||||
Disposition.FAIL_SECURE: 2,
|
||||
}
|
||||
|
||||
|
||||
def test_active_content_module_sources_from_calibration():
|
||||
from llm_ingestion_guard import active_content
|
||||
assert active_content._SEVERITY is cal.ACTIVE_CONTENT_SEVERITY
|
||||
Loading…
Add table
Add a link
Reference in a new issue