feat(ENG-1084): detect-and-gate + sentence-level cleaning (v0.8.2)

README.md

@@ -90,10 +90,12 @@ else:
 
 ### Tier 1 — Pattern detection (sync, ~1 ms)
 
-- **Unicode normalization** — homoglyph resistance (e.g. Cyrillic `а` → ASCII `a`)
-- **Role stripping** — `SYSTEM:`, `ASSISTANT:`, `<system>`, `[INST]`, etc.
-- **Pattern removal** — phrases like “ignore previous instructions”
-- **Encoding detection** — suspicious Base64/URL-shaped payloads
+Detects (never rewrites) injection signals and records them as evidence:
+
+- **Unicode normalization** (for matching only) — homoglyph resistance (e.g. Cyrillic `а` → ASCII `a`)
+- **Role markers** — `SYSTEM:`, `ASSISTANT:`, `<system>`, `[INST]`, etc.
+- **Instruction-override patterns** — phrases like “ignore previous instructions”
+- **Encoding detection** — decode-then-detect: a decoded payload is escalated only if it trips a real attack pattern
 - **Boundary annotation (opt-in)** — `[UD-{id}]…[/UD-{id}]` wrappers when `annotate_boundary=True` (npm: `annotateBoundary`). Use `generate_boundary_instructions` from the package root in prompts when you enable wrapping.
 
 ### Tier 2 — ML classification (ONNX)
@@ -107,7 +109,7 @@ Packed-chunk MiniLM classifier (int8 ONNX ~22 MB, bundled):
 ### Optional SFE preprocessor
 
 - `use_sfe=True` runs a field-level FastText pass to build a **classifier-only** view of the payload
-- **Tier 1** always sanitizes the **original** tool value; **`sanitized`** in `DefenseResult` is unchanged by SFE drops
+- **Tier 1** detects on the raw tool value; SFE drops are classifier-only and never remove fields from the returned `sanitized` payload
 - **Tier 2** extracts strings from the SFE-filtered tree; `fields_dropped` lists paths omitted from that extraction (not removed from `sanitized`)
 - Fails open if the runtime/model is unavailable: payload continues unfiltered
 
@@ -126,7 +128,7 @@ Authoritative LLM-based classification for the cases Tier 2 finds ambiguous. The
 
 Two modes, selected via `defender_mode`:
 - **`"cascade"`** (default): Tier 1 → Tier 2 → Tier 3, with Tier 3 invoked only when the Tier 2 effective score falls in the gray band (default `[0.3, 0.85)`). The Tier 3 verdict overrides Tier 2 on the escalated chunk — a `block` forces a block, an `allow` rescues it. Outside the band defender skips the round trip.
-- **`"tier3_only"`**: skip Tier 2; the block/allow decision is the Tier 3 verdict alone. Tier 1 still sanitizes the returned `sanitized` payload.
+- **`"tier3_only"`**: skip Tier 2; the block/allow decision is the Tier 3 verdict alone. Tier 1 still runs detection; the returned `sanitized` payload is the original content.
 
 Register a provider once at startup, then opt in per instance:
 
