1"""3b. SAT collision search on r-step SHA-256 compression (question C2/C4).23Instance: two message blocks M, M' (fresh variables each) with M != M' (OR of the XOR differences over the words4W_0..W_{min(r,16)-1} that an r-step compression reads; unused words are pinned to 0), equal r-step output. Variants: feedforward in {False, True} x iv in {standard, free} (semi-free-start: IV is 8 free5words shared by both messages). Note that with a *shared* IV the feed-forward adds the same value to both sides,6so a collision with feed-forward is mathematically the same event as without; the two CNFs nevertheless differ7(extra adders), which is what the SAT comparison measures. Found collisions are verified with8wp34.sha256.compress and saved in hex. Solvers: CaDiCaL (pysat) and Kissat; timeout 600 s per cell.9Modes as in preimage_sat.py (--list-cells / --cell / --aggregate).10"""11from __future__ import annotations12import argparse, json, pathlib, random, time13from common import S, provenance, peak_rss_mb, MASK14import cnf_sha256 as C15import solvers as SV1617R_LIST = [4, 6, 8, 10, 12, 14, 16, 18, 20, 22, 24]18TIMEOUT = 600.019VARIANTS = [("fixed", False), ("fixed", True), ("free", False), ("free", True)] # (iv, feedforward)202122def list_cells(seeds, solvers):23 cells = []24 for iv, ff in VARIANTS:25 for r in R_LIST:26 for seed in seeds:27 for sol in solvers:28 cells.append({"r": r, "seed": seed, "solver": sol, "iv": iv, "feedforward": ff})29 return cells303132def build(r: int, free_iv: bool, ff: bool):33 cnf = C.CNF()34 W1 = [cnf.word() for _ in range(16)]35 W2 = [cnf.word() for _ in range(16)]36 IV = [cnf.word() for _ in range(8)] if free_iv else [C.CNF.const_word(h) for h in S.H0]37 O1 = cnf.compress(IV, W1, r, ff)38 O2 = cnf.compress(IV, W2, r, ff)39 for x, y in zip(O1, O2):40 cnf.eq_w(x, y)41 # M != M' restricted to the words an r-step compression actually reads (W_0..W_{min(r,16)-1}); otherwise a42 # difference in an unused word is a trivial "collision".43 used = min(r, 16)44 cnf.neq_w_any(list(zip(W1[:used], W2[:used])))45 for j in range(used, 16): # unused words are forced equal (and to zero) to remove the useless freedom46 cnf.fix_word(W1[j], 0); cnf.fix_word(W2[j], 0)47 return cnf, W1, W2, IV484950def run_cell(cell: dict, timeout: float) -> dict:51 r, seed, sol = cell["r"], cell["seed"], cell["solver"]52 free_iv, ff = cell["iv"] == "free", bool(cell["feedforward"])53 t0 = time.time()54 cnf, W1, W2, IV = build(r, free_iv, ff)55 # seed-dependent symmetry breaking is deliberately NOT added; the seed only changes the solver's random seed56 # (Kissat --seed) and is recorded for repeated runs of CaDiCaL (deterministic => identical runs).57 res = {"cell": cell, "timeout_s": timeout, **cnf.stats(), "encode_s": round(time.time() - t0, 3)}58 if sol == "cadical":59 out = SV.run_cadical(cnf.clauses, timeout, n_vars=cnf.nv, seed=seed)60 else:61 out = SV.run_kissat(cnf.clauses, cnf.nv, timeout, seed=seed)62 model = out.pop("model", None)63 res.update(out)64 if out["status"] == "sat":65 m1 = [C.decode_word(model, w) for w in W1]66 m2 = [C.decode_word(model, w) for w in W2]67 iv = [C.decode_word(model, w) for w in IV]68 b1, b2 = S.words_to_bytes(m1), S.words_to_bytes(m2)69 h1 = S.compress(iv, b1, r, feedforward=ff)70 h2 = S.compress(iv, b2, r, feedforward=ff)71 res.update({"message1_hex": b1.hex(), "message2_hex": b2.hex(), "iv_hex": "".join("%08x" % v for v in iv),72 "output_hex": "".join("%08x" % v for v in h1),73 "verified": (h1 == h2 and b1 != b2),74 "message_xor_difference_hex": (int.from_bytes(b1, "big") ^ int.from_bytes(b2, "big")).to_bytes(64, "big").hex(),75 "differing_words": [j for j in range(16) if m1[j] != m2[j]]})76 res["total_wall_s"] = round(time.time() - t0, 3)77 res["peak_rss_mb"] = max(peak_rss_mb(), peak_rss_mb(children=True))78 res["hostname"] = provenance.host_info()["hostname"]79 return res808182def aggregate(d: str, out: str, started: float):83 cells = [c for c in (json.loads(p.read_text()) for p in sorted(pathlib.Path(d).glob("*.json"))) if isinstance(c, dict) and "cell" in c]84 summary = {}85 for iv, ff in VARIANTS:86 for sol in ("cadical", "kissat"):87 for r in R_LIST:88 cs = [c for c in cells if c["cell"]["solver"] == sol and c["cell"]["r"] == r89 and c["cell"]["iv"] == iv and bool(c["cell"]["feedforward"]) == ff]90 if not cs:91 continue92 solved = [c for c in cs if c["status"] == "sat"]93 times = sorted(c["wall_s"] for c in cs)94 conf = sorted(c["conflicts"] for c in solved if c.get("conflicts") is not None)95 summary[f"{sol}:iv_{iv}:ff{int(ff)}:r{r}"] = {96 "solver": sol, "iv": iv, "feedforward": ff, "r": r, "n_runs": len(cs), "n_sat": len(solved),97 "n_timeout": sum(c["status"] == "timeout" for c in cs), "solved_fraction": len(solved) / len(cs),98 "n_unsat": sum(c["status"] == "unsat" for c in cs),99 "median_wall_s_unsat": sorted(c["wall_s"] for c in cs if c["status"] == "unsat")[sum(c["status"] == "unsat" for c in cs) // 2] if any(c["status"] == "unsat" for c in cs) else None,100 "median_wall_s": times[len(times) // 2],101 "median_wall_s_solved": sorted(c["wall_s"] for c in solved)[len(solved) // 2] if solved else None,102 "median_conflicts_solved": conf[len(conf) // 2] if conf else None,103 "all_verified": all(c.get("verified") for c in solved) if solved else None,104 "n_vars": cs[0]["n_vars"], "n_clauses": cs[0]["n_clauses"],105 "max_peak_rss_mb": max(c["peak_rss_mb"] for c in cs), "hosts": sorted({c["hostname"] for c in cs}),106 }107 provenance.record("sha256_collision_sat", {"r_list": R_LIST, "timeout_s": TIMEOUT, "variants": VARIANTS,108 "solvers": ["cadical (pysat Cadical195)", "kissat 4.0.4"]},109 {"summary": summary, "cells": cells}, seed="per-cell", extra={"n_cells": len(cells)},110 out=out, started=started)111 print("aggregated", len(cells), "cells ->", out)112113114if __name__ == "__main__":115 ap = argparse.ArgumentParser()116 ap.add_argument("--list-cells", action="store_true")117 ap.add_argument("--seeds", default="0,1,2")118 ap.add_argument("--solvers", default="cadical,kissat")119 ap.add_argument("--cell")120 ap.add_argument("--timeout", type=float, default=TIMEOUT)121 ap.add_argument("--aggregate")122 ap.add_argument("--out")123 ap.add_argument("--seed", type=int, default=0)124 a = ap.parse_args()125 t0 = time.time()126 if a.list_cells:127 for c in list_cells([int(x) for x in a.seeds.split(",")], a.solvers.split(",")):128 print(json.dumps(c))129 elif a.cell:130 cell = json.loads(a.cell); cell.setdefault("seed", a.seed)131 res = run_cell(cell, a.timeout)132 if a.out:133 pathlib.Path(a.out).parent.mkdir(parents=True, exist_ok=True)134 pathlib.Path(a.out).write_text(json.dumps(res, indent=1, default=str))135 print(json.dumps(res, default=str)[:1500])136 elif a.aggregate:137 aggregate(a.aggregate, a.out, t0)138