fix: harden PII canary detection with portmanteau middle names and sub-pattern scanning

- Replace real-word middle names with portmanteaus (e.g. Thundaze, Lunarex) that
  cannot appear in legitimate output but survive canonicalization as scan tokens
- Extract CANARY_DOMAIN constant and add sub_pattern_map() to inject the email
  domain and each middle name as additional Aho-Corasick patterns in both the
  daemon scanner and the default (library/test) scanner -- catches leaks where an
  LLM reproduces just the domain or a distinctive middle name without the full value
- Promote canary_paraphrase advisory to block when k_observed >= 10x k_threshold,
  leaving the advisory tier for early warning and reserving block for overwhelming
  n-gram accumulation (195 distinct n-grams on a full canary value in isolation)
- Update TC-066-06 to accept block at high n-gram counts (reflects new behavior;
  exact scanner still fires first in the real pipeline)

Author: tkdtaylor

src/armor/canaries/_generate.py

@@ -60,35 +60,35 @@
     "Zoe",
 ]
 _PII_MIDDLE_NAMES = [
-    "Lightning",
-    "Thunder",
-    "Blaze",
-    "Storm",
-    "Iron",
-    "Silver",
-    "Crimson",
-    "Shadow",
-    "Frost",
-    "Ember",
-    "Quantum",
-    "Cobalt",
-    "Titan",
-    "Inferno",
-    "Steel",
-    "Jade",
-    "Onyx",
-    "Ruby",
-    "Zenith",
-    "Solar",
-    "Lunar",
-    "Apex",
-    "Hyper",
-    "Turbo",
-    "Neon",
-    "Vortex",
-    "Plasma",
-    "Atomic",
-    "Primal",
+    "Thundaze",
+    "Lightnorm",
+    "Silvrost",
+    "Irember",
+    "Crimadow",
+    "Quantobalt",
+    "Titaferno",
+    "Stelade",
+    "Rubnyx",
+    "Zenolar",
+    "Lunarex",
+    "Atomasma",
+    "Vorteon",
+    "Hyturbo",
+    "Primblaze",
+    "Storember",
+    "Silcrim",
+    "Frostitan",
+    "Bladow",
+    "Ironyx",
+    "Balvorn",
+    "Thunderal",
+    "Quantnox",
+    "Jadember",
+    "Plasolar",
+    "Hypferno",
+    "Lunstorm",
+    "Apexmal",
+    "Rubenith",
 ]
 _PII_LAST_NAMES = [
     "Dragon",
@@ -186,6 +186,25 @@
     ("Beldock", "NB", "E"),
 ]
 
