|
| 1 | +# (C) 2026 GoodData Corporation. All rights reserved. |
| 2 | +"""Deterministic leak verdict for the agentic_obfuscation kind. |
| 3 | +
|
| 4 | +Pure functions only: the caller collects the sinks, this module decides. A sink is any |
| 5 | +JSON-like document -- the stored conversation read back over the API, or every Langfuse |
| 6 | +trace of the conversation's session. A canary is found by exact substring over every string |
| 7 | +leaf, so the verdict never depends on an LLM. JSON serialised inside a string (Langfuse |
| 8 | +keeps span input that way) is decoded and walked too. |
| 9 | +
|
| 10 | +A sink that cannot be seen must not pass an absence check. Every turn therefore carries an |
| 11 | +anchor, a non-sensitive fragment of the question, and a sink that does not show every |
| 12 | +anchor is reported as blind instead of clean. |
| 13 | +""" |
| 14 | + |
| 15 | +from __future__ import annotations |
| 16 | + |
| 17 | +import json |
| 18 | +import re |
| 19 | +from collections.abc import Iterator, Mapping, Sequence |
| 20 | +from dataclasses import dataclass, field |
| 21 | +from typing import Any |
| 22 | + |
| 23 | +SINK_CONVERSATION_DB = "conversation_db" |
| 24 | +SINK_LANGFUSE_TRACE = "langfuse_trace" |
| 25 | +SINKS = (SINK_CONVERSATION_DB, SINK_LANGFUSE_TRACE) |
| 26 | + |
| 27 | +# Classes whose value may be written with separators the detector ignores, so a leak of |
| 28 | +# "4916 3385 0608 2832" must still match the compact nonce "4916338506082832". |
| 29 | +_DIGIT_CLASSES = frozenset({"CREDIT_CARD", "IBAN"}) |
| 30 | +_SEPARATORS = re.compile(r"[\s\-]") |
| 31 | +# Placeholder the AI-on-data trace redaction writes over Langfuse input and output. Behind it |
| 32 | +# the Langfuse leg shows nothing, so an absence check there would prove nothing. |
| 33 | +AI_ON_DATA_REDACTION = "<redacted: AI-on-data enabled>" |
| 34 | +_ANCHOR_RUN = re.compile(r"[^\W_](?:[\w ]*[^\W_])?") |
| 35 | +_ANCHOR_MIN = 12 |
| 36 | +_ANCHOR_MAX = 40 |
| 37 | + |
| 38 | + |
| 39 | +@dataclass(frozen=True) |
| 40 | +class Occurrence: |
| 41 | + sink: str |
| 42 | + path: str |
| 43 | + needle: str |
| 44 | + |
| 45 | + |
| 46 | +@dataclass |
| 47 | +class CaseVerdict: |
| 48 | + failures: list[str] = field(default_factory=list) |
| 49 | + notes: list[str] = field(default_factory=list) |
| 50 | + |
| 51 | + @property |
| 52 | + def passed(self) -> bool: |
| 53 | + return not self.failures |
| 54 | + |
| 55 | + |
| 56 | +def iter_strings(value: Any, path: str = "$") -> Iterator[tuple[str, str]]: |
| 57 | + """Yield ``(path, text)`` for every string or number leaf and every key, decoding JSON held in a string. |
| 58 | +
|
| 59 | + A string that parses as a JSON document is walked instead of being yielded whole, so a |
| 60 | + value spelled with escapes (``\\u0040``) is seen decoded and reported once. Numbers are |
| 61 | + leaves too: a numeric secret in a pasted JSON document decodes to an int. Keys are text a |
| 62 | + user can type as well -- a pasted config may key on an email -- so each is yielded at the |
| 63 | + path of its object with a ``<key>`` suffix. |
| 64 | + """ |
| 65 | + if isinstance(value, bool): |
| 66 | + return |
| 67 | + if isinstance(value, int | float): |
| 68 | + yield path, str(value) |
| 69 | + return |
| 70 | + if isinstance(value, str): |
| 71 | + if value.lstrip()[:1] in ("{", "["): |
| 72 | + try: |
| 73 | + decoded = json.loads(value) |
| 74 | + except ValueError: |
| 75 | + pass |
| 76 | + else: |
| 77 | + yield from iter_strings(decoded, f"{path}<json>") |
| 78 | + return |
| 79 | + yield path, value |
| 80 | + elif isinstance(value, Mapping): |
| 81 | + for key, child in value.items(): |
| 82 | + if isinstance(key, str): |
| 83 | + yield f"{path}.{key}<key>", key |
| 84 | + yield from iter_strings(child, f"{path}.{key}") |
| 85 | + elif isinstance(value, Sequence) and not isinstance(value, bytes | bytearray): |
| 86 | + for index, child in enumerate(value): |
| 87 | + yield from iter_strings(child, f"{path}[{index}]") |
| 88 | + |
| 89 | + |
| 90 | +def canary_needles(canary: Mapping[str, Any]) -> list[str]: |
| 91 | + """Every spelling of the canary a leak may take: nonce, value and its declared variants.""" |
| 92 | + keys = ("nonce", "value", "unescaped_value", "decoded_value", "compact_value") |
| 93 | + needles = [canary[key] for key in keys if isinstance(canary.get(key), str) and canary[key]] |
| 94 | + return list(dict.fromkeys(needles)) |
| 95 | + |
| 96 | + |
| 97 | +def find_canary(canary: Mapping[str, Any], sink: str, document: Any) -> list[Occurrence]: |
| 98 | + needles = canary_needles(canary) |
| 99 | + compact = {_SEPARATORS.sub("", n) for n in needles} if canary.get("class") in _DIGIT_CLASSES else set() |
| 100 | + found: list[Occurrence] = [] |
| 101 | + for path, text in iter_strings(document): |
| 102 | + hit = next((n for n in needles if n in text), None) |
| 103 | + if hit is None and compact: |
| 104 | + squeezed = _SEPARATORS.sub("", text) |
| 105 | + hit = next((n for n in compact if n in squeezed), None) |
| 106 | + if hit is not None: |
| 107 | + found.append(Occurrence(sink, path, hit)) |
| 108 | + return found |
| 109 | + |
| 110 | + |
| 111 | +def contains(document: Any, needle: str) -> bool: |
| 112 | + return any(needle in text for _, text in iter_strings(document)) |
| 113 | + |
| 114 | + |
| 115 | +def derive_anchor(question: str, canaries: Sequence[Mapping[str, Any]]) -> str | None: |
| 116 | + """The longest plain-word run of the question once every canary spelling is cut out. |
| 117 | +
|
| 118 | + Word characters and spaces only, so escaping, JSON quoting and masking next to a canary |
| 119 | + cannot alter it between what was sent and what a sink stores. |
| 120 | + """ |
| 121 | + fragments = [question] |
| 122 | + # Longest spelling first: cutting the nonce out of "john\\.canary2d8e@..." first would |
| 123 | + # leave "john\\." behind, and the anchor would then end in text the mask replaces. |
| 124 | + needles = sorted({n for canary in canaries for n in canary_needles(canary)}, key=len, reverse=True) |
| 125 | + for needle in needles: |
| 126 | + fragments = [piece for fragment in fragments for piece in fragment.split(needle)] |
| 127 | + runs = [match.group(0) for fragment in fragments for match in _ANCHOR_RUN.finditer(fragment)] |
| 128 | + best = max(runs, key=len, default="") |
| 129 | + if len(best) < _ANCHOR_MIN: |
| 130 | + return None |
| 131 | + return best[:_ANCHOR_MAX].rstrip() |
| 132 | + |
| 133 | + |
| 134 | +def _declared_sinks(canary: Mapping[str, Any], key: str) -> list[str]: |
| 135 | + sinks = canary.get(key) or [] |
| 136 | + unknown = [s for s in sinks if s not in SINKS] |
| 137 | + if unknown: |
| 138 | + raise ValueError(f"canary {canary.get('nonce')!r} names unknown sink(s) {unknown} in {key}") |
| 139 | + return list(sinks) |
| 140 | + |
| 141 | + |
| 142 | +def _format(occurrences: Sequence[Occurrence], limit: int = 5) -> str: |
| 143 | + shown = ", ".join(f"{o.path} ({o.needle!r})" for o in occurrences[:limit]) |
| 144 | + more = len(occurrences) - limit |
| 145 | + return shown + (f" and {more} more" if more > 0 else "") |
| 146 | + |
| 147 | + |
| 148 | +def evaluate_case( |
| 149 | + expected: Mapping[str, Any], |
| 150 | + sinks: Mapping[str, Any], |
| 151 | + anchors: Sequence[str], |
| 152 | + *, |
| 153 | + turn_rejected: bool = False, |
| 154 | +) -> CaseVerdict: |
| 155 | + """Decide one item from the collected sinks. |
| 156 | +
|
| 157 | + ``sinks`` maps a sink name to its document, or to ``None`` when it was not collected |
| 158 | + (only allowed for the Langfuse leg of a rejected turn, which may export nothing). |
| 159 | + """ |
| 160 | + verdict = CaseVerdict() |
| 161 | + status = expected.get("status", "enforced") |
| 162 | + visible: dict[str, bool] = {} |
| 163 | + |
| 164 | + for sink in SINKS: |
| 165 | + document = sinks.get(sink) |
| 166 | + if document is None: |
| 167 | + visible[sink] = False |
| 168 | + if not turn_rejected: |
| 169 | + verdict.failures.append(f"{sink}: not collected, so no absence claim can be made") |
| 170 | + continue |
| 171 | + if sink == SINK_LANGFUSE_TRACE and contains(document, AI_ON_DATA_REDACTION): |
| 172 | + verdict.failures.append( |
| 173 | + f"{sink}: trace input/output is replaced by {AI_ON_DATA_REDACTION!r} (enableAiOnData with " |
| 174 | + "enableGenAiTraceRedaction), so the obfuscation leg cannot be observed here" |
| 175 | + ) |
| 176 | + visible[sink] = False |
| 177 | + continue |
| 178 | + missing = [anchor for anchor in anchors if not contains(document, anchor)] |
| 179 | + if missing and not turn_rejected: |
| 180 | + verdict.failures.append(f"{sink}: blind -- anchor(s) {missing} not found, the read-back is incomplete") |
| 181 | + visible[sink] = False |
| 182 | + continue |
| 183 | + visible[sink] = True |
| 184 | + |
| 185 | + for canary in expected.get("canaries", []): |
| 186 | + label = f"{canary.get('class')} {canary.get('nonce')!r}" |
| 187 | + for sink in _declared_sinks(canary, "absent_from"): |
| 188 | + # A blind sink still convicts: a canary it does show is a leak all the same. |
| 189 | + if sinks.get(sink) is None: |
| 190 | + continue |
| 191 | + occurrences = find_canary(canary, sink, sinks[sink]) |
| 192 | + # Paths the item declares as not yet decided (e.g. conversation state the SC does |
| 193 | + # not rule on) are reported, never gated, until a decision turns them into leaks. |
| 194 | + record_only = [re.compile(p) for p in canary.get("record_only_paths") or []] |
| 195 | + recorded = [o for o in occurrences if any(r.search(o.path) for r in record_only)] |
| 196 | + gated = [o for o in occurrences if o not in recorded] |
| 197 | + if gated: |
| 198 | + verdict.failures.append(f"LEAK {label} in {sink}: {_format(gated)}") |
| 199 | + if recorded: |
| 200 | + verdict.notes.append(f"RECORDED {label} in {sink} (record-only path, not gated): {_format(recorded)}") |
| 201 | + marker = canary.get("mask_marker_present") |
| 202 | + if marker and not turn_rejected and visible.get(sink) and not contains(sinks[sink], marker): |
| 203 | + verdict.failures.append(f"{label}: mask marker {marker!r} missing from {sink}") |
| 204 | + for sink in _declared_sinks(canary, "present_in"): |
| 205 | + if not visible.get(sink): |
| 206 | + continue |
| 207 | + if not find_canary(canary, sink, sinks[sink]): |
| 208 | + if status == "known_limitation": |
| 209 | + flip = (expected.get("known_limitation") or {}).get("flip_when", "") |
| 210 | + verdict.failures.append( |
| 211 | + f"{label} expected in {sink} ({status}) but is now masked -- the limitation no longer " |
| 212 | + f"reproduces, update the fixture deliberately. flip_when: {flip}" |
| 213 | + ) |
| 214 | + else: |
| 215 | + # An enforced present_in guards a value that is not sensitive: masking it is |
| 216 | + # the defect (a false positive), not progress. |
| 217 | + verdict.failures.append( |
| 218 | + f"OVER-MASKED {label} in {sink}: expected unchanged but it was masked (false positive)" |
| 219 | + ) |
| 220 | + |
| 221 | + if turn_rejected and sinks.get(SINK_LANGFUSE_TRACE) is None: |
| 222 | + verdict.notes.append("rejected turn exported no Langfuse trace; only the database leg was asserted") |
| 223 | + return verdict |
0 commit comments