diff --git a/ownlang/fix_apply.py b/ownlang/fix_apply.py new file mode 100644 index 00000000..375c9078 --- /dev/null +++ b/ownlang/fix_apply.py @@ -0,0 +1,218 @@ +"""S2 slice 1 — the strict apply-input gate: validate the S1 validated-plan.json, bind +it by hash to its candidates.json, cross-check every decision against its candidate, and +guard the source file (SHA + root confinement). NOTHING here parses C# or writes a byte; +it is the untrusted-input gate that must pass before the (later-slice) `Owen.CSharp.Rewriter` +is allowed to touch source. + +Two hash-bound inputs (arbiter): the validated-plan is the sole authority for `action`; +the candidates bundle is the untrusted-but-hash-bound source of the syntax/semantic +context (event / source / handler identity, containing type) the rewriter's span-node +guard needs. `bundle_sha256(candidates) == validated_plan.input_bundle_sha256` binds them. +A stale source, an escaping path, a decision that disagrees with its candidate, or a +convert_acquire on a non-INotifyPropertyChanged contract is a hard refusal. +""" + +from __future__ import annotations + +import hashlib +from itertools import pairwise +from typing import Any + +from ownlang.fix_candidates import ( + S1_ACTIONS, + CollectError, + _field, + _require_canonical_relpath, + _require_finding_id, + _require_sha256, + _resolve_source, + _validate_span, + validate_candidates_bundle, +) +from ownlang.fix_plan import bundle_sha256 + +_PLAN_KEYS = { + "version", "operation", "input_bundle_sha256", "target_api", + "selection", "source_files", "decisions", +} +_DECISION_KEYS = {"finding_id", "action", "file", "acquire_span"} + + +class ApplyError(Exception): + """A malformed / unbound / stale apply input. A hard, controlled refusal — never a + traceback, never a partial apply.""" + + +def validate_validated_plan(plan: Any) -> None: + """Validate the shape of an S1 validated-plan.json (fix_plan.validate_plan's output). + Raises ApplyError on any violation.""" + if not isinstance(plan, dict): + raise ApplyError("validated-plan must be an object") + unknown = sorted(set(plan) - _PLAN_KEYS) + if unknown: + raise ApplyError(f"validated-plan has unknown field(s): {unknown}") + if plan.get("version") != 1 or isinstance(plan.get("version"), bool): + raise ApplyError("validated-plan version must be integer 1") + if plan.get("operation") != "fix-subscriptions": + raise ApplyError("validated-plan operation must be 'fix-subscriptions'") + try: + _require_sha256( + _field(plan, "input_bundle_sha256", "str", "validated-plan"), + "validated-plan.input_bundle_sha256", + ) + _field(_field(plan, "target_api", "dict", "validated-plan"), "subscribe", "str", + "validated-plan.target_api") + sel = _field(plan, "selection", "dict", "validated-plan") + at = _field(sel, "allowed_types", "list", "validated-plan.selection") + if len(at) != 1: + raise ApplyError("validated-plan.selection.allowed_types must have exactly one entry") + sfs = _field(plan, "source_files", "list", "validated-plan") + if len(sfs) != 1: + raise ApplyError("validated-plan.source_files must have exactly one entry") + sf = sfs[0] + if not isinstance(sf, dict): + raise ApplyError("validated-plan.source_files[0] must be an object") + _require_canonical_relpath(_field(sf, "path", "str", "validated-plan.source_files[0]"), + "validated-plan.source_files[0]") + _require_sha256(_field(sf, "sha256", "str", "validated-plan.source_files[0]"), + "validated-plan.source_files[0]") + for i, d in enumerate(_field(plan, "decisions", "list", "validated-plan")): + dctx = f"validated-plan.decisions[{i}]" + if not isinstance(d, dict): + raise ApplyError(f"{dctx}: must be an object") + extra = sorted(set(d) - _DECISION_KEYS) + if extra: + raise ApplyError(f"{dctx}: unknown field(s) {extra}") + _require_finding_id(_field(d, "finding_id", "str", dctx), dctx) + action = _field(d, "action", "str", dctx) + if action not in S1_ACTIONS: + raise ApplyError(f"{dctx}: out-of-scope action {action!r}") + _require_canonical_relpath(_field(d, "file", "str", dctx), dctx) + _validate_span(d.get("acquire_span"), dctx) + except CollectError as exc: + raise ApplyError(f"validated-plan: {exc}") from exc + + +def _sha_file(abs_path: str) -> str: + with open(abs_path, "rb") as fh: + return "sha256:" + hashlib.sha256(fh.read()).hexdigest() + + +def validate_apply_inputs( + validated_plan: dict[str, Any], candidates: dict[str, Any], root: str +) -> dict[str, Any]: + """Run the full S2 pre-apply gate over the two hash-bound inputs + the source root. + Returns an apply CONTEXT (the convert_acquire targets with their candidate identities, + the manual_review ids, the source path, the target API) once every guard passes. + Raises ApplyError on any violation — no source is read for rewriting, only hashed.""" + validate_validated_plan(validated_plan) + try: + validate_candidates_bundle(candidates) + except CollectError as exc: + raise ApplyError(f"invalid candidates bundle: {exc}") from exc + + # Hash binding: the candidates are the plan's own input. + if bundle_sha256(candidates) != validated_plan["input_bundle_sha256"]: + raise ApplyError("candidates sha256 does not match validated-plan.input_bundle_sha256") + + # The plan's materialized authority envelope must EQUAL the canonical projection of + # the candidates' known keys — not a raw dict copy. This rejects a self-consistent + # forged pair that smuggled extra nested fields, a wrong type/file, wrong constraints, + # or mismatched selected_findings into the plan. + the_type = candidates["selection"]["allowed_types"][0] + the_source = candidates["source_files"][0] + expected_target_api = {"subscribe": candidates["target_api"]["subscribe"]} + expected_selection = { + "allowed_types": [{"full_name": the_type["full_name"], "file": the_type["file"]}], + "selected_findings": candidates["selection"].get("selected_findings"), + "constraints": { + "max_types_changed": 1, + "max_files_changed": 1, + "allow_helper_changes": False, + "allow_config_changes": False, + "allow_suppressions": False, + }, + } + expected_source_files = [{"path": the_source["path"], "sha256": the_source["sha256"]}] + if validated_plan["target_api"] != expected_target_api: + raise ApplyError("validated-plan.target_api is not the canonical projection of candidates") + if validated_plan["selection"] != expected_selection: + raise ApplyError("validated-plan.selection is not the canonical projection of candidates") + if validated_plan["source_files"] != expected_source_files: + raise ApplyError("validated-plan.source_files is not the canonical projection") + + candidate_by_id = {c["finding_id"]: c for c in candidates["candidates"]} + seen: set[str] = set() + convert: list[dict[str, Any]] = [] + manual: list[str] = [] + for d in validated_plan["decisions"]: + fid = d["finding_id"] + if fid not in candidate_by_id: + raise ApplyError(f"decision {fid!r} is not a candidate") + if fid in seen: + raise ApplyError(f"duplicate decision {fid!r}") + seen.add(fid) + c = candidate_by_id[fid] + if d["file"] != c["file"]: + raise ApplyError(f"{fid}: decision file != candidate file") + if d["acquire_span"] != c["acquire_span"]: + raise ApplyError(f"{fid}: decision acquire_span != candidate acquire_span") + if d["action"] not in c["allowed_actions"]: + raise ApplyError(f"{fid}: action {d['action']!r} not allowed by candidate") + if d["action"] == "convert_acquire": + if c["event_contract"] != "inotify_property_changed": + raise ApplyError( + f"{fid}: convert_acquire requires an inotify_property_changed contract" + ) + # The FULL identity contract the future span-node guard needs (not just the + # human display). validate_candidates_bundle guarantees all are present + str, + # so the direct index cannot KeyError. + convert.append({ + "finding_id": fid, + "file": c["file"], + "acquire_span": c["acquire_span"], + "containing_type": c["containing_type"], + "event": c["event"], + "event_identity": c["event_identity"], + "source": c["source"], + "source_identity": c["source_identity"], + "source_identity_kind": c["source_identity_kind"], + "handler": c["handler"], + "handler_identity": c["handler_identity"], + "handler_identity_kind": c["handler_identity_kind"], + }) + else: + manual.append(fid) + missing = set(candidate_by_id) - seen + if missing: + raise ApplyError(f"missing decisions for: {sorted(missing)}") + + # convert_acquire edit spans must not overlap. + ranges = sorted( + (e["acquire_span"]["start"], e["acquire_span"]["start"] + e["acquire_span"]["length"]) + for e in convert + ) + for (_, end_a), (start_b, _) in pairwise(ranges): + if start_b < end_a: + raise ApplyError("overlapping convert_acquire spans") + + # Source guard: confine to root, then match the pristine preimage SHA. A read that + # fails between the confinement check and the hash (file vanished / permission / + # replaced) is normalized to ApplyError, never a leaked OSError. + source = candidates["source_files"][0] + try: + canonical, abs_path = _resolve_source(root, source["path"]) + actual_sha = _sha_file(abs_path) + except CollectError as exc: + raise ApplyError(f"source path: {exc}") from exc + except OSError as exc: + raise ApplyError(f"cannot read source file: {exc}") from exc + if actual_sha != source["sha256"]: + raise ApplyError(f"STALE SOURCE / PREIMAGE MISMATCH for {canonical}") + + return { + "source_file": canonical, + "target_subscribe": validated_plan["target_api"]["subscribe"], + "convert_acquire": convert, + "manual_review": manual, + } diff --git a/ownlang/fix_candidates.py b/ownlang/fix_candidates.py index a4fe2b08..3fd34116 100644 --- a/ownlang/fix_candidates.py +++ b/ownlang/fix_candidates.py @@ -416,7 +416,10 @@ def validate_candidates_bundle(bundle: Any) -> None: raise CollectError(f"{cctx}: containing_type must be {type_name!r}") if _field(c, "file", "str", cctx) != source_path: raise CollectError(f"{cctx}: file must be {source_path!r}") - for k in ("event", "source", "handler"): + # Display + full identity set — required so downstream (S2 apply) can index them + # without a KeyError and the span-node guard gets the real identity contract. + for k in ("event", "source", "handler", "event_identity", "source_identity", + "source_identity_kind", "handler_identity", "handler_identity_kind"): _field(c, k, "str", cctx) contract = _field(c, "event_contract", "str", cctx) if contract not in _CONTRACTS: diff --git a/tests/test_fix_apply.py b/tests/test_fix_apply.py new file mode 100644 index 00000000..ee2ac817 --- /dev/null +++ b/tests/test_fix_apply.py @@ -0,0 +1,275 @@ +#!/usr/bin/env python3 +"""S2 slice 1 — the apply-input gate (validate + hash-bind + source guard), SDK-free. + +Covers the locked pre-apply contract that must pass before any source is rewritten: +validated-plan shape, candidates hash binding, decision↔candidate cross-checks, the +frozen convert_acquire-only-for-INPC tiering (via the allowed_actions path), overlapping +spans, root confinement, and the pristine preimage SHA guard. No C# / Roslyn here. + +Run: python tests/test_fix_apply.py + python tests/run_tests.py (auto-discovered) +""" + +from __future__ import annotations + +import hashlib +import os +import sys +import tempfile + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from ownlang.fix_apply import ApplyError, validate_apply_inputs +from ownlang.fix_plan import bundle_sha256 + +_FID_A = "OWN001:sha256:" + "a" * 64 +_FID_B = "OWN001:sha256:" + "b" * 64 +_EV = "System.ComponentModel.INotifyPropertyChanged.PropertyChanged" +_REL = "N/C.cs" + + +def _cand(fid: str, actions: tuple[str, ...], contract: str, start: int) -> dict: + return { + "finding_id": fid, "diagnostic_code": "OWN001", "containing_type": "N.C", + "file": _REL, "enclosing_member": "N.C.C()", "event": "PropertyChanged", + "event_identity": _EV, "event_contract": contract, "source": "_pub", + "source_identity": "N.C._pub", "source_identity_kind": "stable_symbol", + "handler": "OnChanged", "handler_identity": "N.C.OnChanged(object, ...)", + "handler_identity_kind": "stable_symbol", "occurrence_ordinal": 0, + "acquire_span": {"start": start, "length": 30, "start_line": 1, + "start_column": 1, "end_line": 1, "end_column": 31}, + "teardown": {"status": "none", "candidates": []}, + "allowed_actions": list(actions), + } + + +def _bundle(cands: list[dict], sha: str) -> dict: + return { + "version": 1, "operation": "fix-subscriptions", + "target_api": {"subscribe": "WeakEvents.AddPropertyChanged"}, + "selection": { + "allowed_types": [{"full_name": "N.C", "file": _REL}], + "selected_findings": None, + "constraints": {"max_types_changed": 1, "max_files_changed": 1, + "allow_helper_changes": False, "allow_config_changes": False, + "allow_suppressions": False}, + }, + "source_files": [{"path": _REL, "sha256": sha}], + "candidates": cands, + } + + +def _decision(fid: str, action: str, start: int) -> dict: + return {"finding_id": fid, "action": action, "file": _REL, + "acquire_span": {"start": start, "length": 30, "start_line": 1, + "start_column": 1, "end_line": 1, "end_column": 31}} + + +def _vplan(cands: dict, decisions: list[dict]) -> dict: + the_type = cands["selection"]["allowed_types"][0] + sf = cands["source_files"][0] + return { + "version": 1, "operation": "fix-subscriptions", + "input_bundle_sha256": bundle_sha256(cands), + "target_api": {"subscribe": cands["target_api"]["subscribe"]}, + "selection": { + "allowed_types": [{"full_name": the_type["full_name"], "file": the_type["file"]}], + "selected_findings": cands["selection"].get("selected_findings"), + "constraints": {"max_types_changed": 1, "max_files_changed": 1, + "allow_helper_changes": False, "allow_config_changes": False, + "allow_suppressions": False}, + }, + "source_files": [{"path": sf["path"], "sha256": sf["sha256"]}], + "decisions": decisions, + } + + +def run() -> int: + ok = 0 + bad = 0 + + def check(cond: bool, label: str) -> None: + nonlocal ok, bad + if cond: + ok += 1 + else: + bad += 1 + print(f" FAIL: {label}") + + def raises(vplan: dict, cands: dict, root: str) -> bool: + try: + validate_apply_inputs(vplan, cands, root) + except ApplyError: + return True + return False + + with tempfile.TemporaryDirectory() as root: + src = os.path.join(root, "N", "C.cs") + os.makedirs(os.path.dirname(src), exist_ok=True) + content = b"// pretend source with a subscription\nclass C {}\n" + with open(src, "wb") as fh: + fh.write(content) + sha = "sha256:" + hashlib.sha256(content).hexdigest() + + inpc = ("convert_acquire", "manual_review") + cands = _bundle([_cand(_FID_A, inpc, "inotify_property_changed", 100), + _cand(_FID_B, ("manual_review",), "name_only", 200)], sha) + + # happy path: one convert_acquire + one manual_review + vplan = _vplan(cands, [_decision(_FID_A, "convert_acquire", 100), + _decision(_FID_B, "manual_review", 200)]) + ctx = validate_apply_inputs(vplan, cands, root) + conv = ctx["convert_acquire"] + check(len(conv) == 1 and conv[0]["finding_id"] == _FID_A, "one convert_acquire target") + check(ctx["manual_review"] == [_FID_B], "one manual_review") + check(ctx["source_file"] == _REL + and ctx["target_subscribe"] == "WeakEvents.AddPropertyChanged", + "context carries source + target") + target0 = ctx["convert_acquire"][0] + check(target0["source"] == "_pub" and target0["handler"] == "OnChanged", + "convert target carries candidate display identity") + check(set(target0) >= {"source_identity", "source_identity_kind", "handler_identity", + "handler_identity_kind", "event_identity", "containing_type"}, + "convert target carries the FULL identity contract") + + # Blocker 1: the plan envelope must be the canonical projection of candidates. + def vgood() -> dict: + return _vplan(cands, [_decision(_FID_A, "convert_acquire", 100), + _decision(_FID_B, "manual_review", 200)]) + v = vgood() + v["selection"]["allowed_types"][0]["full_name"] = "Other.Type" + check(raises(v, cands, root), "wrong selected type refused") + v = vgood() + v["selection"]["allowed_types"][0]["file"] = "N/Other.cs" + check(raises(v, cands, root), "wrong selected type file refused") + v = vgood() + v["selection"]["constraints"]["max_types_changed"] = 2 + check(raises(v, cands, root), "wrong constraints refused") + v = vgood() + v["selection"]["selected_findings"] = [_FID_A] + check(raises(v, cands, root), "selected_findings mismatch refused") + v = vgood() + v["target_api"]["extra"] = 1 + check(raises(v, cands, root), "unknown nested target_api field refused") + v = vgood() + v["selection"]["extra"] = 1 + check(raises(v, cands, root), "unknown nested selection field refused") + v = vgood() + v["source_files"][0]["extra"] = 1 + check(raises(v, cands, root), "unknown nested source_files field refused") + + # Blocker 2: a candidate missing an identity field is a controlled ApplyError. + for missing in ("event_identity", "source_identity", "handler_identity", + "handler_identity_kind"): + bc_cand = _cand(_FID_A, inpc, "inotify_property_changed", 100) + del bc_cand[missing] + bc = _bundle([bc_cand], sha) + bv = _vplan(bc, [_decision(_FID_A, "convert_acquire", 100)]) + check(raises(bv, bc, root), f"candidate missing {missing} -> ApplyError") + bad_type_cand = _cand(_FID_A, inpc, "inotify_property_changed", 100) + bad_type_cand["source_identity"] = 42 + bt = _bundle([bad_type_cand], sha) + btv = _vplan(bt, [_decision(_FID_A, "convert_acquire", 100)]) + check(raises(btv, bt, root), "candidate identity of wrong type -> ApplyError") + + # hash binding: mutate candidates after the plan was built + mutated = _bundle([_cand(_FID_A, inpc, "inotify_property_changed", 100), + _cand(_FID_B, ("manual_review",), "name_only", 999)], sha) + check(raises(vplan, mutated, root), "candidates/plan hash mismatch refused") + + # target / source_files mismatch + bad_target = _vplan(cands, [_decision(_FID_A, "convert_acquire", 100), + _decision(_FID_B, "manual_review", 200)]) + bad_target["target_api"] = {"subscribe": "Other.Add"} + check(raises(bad_target, cands, root), "target_api mismatch refused") + + # decision file != candidate file + v = _vplan(cands, [_decision(_FID_A, "convert_acquire", 100), + _decision(_FID_B, "manual_review", 200)]) + v["decisions"][0]["file"] = "N/Other.cs" + check(raises(v, cands, root), "decision file != candidate refused") + # decision span != candidate span + v = _vplan(cands, [_decision(_FID_A, "convert_acquire", 101), + _decision(_FID_B, "manual_review", 200)]) + check(raises(v, cands, root), "decision span != candidate refused") + + # action not allowed by the candidate (convert on a manual-only finding) + v = _vplan(cands, [_decision(_FID_A, "convert_acquire", 100), + _decision(_FID_B, "convert_acquire", 200)]) + check(raises(v, cands, root), "convert_acquire on a manual-only candidate refused") + + # unknown / missing / duplicate decisions + v = _vplan(cands, [_decision(_FID_A, "convert_acquire", 100), + _decision(_FID_B, "manual_review", 200), + _decision("OWN001:sha256:" + "c" * 64, "manual_review", 300)]) + check(raises(v, cands, root), "unknown decision refused") + v = _vplan(cands, [_decision(_FID_A, "convert_acquire", 100)]) + check(raises(v, cands, root), "missing decision refused") + v = _vplan(cands, [_decision(_FID_A, "convert_acquire", 100), + _decision(_FID_A, "manual_review", 100), + _decision(_FID_B, "manual_review", 200)]) + check(raises(v, cands, root), "duplicate decision refused") + + # overlapping convert spans + over = _bundle([_cand(_FID_A, inpc, "inotify_property_changed", 100), + _cand(_FID_B, inpc, "inotify_property_changed", 110)], sha) + vo = _vplan(over, [_decision(_FID_A, "convert_acquire", 100), + _decision(_FID_B, "convert_acquire", 110)]) + check(raises(vo, over, root), "overlapping convert spans refused") + + # unknown fields / bad version / out-of-scope action in the plan + v = _vplan(cands, [_decision(_FID_A, "convert_acquire", 100), + _decision(_FID_B, "manual_review", 200)]) + v["oops"] = 1 + check(raises(v, cands, root), "unknown top-level plan field refused") + v = _vplan(cands, [_decision(_FID_A, "convert_acquire", 100), + _decision(_FID_B, "manual_review", 200)]) + v["decisions"][0]["confidence"] = "high" + check(raises(v, cands, root), "unknown decision field refused") + v = _vplan(cands, [_decision(_FID_A, "convert_exact_teardown", 100), + _decision(_FID_B, "manual_review", 200)]) + check(raises(v, cands, root), "out-of-scope action refused") + + # stale source SHA + stale_sha = "sha256:" + "0" * 64 + stale_cands = _bundle([_cand(_FID_A, inpc, "inotify_property_changed", 100)], stale_sha) + stale_v = _vplan(stale_cands, [_decision(_FID_A, "convert_acquire", 100)]) + check(raises(stale_v, stale_cands, root), "stale source preimage SHA refused") + + # root confinement: a candidates path escaping the root + with tempfile.TemporaryDirectory() as root2: + esc = _bundle([_cand(_FID_A, ("convert_acquire", "manual_review"), + "inotify_property_changed", 100)], "sha256:" + "0" * 64) + esc["source_files"][0]["path"] = "N/C.cs" # keep candidates valid; file just won't exist + ev = _vplan(esc, [_decision(_FID_A, "convert_acquire", 100)]) + # the file does not exist under root2 -> _resolve_source refuses (not a regular file) + check(raises(ev, esc, root2), "missing/uncontained source refused") + + # Blocker 3: an OSError reading the source (perms) is normalized to ApplyError. + with tempfile.TemporaryDirectory() as root3: + p = os.path.join(root3, "N", "C.cs") + os.makedirs(os.path.dirname(p), exist_ok=True) + body = b"class C {}\n" + with open(p, "wb") as fh: + fh.write(body) + os.chmod(p, 0) + unreadable = True + try: + with open(p, "rb"): + unreadable = False # perms not enforced (Windows / running as root) -> skip + except OSError: + pass + if unreadable: + sha3 = "sha256:" + hashlib.sha256(body).hexdigest() + oc = _bundle([_cand(_FID_A, ("convert_acquire", "manual_review"), + "inotify_property_changed", 100)], sha3) + ov = _vplan(oc, [_decision(_FID_A, "convert_acquire", 100)]) + check(raises(ov, oc, root3), "source read OSError -> ApplyError") + os.chmod(p, 0o644) # restore so the tempdir cleans up + + print(f"fix-apply (S2 slice 1): {ok} ok, {bad} bad") + return bad + + +if __name__ == "__main__": + raise SystemExit(run())