@@ -163,8 +165,9 @@ defense = create_prompt_defense(
 
 ### `allowed` vs `risk_level`
 
+- **`DefenseResult.sanitized`** is a **sentence-level cleaned** copy of the tool result (high-scoring sentences dropped within high-risk fields, optionally `[UD-…]` boundary-wrapped). Cleaning is best-effort (capped by detection) — still gate on `allowed`. Set `sanitize_content=False` for pure detect-and-gate: `sanitized` is then the input verbatim.
 - Use **`allowed`** for gating when `block_high_risk=True`: `False` means do not pass `sanitized` to the model as-is.
-- **`risk_level`** is diagnostic: it starts at `default_risk_level` (default `"medium"`) and is **escalated** by Tier 1 / Tier 2 signals — not reduced. Use it for logging, not as the sole block signal unless you implement your own policy.
+- **`risk_level`** is diagnostic: it starts at `default_risk_level` (default `"low"`) and is **escalated** by Tier 1 / Tier 2 signals — not reduced. Use it for logging, not as the sole block signal unless you implement your own policy.
 
 | Level | Typical trigger |
 |-------|------------------|
@@ -180,8 +183,9 @@ defense = create_prompt_defense(
 defense = create_prompt_defense(
     enable_tier1=True,
     enable_tier2=True,
+    require_tier2=False,  # True: raise if Tier 2 can't load (fail closed) instead of degrading to Tier 1
     block_high_risk=False,
-    default_risk_level="medium",
+    default_risk_level="low",
     annotate_boundary=False,  # True: wrap risky strings with [UD-…] tags (npm: annotateBoundary)
     tier2_fields=["subject", "body", "snippet"],  # optional: scope Tier 2 to these JSON keys (default: all strings)
     use_sfe=True,  # optional: enable semantic field extractor preprocessing
@@ -203,18 +207,31 @@ from dataclasses import dataclass, field
 
 @dataclass
 class DefenseResult:
-    allowed: bool
-    risk_level: RiskLevel
-    sanitized: Any
-    detections: list[str]
-    fields_sanitized: list[str]
-    patterns_by_field: dict[str, list[str]]
+    allowed: bool                             # gating decision (respects block_high_risk)
+    risk_level: RiskLevel                     # diagnostic; max of Tier 1 / Tier 2
+    sanitized: Any                            # sentence-cleaned copy (input verbatim when sanitize_content=False); dropped runs leave a [CONTENT SANITISED] marker; best-effort, still gate on allowed
+    detections: list[str]                     # Tier 1 pattern names detected
+    fields_sanitized: list[str]               # fields whose content the cleaner changed in sanitized (empty when sanitize_content=False or no Tier 2); for detections read detections/patterns_by_field
+    patterns_by_field: dict[str, list[str]]   # patterns detected per field
+    detected_field_count: int                 # count of fields with a Tier-1 detection (keys of patterns_by_field); threat-count signal (fields_sanitized len no longer tracks this)
     tier2_score: float | None = None
+    tier2_raw_score: float | None = None
+    tier2_aux_score: float | None = None      # multi-head models only
+    tier2_multihead_blocked: bool | None = None
     tier2_skip_reason: str | None = None
     max_sentence: str | None = None
+    tier3: Tier3Result | None = None          # present when Tier 3 ran
     fields_dropped: list[str] = field(default_factory=list)
     truncated_at_depth: bool | None = None
     latency_ms: float = 0.0
+    # Cost telemetry — present only when the batched Tier 2 classifier ran
+    phase_timings: PhaseTimings | None = None       # prepare / infer / aggregate ms
+    tier2_stats: Tier2Stats | None = None           # string/chunk/unique counts, real/padded tokens
+    tier1_ms: float | None = None
+    cold_load: bool | None = None
+    # Operational signals
+    tier2_available: bool | None = None       # False when Tier 2 enabled but failed to load
+    coverage_degraded: bool | None = None      # True when Tier 1 detection coverage was capped
 ```
 
 ### `defense.defend_tool_results(items)`
@@ -229,7 +246,7 @@ results = defense.defend_tool_results([
 ])
 for r in results:
     if not r.allowed:
-        print("Blocked:", ", ".join(r.fields_sanitized))
+        print("Blocked:", ", ".join(r.detections))
 ```
 
 ### `await defense.defend_tool_results_async(items)`
@@ -274,7 +291,7 @@ sanitized = run_tool_and_defend(gmail_api.get_message(msg_id), "gmail_get_messag
 
 ## Risky field detection
 
-Only **string** values under configured “risky” keys are scanned and sanitized. [`RiskyFieldConfig`](https://github.com/StackOneHQ/stackone-defender/blob/main/src/stackone_defender/types.py) provides global names/patterns plus **`tool_overrides`** (wildcard tool names → field list), same idea as the npm package.
+Only **string** values under configured “risky” keys are Tier-1-scanned — including strings nested inside arrays/objects under those keys (e.g. `{"name": ["…"]}`). [`RiskyFieldConfig`](https://github.com/StackOneHQ/stackone-defender/blob/main/src/stackone_defender/types.py) provides global names/patterns plus **`tool_overrides`** (wildcard tool names → field list), same idea as the npm package. (Tier 2 scans all extracted strings regardless.)
 
 | Tool pattern | Scanned fields |
 |--------------|----------------|

src/stackone_defender/classifiers/onnx_classifier.py

@@ -75,14 +75,38 @@ class OnnxClassifier:
     # function of the string alone, not its batch neighbours (deterministic).
     _PAD_BUCKETS = (32, 64, 128, 256)
 
-    def __init__(self, model_path: str | None = None, temperature_t: float | None = None):
+    # Token-degeneracy (OOD) guard. Below this many content tokens, token-share
+    # is too coarse to mean anything (a 1-2 token row is trivially "dominated");
+    # matches the shortest decorative run that still false-fires post-collapse
+    # (``─``x3 -> 3 content tokens). Damp only when the row draws on at most
+    # _DEGENERACY_MAX_DISTINCT_TOKENS distinct tokens — padding an attack with
+    # copies of one token can push the share past 2/3 but cannot remove the
+    # attack's own vocabulary, so the distinct-token floor is a structural, not
+    # tuned, backstop against that bypass.
+    _DEGENERACY_MIN_CONTENT_TOKENS = 3
+    _DEGENERACY_MAX_DISTINCT_TOKENS = 4
+
+    def __init__(
+        self,
+        model_path: str | None = None,
+        temperature_t: float | None = None,
+        degeneracy_max_token_share: float | None = None,
+    ):
         self._model_path = model_path or get_default_model_path()
         self._session = None
         self._tokenizer = None
         # Non-truncating tokenizer used only by count_tokens (see _load_model).
         self._count_tokenizer = None
         self._max_length = 256
         self._load_failed = False
+        # Token-degeneracy guard threshold; > 1 disables. See _is_degenerate.
+        self._degeneracy_max_token_share = 2 / 3
+        if degeneracy_max_token_share is not None and math.isfinite(degeneracy_max_token_share):
+            self._degeneracy_max_token_share = float(degeneracy_max_token_share)
+        # Cached [UNK] id, resolved lazily from the loaded tokenizer. ``None`` =
+        # not yet resolved or no [UNK] concept; see _get_unk_token_id.
+        self._unk_token_id: int | None = None
+        self._unk_resolved = False
         # Output mode is detected lazily from the logits shape on the first
         # inference call. ``None`` until then.
         self._output_mode: Literal["single", "multi"] | None = None
@@ -145,8 +169,10 @@ def _load_model(self) -> None:
                 import onnxruntime as ort
                 from tokenizers import Tokenizer
             except ImportError as e:
+                # No warning here -- the ImportError propagates to the caller,
+                # which owns user-facing messaging (PromptDefense warns once per
+                # instance). Warning here logged a line on every failed call.
                 self._load_failed = True
-                _logger.warning("[defender] ONNX model failed to load: %s", e)
                 raise ImportError(
                     "ONNX dependencies not installed. Install with: pip install stackone-defender[onnx]"
                 ) from e
@@ -198,6 +224,10 @@ def classify_pair(self, text: str) -> tuple[float, float | None]:
         import numpy as np
 
         encoding = self._tokenizer.encode(text)
+        # Fix 3: token-degeneracy guard — skip inference on off-distribution
+        # input; its mean-pooled score is arbitrary. Damp to a benign 0.
+        if self._is_degenerate(encoding.ids):
+            return 0.0, None
         input_ids = np.array([encoding.ids], dtype=np.int64)
         attention_mask = np.array([encoding.attention_mask], dtype=np.int64)
 
@@ -264,8 +294,61 @@ def classify_batch_pair(
                 for k, orig_idx in enumerate(idxs):
                     pairs[orig_idx] = chunk_pairs[k]
 
+        # Fix 3: token-degeneracy guard — damp off-distribution rows to a benign
+        # 0 so they drop out of any upstream max. Reuses the already-computed ids.
+        for i, enc in enumerate(encodings):
+            if self._is_degenerate(enc.ids):
+                pairs[i] = (0.0, None)
+
         return cast(list[tuple[float, float | None]], pairs)
 
+    def _get_unk_token_id(self) -> int | None:
+        """Resolve the tokenizer's ``[UNK]`` id, cached. Returns ``None`` when the
+        tokenizer has no ``[UNK]`` concept (the guard then skips its factor-3 check)."""
+        if self._unk_resolved:
+            return self._unk_token_id
+        self._unk_resolved = True
+        try:
+            self._unk_token_id = self._tokenizer.token_to_id("[UNK]")
+        except Exception:
+            self._unk_token_id = None
+        return self._unk_token_id
+
+    def _is_degenerate(self, ids: list[int]) -> bool:
+        """Token-degeneracy (OOD) test over a tokenized row. Damps only when ALL
+        of these hold over the content tokens (excluding [CLS]/[SEP]):
+
+        1. the single most-frequent token covers >= ``degeneracy_max_token_share``,
+        2. the row draws on <= ``_DEGENERACY_MAX_DISTINCT_TOKENS`` distinct tokens, and
+        3. the dominant token is NOT [UNK].
+
+        Factor 2 blocks a padding attack; factor 3 blocks a homoglyph attack —
+        fullwidth / zero-width / other OOV chars collapse to repeated [UNK], the
+        signature of encoding evasion (more suspicious, not less), so those rows
+        are left to score rather than suppressed. Reuses the ids the model runs on.
+        """
+        threshold = self._degeneracy_max_token_share
+        if not (0 < threshold <= 1):  # disabled
+            return False
+        has_specials = len(ids) >= 2
+        content = ids[1:-1] if has_specials else ids
+        n = len(content)
+        if n < self._DEGENERACY_MIN_CONTENT_TOKENS:
+            return False
+        counts: dict[int, int] = {}
+        max_freq = 0
+        dominant_id = -1
+        for tok in content:
+            c = counts.get(tok, 0) + 1
+            counts[tok] = c
+            if c > max_freq:
+                max_freq = c
+                dominant_id = tok
+        if max_freq / n < threshold or len(counts) > self._DEGENERACY_MAX_DISTINCT_TOKENS:
+            return False
+        unk = self._get_unk_token_id()
+        return unk is None or dominant_id != unk
+
     def _classify_batch_chunk_pair(
         self, encodings: list, pad_to: int | None = None
     ) -> list[tuple[float, float | None]]:

src/stackone_defender/classifiers/patterns.py

@@ -134,7 +134,7 @@
     PatternDefinition("html_entity_abuse", re.compile(r"(?:&#\d{2,4};){4,}|(?:&#x[0-9a-fA-F]{2,4};){4,}", re.I), "encoding_suspicious", "medium", "HTML entity encoding (potential obfuscation)"),
     PatternDefinition("rot13_mention", re.compile(r"rot13|caesar\s+cipher|decode\s+this", re.I), "encoding_suspicious", "medium", "Mention of ROT13 or similar encoding schemes"),
     PatternDefinition("binary_string_encoding", re.compile(r"\b[01]{8}(?:\s+[01]{8}){2,}\b"), "encoding_suspicious", "medium", "Binary-encoded string (potential obfuscation)"),
-    PatternDefinition("morse_code_encoding", re.compile(r"(?:[.-]+\s){4,}[.-]+"), "encoding_suspicious", "low", "Morse code pattern (potential obfuscation)"),
+    PatternDefinition("morse_code_encoding", re.compile(r"(?:[.-]{1,8}\s){4,}[.-]{1,8}"), "encoding_suspicious", "low", "Morse code pattern (potential obfuscation)"),
     PatternDefinition("leetspeak_injection", re.compile(r"1gn0r3|f0rg3t|byp4ss|syst3m|4dm1n|h4ck", re.I), "encoding_suspicious", "medium", "Leetspeak obfuscation of injection keywords"),
 ]
 
@@ -170,10 +170,14 @@
     # ``[config](https://.../system-setup)`` triggered. Real smuggled-
     # instruction attacks include the full "ignore (all|the|previous|prior)"
     # phrasing in the URL/anchor.
+    # Negated, bounded char classes ([^\]] / [^)]) instead of ``.*?`` so the
+    # regex is linear -- the lazy-dot form could backtrack on long unclosed
+    # link/URL spans (ReDoS-prone).
     PatternDefinition(
         "markdown_hidden_instruction",
         re.compile(
-            r"\[.*?\]\(.*?(?:ignore|disregard|forget|override)\W+(?:all|the|previous|prior)\W+.*?\)",
+            r"\[[^\]\n]{0,200}\]\([^)\n]{0,300}(?:ignore|disregard|forget|override)\W{1,8}"
+            r"(?:all|the|previous|prior)\W{1,8}[^)\n]{0,300}\)",
             re.I,
         ),
         "structural",