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 **original** tool value; **`sanitized`** in `DefenseResult` is the original content (unchanged by SFE drops)
 - **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`
 
+- **Detect-and-gate (v0.8.0):** defender **never rewrites or redacts** content. `sanitized` is the **original** payload (optionally `[UD-…]` boundary-wrapped); threats are reported as detection evidence and blocking is expressed via `allowed`. **Migration:** if you relied on `sanitized` being redacted, gate on `allowed` instead.
 - 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 |
 |-------|------------------|
@@ -181,7 +184,7 @@ defense = create_prompt_defense(
     enable_tier1=True,
     enable_tier2=True,
     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

src/stackone_defender/classifiers/onnx_classifier.py

@@ -145,8 +145,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

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",

src/stackone_defender/classifiers/tier2_classifier.py

@@ -9,6 +9,7 @@
 
 import json
 import logging
+import math
 import os
 import re
 import time
@@ -185,6 +186,18 @@ def classify(self, text: str) -> Tier2Result:
 
         try:
             main, aux = self._onnx.classify_pair(analysis_text)
+            # A non-finite score (NaN/Infinity) means the model produced no
+            # usable output. Report a SKIP, not score 0 -- score 0 yields
+            # confidence 1.0 (|0 - 0.5| * 2), making a broken inference look
+            # like a max-confidence benign classification.
+            if not math.isfinite(main):
+                return Tier2Result(
+                    score=0,
+                    confidence=0,
+                    skipped=True,
+                    skip_reason="Non-finite model output (NaN/Infinity)",
+                    latency_ms=_ms(start),
+                )
             confidence = abs(main - 0.5) * 2
             return Tier2Result(
                 score=main, confidence=confidence, skipped=False, latency_ms=_ms(start), aux=aux

src/stackone_defender/config.py

@@ -65,6 +65,11 @@
     "patterns_fraction": 0.25,
 }
 
+# Per-field cap (characters) on the text Tier 1 runs heavy regex / encoding
+# detection over. ReDoS guard — content past the cap is not analysed and
+# metadata.analysis_truncated is set. The full content is still returned.
+DEFAULT_MAX_FIELD_ANALYSIS_LENGTH = 50000
+
 DEFAULT_TIER2_CONFIG = Tier2Config(
     high_risk_threshold=0.8,
     medium_risk_threshold=0.5,

src/stackone_defender/core/prompt_defense.py

@@ -40,6 +40,10 @@
 
 _logger = logging.getLogger(__name__)
 
+# Module-scoped (not per-instance): PromptDefense is constructed per-request in
+# some hosts, so an instance flag would warn at full request volume.
+_tier2_unavailable_warned = False
+
 _DEFAULT_TIER3_BAND = Tier3EscalationBand(lower=0.3, upper=0.85)
 _DEFAULT_TIER3_MAX_TEXT_LENGTH = 10000
 
@@ -85,6 +89,8 @@ class _Tier2Outcome:
     phase_timings: PhaseTimings | None = None
     tier2_stats: Tier2Stats | None = None
     cold_load: bool | None = None
+    # False when Tier 2 was enabled but the model/runtime failed to load.
+    tier2_available: bool | None = None
 
 
 def _extract_strings(
@@ -183,15 +189,17 @@ def __init__(
         tier2_fields: list[str] | None = None,
         use_sfe: bool | dict[str, Any] = False,
         block_high_risk: bool = False,
-        default_risk_level: RiskLevel = "medium",
+        default_risk_level: RiskLevel = "low",
         annotate_boundary: bool = False,
+        require_tier2: bool = False,
         enable_tier3: bool = False,
         defender_mode: DefenderMode = "cascade",
         tier3: dict[str, Any] | None = None,
     ):
         self._config: PromptDefenseConfig = create_config(config)
         if block_high_risk:
             self._config.block_high_risk = True
+        self._tier2_required = require_tier2
 
         self._tier2_fields = tier2_fields
         self._sfe_enabled = False
@@ -211,7 +219,6 @@ def __init__(
             traversal=self._config.traversal,
             default_risk_level=default_risk_level,
             use_tier1_classification=enable_tier1,
-            block_high_risk=block_high_risk,
             cumulative_risk_thresholds=self._config.cumulative_risk_thresholds,
             annotate_boundary=annotate_boundary,
         )
@@ -314,6 +321,34 @@ def warmup_tier2(self) -> None:
     def is_tier2_ready(self) -> bool:
         return self._tier2.is_ready() if self._tier2 else False
 
+    def _handle_tier2_unavailable(self, err: Exception) -> None:
+        """Tier 2 enabled but the model/runtime failed to load. Fail closed when
+        require_tier2 is set; otherwise warn once per process and continue
+        Tier-1-only (fail open)."""
+        if self._tier2_required:
+            raise RuntimeError(
+                f"[defender] Tier 2 is required (require_tier2=True) but the model/runtime "
+                f"failed to load: {err}. Install the optional dependencies with "
+                f"`pip install stackone-defender[onnx]`."
+            )
+        global _tier2_unavailable_warned
+        if not _tier2_unavailable_warned:
+            _tier2_unavailable_warned = True
+            _logger.warning(
+                "[defender] Tier 2 unavailable (model/runtime failed to load); "
+                "continuing Tier-1-only. Reason: %s",
+                err,
+            )
+
+    @staticmethod
+    def _coverage_degraded(metadata: Any, depth_flag: dict[str, bool]) -> bool | None:
+        """True when Tier 1 detection coverage was reduced (depth/size limit hit,
+        or a wide payload's analysis was capped). Content is still returned in full."""
+        sm = metadata.size_metrics
+        if depth_flag.get("hit") or metadata.analysis_truncated or sm.size_limit_hit or sm.depth_limit_hit:
+            return True
+        return None
+
     def _resolve_tier3_provider(self) -> Tier3Provider | None:
         return self._tier3_custom_provider or get_default_tier3_provider()
 
@@ -535,6 +570,7 @@ async def _run_tier3_only(
             fields_dropped=[],
             truncated_at_depth=depth_flag["hit"] or None,
             latency_ms=(time.perf_counter() - start_time) * 1000,
+            coverage_degraded=self._coverage_degraded(sanitized.metadata, depth_flag),
         )
 
     def defend_tool_result(self, value: Any, tool_name: str) -> DefenseResult:
@@ -662,6 +698,8 @@ async def _defend_tool_result_async_impl(
             tier2_stats=tier2.tier2_stats,
             tier1_ms=tier1_ms,
             cold_load=tier2.cold_load,
+            tier2_available=tier2.tier2_available,
+            coverage_degraded=self._coverage_degraded(sanitized.metadata, depth_flag),
         )
 
     def _defend_tool_result_sync(
@@ -777,6 +815,8 @@ def _defend_tool_result_sync(
             tier2_stats=tier2.tier2_stats,
             tier1_ms=tier1_ms,
             cold_load=tier2.cold_load,
+            tier2_available=tier2.tier2_available,
+            coverage_degraded=self._coverage_degraded(sanitized.metadata, depth_flag),
         )
 
     # ------------------------------------------------------------------
@@ -802,6 +842,18 @@ def _evaluate_tier2(
         """
         out = _Tier2Outcome()
 
+        # Sample cold-start BEFORE warmup, then load the model. A load failure
+        # (missing optional deps) is a hard "Tier 2 unavailable" — fail closed when
+        # require_tier2, else warn once and continue Tier-1-only.
+        was_cold = not tier2.is_ready()
+        try:
+            tier2.warmup()
+        except Exception as e:
+            out.tier2_available = False
+            out.skip_reason = f"Tier 2 unavailable (model/runtime failed to load): {e}"
+            self._handle_tier2_unavailable(e)
+            return out
+
         fields_for_tier2 = (
             self._tier2_fields
             if self._tier2_fields is not None
@@ -836,7 +888,7 @@ def _evaluate_tier2(
         t_infer_start = time.perf_counter()
         # Set now (before the failure early-return) so cold_load is a bool
         # whenever inference was attempted — success or failure (TS 0.7.4 parity).
-        out.cold_load = not tier2.is_ready()
+        out.cold_load = was_cold
         stats = BatchTokenStats()
         multihead_cfg = tier2.get_multihead_config()
         all_scores, all_pairs, infer_skip, unique_count = self._tier2_run_inference(