+CANARY_DOMAIN = "armor-trap.invalid"
+
+
+def sub_pattern_map() -> dict[str, str]:
+    """Return additional Aho-Corasick patterns that are sub-components of generated canary values.
+
+    These supplement the full-value canary scanner with high-signal fragments that
+    survive mild reformatting (e.g. an LLM that omits the email local-part but
+    reproduces the domain, or outputs a distinctive middle name without the full name).
+
+    Each key is a synthetic canary ID (never a real catalogue entry).
+    """
+    patterns: dict[str, str] = {
+        "canary.sub:domain": CANARY_DOMAIN,
+    }
+    for name in _PII_MIDDLE_NAMES:
+        patterns[f"canary.sub:middle:{name.lower()}"] = name
+    return patterns
+
 
 def _generate_pii_value(marker_rule: str) -> str:
     """Generate a fake PII value for a pii: prefixed marker rule.
@@ -346,7 +365,7 @@ def _generate_value_for_pattern(marker_rule: str) -> str:
     # Fake URLs: https://canary.armor-trap.invalid/<id>
     if marker_rule == r"^https://canary\.armor-trap\.invalid/[a-z0-9\-]+$":
         suffix = "".join(random.choice(string.ascii_lowercase + string.digits + "-") for _ in range(12))
-        return f"https://canary.armor-trap.invalid/{suffix}"
+        return f"https://canary.{CANARY_DOMAIN}/{suffix}"
 
     # Slack webhook URL
     if marker_rule == r"^https://hooks\.slack\.com/services/T[A-Z0-9]+/B[A-Z0-9]+/[A-Za-z0-9]+$":
@@ -369,12 +388,12 @@ def _generate_value_for_pattern(marker_rule: str) -> str:
     # Fake hostnames: <id>.canary.armor-trap.invalid
     if marker_rule == r"^[a-z0-9\-]+\.canary\.armor-trap\.invalid$":
         prefix = "".join(random.choice(string.ascii_lowercase + string.digits + "-") for _ in range(12))
-        return f"{prefix}.canary.armor-trap.invalid"
+        return f"{prefix}.canary.{CANARY_DOMAIN}"
 
     # Fake email addresses: canary-<id>@armor-trap.invalid
     if marker_rule == r"^canary-[a-z0-9\-]+@armor-trap\.invalid$":
         suffix = "".join(random.choice(string.ascii_lowercase + string.digits + "-") for _ in range(12))
-        return f"canary-{suffix}@armor-trap.invalid"
+        return f"canary-{suffix}@{CANARY_DOMAIN}"
 
     # Fake wallet addresses: 1ARMORTRAP + 32 hex chars
     if marker_rule == r"^1ARMORTRAP[0-9a-f]{32}$":
@@ -419,7 +438,7 @@ def _generate_value_for_pattern(marker_rule: str) -> str:
         password = "".join(random.choice(string.ascii_letters + string.digits) for _ in range(16))
         host = "canary-" + "".join(random.choice(string.ascii_lowercase + string.digits) for _ in range(8))
         db_name = "canary_db"
-        return f"{db_type}://{username}:{password}@{host}.armor-trap.invalid/{db_name}"
+        return f"{db_type}://{username}:{password}@{host}.{CANARY_DOMAIN}/{db_name}"
 
     # BIP39 seed (12-word phrase): [a-z]+ (space [a-z]+){11}
     if marker_rule == r"^[a-z]+( [a-z]+){11}$":

src/armor/daemon/server.py

@@ -18,6 +18,7 @@
 from pathlib import Path
 from typing import Any
 
+from armor.canaries._generate import sub_pattern_map
 from armor.canaries.catalogue import Catalogue
 from armor.canaries.scanner import CanaryScanner
 from armor.daemon.honeypot_gate import should_invoke_honeypot
@@ -188,8 +189,10 @@ def __init__(
                     logger.error("Catalogue is empty (no active canaries)")
                     sys.exit(78)
 
-                # Build canary value map for the scanner
+                # Build canary value map for the scanner, then extend with sub-patterns
+                # (email domain, distinctive middle names) that survive mild LLM reformatting.
                 canary_map = {entry.canary_id: entry.value for entry in active_canaries}
+                canary_map.update(sub_pattern_map())
                 self.canary_scanner = CanaryScanner(canary_map)
                 logger.info(f"Loaded canary catalogue with {len(active_canaries)} active canaries")
 

src/armor/detectors/canary_paraphrase.py

@@ -167,19 +167,29 @@ def check(self, payload: Payload, ctx: SessionContext) -> Verdict:
             for canary_id, distinct_ngrams in distinct_ngrams_per_canary.items():
                 k_observed = len(distinct_ngrams)
                 if k_observed >= self.k_threshold:
-                    # Calculate confidence: min(1.0, K_observed / K_threshold * 0.5)
                     confidence = min(1.0, (k_observed / self.k_threshold) * 0.5)
+                    details: dict[str, object] = {
+                        "canary_id": canary_id,
+                        "ngram_count": k_observed,
+                        "k_threshold": self.k_threshold,
+                        "confidence": confidence,
+                    }
+                    signal_id = f"canary.paraphrase:{canary_id}:ngram"
+                    message = f"Potential paraphrased canary leak detected (n-grams: {k_observed})"
+
+                    if k_observed >= self.k_threshold * 10:
+                        return Verdict.block_verdict(
+                            signal_id=signal_id,
+                            message="Output suppressed by armor.",
+                            severity="critical",
+                            details=details,
+                        )
 
                     return Verdict.advisory_verdict(
-                        signal_id=f"canary.paraphrase:{canary_id}:ngram",
+                        signal_id=signal_id,
                         severity="high",
-                        message=f"Potential paraphrased canary leak detected (n-grams: {k_observed})",
-                        details={
-                            "canary_id": canary_id,
-                            "ngram_count": k_observed,
-                            "k_threshold": self.k_threshold,
-                            "confidence": confidence,
-                        },
+                        message=message,
+                        details=details,
                     )
 
             return Verdict.pass_verdict()

src/armor/detectors/canary_scanner.py

@@ -12,7 +12,7 @@
 from functools import lru_cache
 from pathlib import Path
 
-from armor.canaries._generate import write_values_file
+from armor.canaries._generate import sub_pattern_map, write_values_file
 from armor.canaries.catalogue import Catalogue
 from armor.canaries.scanner import CanaryScanner
 from armor.types import Payload, SessionContext, Verdict
@@ -43,6 +43,7 @@ def _default_scanner() -> CanaryScanner:
         # Load with the generated values
         catalogue = Catalogue.load(temp_values_path)
         canary_map = {entry.canary_id: entry.value for entry in catalogue.active_canaries()}
+        canary_map.update(sub_pattern_map())
         return CanaryScanner(canary_map)
     finally:
         # Clean up temp file

tests/unit/detectors/test_canary_paraphrase.py

@@ -151,12 +151,12 @@ def test_tc_066_05_benign_short_substring(
     def test_tc_066_06_full_match_not_paraphrase(
         self, detector: CanaryParaphraseDetector, context: SessionContext
     ) -> None:
-        """TC-066-06: Full canary value should not be detected as paraphrase (out of scope).
+        """TC-066-06: Full canary value in rolling buffer triggers block at high n-gram count.
 
-        This is a regression check: the paraphrase detector should not fire on full values
-        (that's the canary_scanner's job). However, since the paraphrase detector does
-        scan n-grams, a full value will match multiple n-grams. This test documents
-        that behavior.
+        A full canary value produces far more distinct n-grams than the 10x block
+        threshold (k_threshold=3, k_block=30). The paraphrase detector correctly
+        blocks in this case. In the real pipeline the canary_scanner fires first,
+        so this code path is only exercised when the detector runs in isolation.
         """
         buf = context.rolling_buffer
         assert buf is not None
@@ -166,12 +166,10 @@ def test_tc_066_06_full_match_not_paraphrase(
         payload = Payload(text="")
         verdict = detector.check(payload, context)
 
-        # The paraphrase detector will likely fire on the full value
-        # (since it contains many n-grams). This is acceptable — the
-        # canary_scanner will block it first in the pipeline anyway.
-        # We just document that the paraphrase detector is not suppressed
-        # on full values (it's not its job to be smart about that).
-        assert verdict.decision in ("pass", "advisory")
+        # 195 distinct n-grams >> 30 (10x k_threshold) -> block.
+        # In the real pipeline the exact canary_scanner fires first; the
+        # paraphrase detector correctly escalates when it runs in isolation.
+        assert verdict.decision in ("advisory", "block")
 
     # TC-066-07: Confidence formula K=4, K_threshold=3
     def test_tc_066_07_confidence_formula(self, detector: CanaryParaphraseDetector, context: SessionContext) -> None: