#!/usr/bin/env python3 """semantic_merge.py — band-1 finding dedup, the #17a semantic-merge leg. PROBLEM (MEASURED, docs/validation + zync STEP2): the k-roll union in gate.py dedupes by normalized-title word-overlap (union_rolls). Independent rolls paraphrase the SAME bug in different words; the lexical overlap falls below threshold, so one canonical fragments into several "1/3" groups. A bug all 3 rolls found reads as flaky 1/3 -> the recall signal is corrupted (S6-001: SSRF-on-PATCH surfaced as THREE 1/3 entries; true union = 3/3). THIS LEG: a semantic merge pass over the already-unioned groups. Groups that describe the SAME vulnerability (same location + same root cause + same fix) collapse into one; the merged group's roll set is the UNION of its members' rolls -> the recall count is restored. SAFETY (a gate's asymmetric harm): over-merge HIDES a real bug (catastrophic); under-merge only leaves noise (safe). So the merge is CONSERVATIVE / under-merge-biased, and it FAILS LOUD: on any LLM/parse/partition error it returns the input groups UNCHANGED with degraded=True — it never silently drops or invents a finding (same non-silent-on-degraded discipline as the oracle). merge_groups(groups, ...) consumes and returns gate.py's group dicts: {"title", "sev", "rolls":set}. """ import argparse, json, os, re, sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from llm_runner import run_llm # noqa: E402 SEV_RANK = {"critical": 0, "high": 1, "medium": 2, "low": 3, "unrated": 4} class MergeError(Exception): """LLM call, JSON parse, or partition check failed — caller degrades safely, non-silent.""" def _build_prompt(titles): numbered = "\n".join(f"{i}. [{sev}] {t}" for i, (t, sev) in enumerate(titles, 1)) n = len(titles) return f"""You are deduplicating one security review's findings list. Independent review passes ran over the SAME file, so MANY entries are PARAPHRASES: the same underlying vulnerability worded differently across passes. Group findings that describe the SAME vulnerability — they must share ALL THREE of: the same code location, the same root cause, AND the same fix. Two findings that share keywords but differ in location, root cause, or fix are DIFFERENT bugs and stay SEPARATE (e.g. "missing SSRF guard on the PATCH route" and "DNS-rebinding TOCTOU inside the SSRF guard" are TWO bugs — different root cause and fix — never merge them). CONSERVATIVE RULE: when uncertain whether two findings are the same bug, keep them SEPARATE. Merging two distinct bugs hides a real vulnerability and is unacceptable. Under-merging is safe. Findings ({n} total): {numbered} Output ONLY a JSON object, no prose, no markdown fence: {{"groups": [[1,4],[2],[3],...]}} where each inner list holds the 1-based indices of findings that are the SAME vulnerability. Every index from 1 to {n} MUST appear EXACTLY ONCE across all groups.""" def _call_llm(prompt, config_dir, model, effort): try: r = run_llm(prompt, model=model, effort=effort, config_dir=config_dir) except Exception as e: # noqa: BLE001 raise MergeError(f"claude -p failed: {e}") from e if r.returncode != 0: raise MergeError(f"claude -p exit {r.returncode}: {r.stderr.strip()[:200]}") return r.stdout def _parse_groups(text, n): """extract {"groups": [[...]]} and verify it is a PARTITION of 1..n (each index exactly once).""" m = re.search(r"\{.*\"groups\".*\}", text, re.S) if not m: raise MergeError(f"no JSON groups object in LLM output: {text.strip()[:200]}") try: obj = json.loads(m.group(0)) groups = obj["groups"] except (json.JSONDecodeError, KeyError, TypeError) as e: raise MergeError(f"bad JSON groups: {e}") from e seen, flat = set(), [] for grp in groups: if not isinstance(grp, list) or not grp: raise MergeError(f"group not a non-empty list: {grp!r}") for idx in grp: if not isinstance(idx, int) or not (1 <= idx <= n): raise MergeError(f"index out of range 1..{n}: {idx!r}") if idx in seen: raise MergeError(f"index {idx} appears in >1 group — not a partition") seen.add(idx) flat.append([i - 1 for i in grp]) # to 0-based if seen != set(range(1, n + 1)): raise MergeError(f"not all indices covered: missing {set(range(1, n + 1)) - seen}") return flat def _fold(members): """fold a list of group dicts into one: UNION rolls, min-rank (worst) severity, representative title. representative = the member found by the most rolls (ties -> highest severity, then shortest title). Carry the finding-citation anchor through the merge: first non-empty snippet/line_hint across members (gate.py's build_emit_dict reads these POST-merge; dropping them here re-hides the navigable line).""" rolls = set().union(*(g["rolls"] for g in members)) sev = min((g["sev"] for g in members), key=lambda s: SEV_RANK.get(s, 9)) rep = max(members, key=lambda g: (len(g["rolls"]), -SEV_RANK.get(g["sev"], 9), -len(g["title"]))) snippet = next((g["snippet"] for g in members if g.get("snippet")), "") line_hint = next((g["line_hint"] for g in members if g.get("line_hint")), 0) return {"title": rep["title"], "sev": sev, "rolls": rolls, "members": len(members), "snippet": snippet, "line_hint": line_hint} def merge_groups(groups, config_dir=None, model="sonnet", effort="medium"): """semantic-merge gate.py union groups. Returns (merged_groups, degraded). degraded=True => merge failed; merged_groups == input (un-deduped), caller must label it non-silent. A 0/1-finding list needs no merge.""" if len(groups) < 2: return list(groups), False titles = [(g["title"], g["sev"]) for g in groups] try: idx_groups = _parse_groups(_call_llm(_build_prompt(titles), config_dir, model, effort), len(groups)) except MergeError as e: sys.stderr.write(f"[semantic_merge] DEGRADED — {e}\n") return list(groups), True merged = [_fold([groups[i] for i in grp]) for grp in idx_groups] merged.sort(key=lambda g: (SEV_RANK.get(g["sev"], 9), -len(g["rolls"]))) return merged, False # --------------------------------------------------------------------------- # self-test — grounded in the owned, MEASURED S6-001 fixture (zync-pilot/S6_gate_v2.md) # --------------------------------------------------------------------------- # The 17-finding k=3 union the gate produced on endpoints.ts (pre-fix bec8bc8~1). Titles verbatim. S6_FIXTURE = [ ("GET /:id/deliveries/:did ignores the :id path parameter — intra-tenant IDOR", "high"), ("Missing SSRF guard on PATCH /:id — stored URL bypasses all IP-block rules", "high"), ("PATCH endpoint stores arbitrary URLs without SSRF validation — bypass enables internal URL persistence", "high"), ("Retry endpoint has no idempotency guard — concurrent retries produce duplicate outbound dispatches", "high"), ("SSRF bypass via PATCH — assertSafeWebhookUrl never called on URL update", "high"), ("JSON.parse(endpointRaw.secretEncrypted) throws uncaught on malformed or null DB value", "medium"), ("Outbound fetch precedes DB record insertion in both retry and test paths — Worker eviction leaves delivery unrecorded", "medium"), ("POST /:id/deliveries/:did/retry has no status guard — already-delivered events re-dispatched unconditionally", "medium"), ("assertSafeWebhookUrl does not block non-dotted-decimal or octal IPv4 representations", "medium"), ("Rotate-secret has a TOCTOU race — one concurrent caller receives a non-functional plaintext secret", "medium"), ("No rate limiting on test dispatch or manual retry — authenticated HTTP amplification / DoS proxy", "medium"), ("No timeout on outbound fetch — Worker held indefinitely by slow target", "medium"), ("Role check for rotate-secret is case-sensitive only for two variants and casts session to any-shaped object", "low"), ("Test deliveries can be retried as real deliveries — no status guard on retry", "low"), ("Owner role check silently excludes title-cased 'Owner' — unverified cross-file dependency", "low"), ("Fragile role comparison using dual-case strings with an unsafe type cast", "low"), ("DNS rebinding / TOCTOU window in SSRF guard", "low"), ] # canonical clusters (0-based indices) the merge MUST recover. From STEP2 + reading the fixture: SSRF_PATCH = {1, 2, 4} # the three SSRF-on-PATCH paraphrases (STEP2: 3 distinct rolls -> 3/3) ROLE_CASE = {12, 14, 15} # role comparison is case-sensitive, restated 3 ways MUST_STAY_SEPARATE = [ (1, 16), # missing-guard-on-PATCH vs DNS-rebinding-inside-guard : different root cause + fix (1, 8), # missing-guard-on-PATCH vs octal-IPv4-parsing-gap (S6-004 canonical) : different bug (0, 1), # IDOR vs SSRF : different class entirely ] def _unit_checks(): """LLM-free safety units: roll-union recount + the partition guard (fail-loud, never drop a finding) + degraded-fallback. These protect the gate's invariant that merge can NEVER silently lose a bug.""" fails = [] # recount arithmetic: distinct-roll paraphrases UNION to full count toy = [ {"title": "Missing SSRF guard on PATCH", "sev": "high", "rolls": {0}}, {"title": "SSRF bypass via PATCH update", "sev": "high", "rolls": {1}}, {"title": "PATCH stores arbitrary URLs", "sev": "high", "rolls": {2}}, ] folded = _fold(toy) ok = folded["rolls"] == {0, 1, 2} and folded["members"] == 3 print(f"[{'PASS' if ok else 'FAIL'}] recount: 3 distinct-roll paraphrases -> rolls={sorted(folded['rolls'])} (want [0,1,2])") if not ok: fails.append("recount-arithmetic") # severity fold takes the WORST (lowest rank) across members sev_ok = _fold([{"title": "a", "sev": "low", "rolls": {0}}, {"title": "b", "sev": "high", "rolls": {1}}])["sev"] == "high" print(f"[{'PASS' if sev_ok else 'FAIL'}] fold keeps worst severity (low+high -> high)") if not sev_ok: fails.append("sev-fold") # partition guard: every malformed grouping MUST raise (no silent finding-drop / no double-count) bad = { "duplicate-index": '{"groups": [[1,2],[2,3]]}', # idx 2 in two groups "missing-index": '{"groups": [[1,2]]}', # idx 3 dropped "out-of-range": '{"groups": [[1,2,3,4]]}', # n=3, idx 4 invalid "no-json": 'I merged them into two groups.', "empty-group": '{"groups": [[1,2,3],[]]}', } for name, payload in bad.items(): try: _parse_groups(payload, 3) print(f"[FAIL] partition guard let '{name}' through") fails.append(f"guard-{name}") except MergeError: print(f"[PASS] partition guard rejects '{name}'") # well-formed partition parses to 0-based groups try: good = _parse_groups('{"groups": [[1,3],[2]]}', 3) gok = sorted(sorted(g) for g in good) == [[0, 2], [1]] except MergeError: gok = False print(f"[{'PASS' if gok else 'FAIL'}] partition guard accepts a valid partition") if not gok: fails.append("guard-valid") # degraded-fallback: a merge failure => return input UNCHANGED + degraded=True (non-silent). # monkeypatch _call_llm to force the failure deterministically (no network call). global _call_llm real = _call_llm _call_llm = lambda *a, **k: (_ for _ in ()).throw(MergeError("forced for test")) try: src = [{"title": "x", "sev": "high", "rolls": {0}}, {"title": "y", "sev": "low", "rolls": {1}}] merged, degraded = merge_groups(src) dok = degraded and len(merged) == len(src) and merged == src finally: _call_llm = real print(f"[{'PASS' if dok else 'FAIL'}] degraded-fallback: merge failure returns input unchanged, degraded={degraded}") if not dok: fails.append("degraded-fallback") return fails def _selftest(config_dir, k): """GROUPING validation on the real fixture, k rolls -> stability; + the LLM-free safety units.""" print(f"# semantic_merge self-test — S6-001 fixture, {len(S6_FIXTURE)} findings, k={k}\n") fails = _unit_checks() print() # --- grouping validation on the real fixture, k independent merge rolls --- # build index-tagged input so we can recover membership deterministically after merge groups = [{"title": t, "sev": s, "rolls": {0}, "_i": i} for i, (t, s) in enumerate(S6_FIXTURE)] runs = [] for r in range(k): titles = [(g["title"], g["sev"]) for g in groups] try: idx_groups = _parse_groups( _call_llm(_build_prompt(titles), config_dir, "sonnet", "medium"), len(groups)) except MergeError as e: print(f"[FAIL] roll {r}: merge errored — {e}") fails.append(f"roll{r}-error") continue # map each fixture index -> its cluster (frozenset of co-members) for this roll cluster_of = {} for grp in idx_groups: fs = frozenset(grp) for i in grp: cluster_of[i] = fs runs.append(cluster_of) # per-roll assertions ssrf_ok = all(cluster_of[i] == cluster_of[1] for i in SSRF_PATCH) and SSRF_PATCH <= set(cluster_of[1]) role_ok = all(cluster_of[i] == cluster_of[12] for i in ROLE_CASE) and ROLE_CASE <= set(cluster_of[12]) sep_ok = all(cluster_of[a] != cluster_of[b] for a, b in MUST_STAY_SEPARATE) print(f" roll {r}: groups={len(idx_groups)} ssrf-patch-merged={ssrf_ok} role-case-merged={role_ok} distinct-kept={sep_ok}") if not ssrf_ok: fails.append(f"roll{r}-ssrf") if not role_ok: fails.append(f"roll{r}-role") if not sep_ok: fails.append(f"roll{r}-separate") # --- stability across rolls: SSRF-PATCH + role-case clustering identical every roll --- if len(runs) >= 2: stable = all( runs[0].get(1) == r.get(1) and runs[0].get(12) == r.get(12) for r in runs[1:]) print(f"\n[{'PASS' if stable else 'WARN'}] cross-roll stability of the two canonical clusters: {stable}") if not stable: fails.append("unstable") # a flaky merge tool is itself a defect print(f"\n{'PASS — #17a mechanism validated' if not fails else 'FAIL: ' + ', '.join(fails)}") return 1 if fails else 0 def main(): ap = argparse.ArgumentParser() ap.add_argument("--selftest", action="store_true", help="LLM-free units + k-roll grouping on S6 fixture") ap.add_argument("--unittest", action="store_true", help="LLM-free safety units only (recount, partition guard, fallback)") ap.add_argument("--config-dir", default=os.environ.get("SG_CONFIG_DIR")) ap.add_argument("--k", type=int, default=3) a = ap.parse_args() if a.unittest: fails = _unit_checks() print(f"\n{'PASS — safety units' if not fails else 'FAIL: ' + ', '.join(fails)}") sys.exit(1 if fails else 0) if a.selftest: sys.exit(_selftest(a.config_dir, a.k)) ap.error("need --selftest or --unittest (library: import merge_groups)") if __name__ == "__main__": main()