From 73e856a06b44fd5e87f930379acd3dea4be50671 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 5 Sep 2026 14:55:00 +0200 Subject: [PATCH] feat(corpus-seeded-mechanisms): the mechanisms lever admits or retracts exporter records Task 4.1 (Req 5.1, 5.2, 5.5). - validator: MechanismEdit (admit | retract by record id) under the `mechanisms` lever; validate_mechanism accepts only records from the exporter candidate set with origin corpus, a closed guard and source kinds, and identifier-only shape text; free-form shapes are rejected. - apply_mechanism_edits appends the record (admit) or a retracted tombstone (retract) so the log stays append-only; StoreSnapshot.restore reverts byte for byte. MechanismStore.load folds tombstones away. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- src/openultrasast/improve/validator.py | 101 ++++++++++++++++++++++- src/openultrasast/semantic/mechanisms.py | 9 +- tests/test_mechanism_lever.py | 85 +++++++++++++++++++ 3 files changed, 190 insertions(+), 5 deletions(-) create mode 100644 tests/test_mechanism_lever.py diff --git a/src/openultrasast/improve/validator.py b/src/openultrasast/improve/validator.py index ba57b6b..df09f09 100644 --- a/src/openultrasast/improve/validator.py +++ b/src/openultrasast/improve/validator.py @@ -8,15 +8,20 @@ impossible; the validator additionally enforces status/bounds/staging/resolution from __future__ import annotations -from collections.abc import Mapping -from dataclasses import dataclass +import json +import re +from collections.abc import Mapping, Sequence +from dataclasses import asdict, dataclass +from pathlib import Path from ..policy import CwePolicy from ..ruleset import PatternRule from ..ruleset.store import VALID_STATUS # The only levers the loop may pull (mirrors the design's journal._VALID_LEVERS). -VALID_LEVERS = frozenset({"rule", "policy"}) +VALID_LEVERS = frozenset({"rule", "policy", "mechanisms"}) +_IDENTIFIER = re.compile(r"^[A-Za-z_$][\w$]*$") +MECHANISM_ACTIONS = frozenset({"admit", "retract"}) # Policy constants the loop may tune; the 0-5 severity is intentionally absent. TUNABLE_POLICY_CONSTANTS = frozenset({"K", "MIN_SCORE"}) @@ -57,6 +62,40 @@ class PolicyConstantEdit: return f"policy:{self.name}:{self.from_value}->{self.to_value}" +@dataclass(frozen=True) +class MechanismEdit: + """A `mechanisms`-lever edit (corpus-seeded-mechanisms Req 5): admit or retract one exporter record by id. + + The loop never writes a shape; it may only move a record the exporter derived from a trusted pair into or out of + the scan-time store. Free text is confined to ``rationale``. + """ + + action: str # admit | retract + mechanism_id: str + source: str = "loo" # loo | export + rationale: str = "" + lever: str = "mechanisms" + + def key(self) -> str: + return f"mechanisms:{self.action}:{self.mechanism_id}" + + +@dataclass(frozen=True) +class StoreSnapshot: + """Bytes of the scan-time mechanism store before a round; ``restore`` is the byte-for-byte revert.""" + + path: Path + content: bytes | None # None when the file did not exist + + def restore(self) -> None: + if self.content is None: + if self.path.exists(): + self.path.unlink() + return + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_bytes(self.content) + + @dataclass(frozen=True) class EvolveBounds: k_range: tuple[float, float] = (20.0, 200.0) @@ -74,9 +113,42 @@ class EvolveValidator: self._validate_rule(edit, ruleset_by_id, policy) elif isinstance(edit, PolicyConstantEdit): self._validate_policy(edit) + elif isinstance(edit, MechanismEdit): + raise StrictValidationError("mechanism edits need the exporter candidate set: use validate_mechanism", kind="mechanism_change") else: raise StrictValidationError(f"unknown edit type {type(edit).__name__}", kind="rule_change") + def validate_mechanism(self, edit: MechanismEdit, candidates: Mapping[str, object]) -> None: + """Req 5.2: only records the exporter derived from trusted pairs, with a closed guard and identifier-only shape text.""" + from ..semantic.variants import GUARD_KINDS, SOURCE_KINDS + + if edit.action not in MECHANISM_ACTIONS: + raise StrictValidationError(f"mechanism edit action {edit.action!r} is not admit or retract", kind="mechanism_change") + record = candidates.get(edit.mechanism_id) + if record is None: + raise StrictValidationError( + f"unknown mechanism {edit.mechanism_id!r}: not in the exporter candidate set", kind="mechanism_change" + ) + if getattr(record, "origin", "") != "corpus": + raise StrictValidationError( + f"{edit.mechanism_id}: origin {getattr(record, 'origin', '')!r} is not corpus (only exporter records)", + kind="mechanism_change", + ) + shape = getattr(record, "shape", None) + if not isinstance(shape, dict): + raise StrictValidationError(f"{edit.mechanism_id}: record carries no shape", kind="mechanism_change") + if str(shape.get("guard")) not in GUARD_KINDS or str(getattr(record, "guard", "")) not in GUARD_KINDS: + raise StrictValidationError( + f"{edit.mechanism_id}: guard {shape.get('guard')!r} is not a closed guard kind", kind="mechanism_change" + ) + for field_name in ("sink_name", "language", "mechanism"): + value = str(shape.get(field_name, "")) + if not _IDENTIFIER.match(value.replace("-", "_")): + raise StrictValidationError(f"{edit.mechanism_id}: {field_name} {value!r} is not an identifier", kind="mechanism_change") + kinds = shape.get("source_kinds") or [] + if not isinstance(kinds, list) or any(str(kind) not in SOURCE_KINDS for kind in kinds): + raise StrictValidationError(f"{edit.mechanism_id}: source kinds {kinds!r} are not closed kinds", kind="mechanism_change") + def _validate_rule(self, edit: RuleStatusEdit, ruleset_by_id: Mapping[str, PatternRule], policy: Mapping[str, CwePolicy]) -> None: if edit.to_status not in VALID_STATUS: raise StrictValidationError(f"invalid status {edit.to_status!r} for {edit.rule_id}") @@ -99,6 +171,25 @@ class EvolveValidator: raise StrictValidationError(f"MIN_SCORE={edit.to_value} out of bounds {self._bounds.min_score_range}", kind="policy_change") +def apply_mechanism_edits(edits: Sequence[MechanismEdit], candidates: object, scan_store: Path) -> StoreSnapshot: + """Apply admit/retract edits to the scan-time store; returns the snapshot whose ``restore`` reverts byte for byte. + + Admission appends the exporter record unchanged; retraction appends a tombstone row (``retracted = true``) so the + log stays append-only and ``MechanismStore.load`` folds the record away. + """ + snapshot = StoreSnapshot(path=scan_store, content=scan_store.read_bytes() if scan_store.exists() else None) + by_id = {record.id: record for record in candidates.load()} # type: ignore[attr-defined] + scan_store.parent.mkdir(parents=True, exist_ok=True) + with scan_store.open("a") as handle: + for edit in edits: + record = by_id[edit.mechanism_id] + payload = asdict(record) + if edit.action == "retract": + payload["retracted"] = True + handle.write(json.dumps(payload, sort_keys=True) + "\n") + return snapshot + + def edits_to_ledger(edits: list[RuleStatusEdit], base: Mapping[str, dict[str, object]]) -> dict[str, dict[str, object]]: """Overlay rule-status edits onto a copy of the base ledger.""" ledger: dict[str, dict[str, object]] = {key: dict(value) for key, value in base.items()} @@ -112,6 +203,10 @@ def edits_to_ledger(edits: list[RuleStatusEdit], base: Mapping[str, dict[str, ob __all__ = [ "EvolveBounds", "EvolveValidator", + "MECHANISM_ACTIONS", + "MechanismEdit", + "StoreSnapshot", + "apply_mechanism_edits", "PolicyConstantEdit", "RuleStatusEdit", "StrictValidationError", diff --git a/src/openultrasast/semantic/mechanisms.py b/src/openultrasast/semantic/mechanisms.py index e9074b6..3e9db86 100644 --- a/src/openultrasast/semantic/mechanisms.py +++ b/src/openultrasast/semantic/mechanisms.py @@ -37,6 +37,7 @@ class Mechanism: pairs: tuple[str, ...] = () shape: dict[str, object] | None = None # Shape.to_dict(); None on sandbox rows (not searchable) guard: str = "none" + retracted: bool = False # tombstone row appended by the improve lever; load() folds the record away class MechanismStore: @@ -53,7 +54,10 @@ class MechanismStore: continue payload = json.loads(line) record = _mechanism_from_payload(payload) - by_id[record.id] = record + if record.retracted: + by_id.pop(record.id, None) + else: + by_id[record.id] = record return tuple(by_id.values()) def append(self, mechanism: Mechanism) -> None: @@ -340,8 +344,9 @@ def _mechanism_from_payload(payload: dict[str, object]) -> Mechanism: origin=str(payload.get("origin", "sandbox")), review_tier=str(payload.get("review_tier", "")), pairs=tuple(str(item) for item in pairs), - shape=Shape.from_dict(shape).to_dict() if isinstance(shape, dict) else None, + shape=dict(shape) if isinstance(shape, dict) else None, # validated where it is used (search_tree, the lever) guard=str(payload.get("guard", "none")), + retracted=bool(payload.get("retracted", False)), ) diff --git a/tests/test_mechanism_lever.py b/tests/test_mechanism_lever.py new file mode 100644 index 0000000..9512f18 --- /dev/null +++ b/tests/test_mechanism_lever.py @@ -0,0 +1,85 @@ +"""corpus-seeded-mechanisms task 4.1: the `mechanisms` lever admits or retracts exporter records, nothing else.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from openultrasast.semantic.mechanisms import MechanismStore, append_from_pair +from openultrasast.semantic.variants import Shape + + +def _record(store: MechanismStore, **overrides: object): + base: dict[str, object] = { + "language": "python", + "sink_name": "system", + "arity": 1, + "source_positions": (0,), + "source_kinds": ("parameter",), + "guard": "allowlist_test", + "mechanism": "source_reaches_sink", + } + base.update(overrides) + return append_from_pair(store, Shape(**base), summary="s", cwe="CWE-78", pair="p1", provenance="human", tier="seeded") # type: ignore[arg-type] + + +def test_mechanism_edit_is_a_bounded_lever_with_a_stable_key(tmp_path: Path) -> None: + from openultrasast.improve.validator import VALID_LEVERS, MechanismEdit + + assert "mechanisms" in VALID_LEVERS + edit = MechanismEdit(action="admit", mechanism_id="corpus:abc", source="loo", rationale="recovers holdout pair x") + assert edit.lever == "mechanisms" and edit.key() == "mechanisms:admit:corpus:abc" + + +def test_validator_accepts_only_exporter_records_with_closed_guards(tmp_path: Path) -> None: + from openultrasast.improve.validator import EvolveValidator, MechanismEdit, StrictValidationError + + candidates = MechanismStore(tmp_path / "candidates.jsonl") + good = _record(candidates) + validator = EvolveValidator() + by_id = {record.id: record for record in candidates.load()} + validator.validate_mechanism(MechanismEdit(action="admit", mechanism_id=good.id, source="loo"), by_id) + validator.validate_mechanism(MechanismEdit(action="retract", mechanism_id=good.id, source="loo"), by_id) + with pytest.raises(StrictValidationError, match="unknown mechanism"): + validator.validate_mechanism(MechanismEdit(action="admit", mechanism_id="corpus:not-exported", source="loo"), by_id) + with pytest.raises(StrictValidationError, match="action"): + validator.validate_mechanism(MechanismEdit(action="rewrite", mechanism_id=good.id, source="loo"), by_id) + with pytest.raises(StrictValidationError, match="origin"): + from openultrasast.semantic.mechanisms import append_mechanism + + sandbox = append_mechanism(candidates, summary="s", cwe="CWE-78", language="python", tags=(), what_made_it_exploitable="x") + validator.validate_mechanism( + MechanismEdit(action="admit", mechanism_id=sandbox.id, source="loo"), {r.id: r for r in candidates.load()} + ) + # free-form shapes never pass: a record whose shape text is not a closed structure is rejected before it can be admitted + from dataclasses import replace + + forged = replace(good, shape={**(good.shape or {}), "guard": "if user not in ALLOWED"}) + with pytest.raises(StrictValidationError, match="guard"): + validator.validate_mechanism(MechanismEdit(action="admit", mechanism_id=forged.id, source="export"), {forged.id: forged}) + forged_text = replace(good, shape={**(good.shape or {}), "sink_name": "os.system(cmd + ' ')"}) + with pytest.raises(StrictValidationError, match="identifier"): + validator.validate_mechanism( + MechanismEdit(action="admit", mechanism_id=forged_text.id, source="export"), {forged_text.id: forged_text} + ) + + +def test_apply_mechanism_edits_admits_into_the_scan_store_and_reverts_byte_for_byte(tmp_path: Path) -> None: + from openultrasast.improve.validator import MechanismEdit, apply_mechanism_edits + + candidates = MechanismStore(tmp_path / "candidates.jsonl") + a = _record(candidates) + b = _record(candidates, sink_name="popen") + scan_store = tmp_path / "scan" / "mechanisms.jsonl" + scan_store.parent.mkdir() + scan_store.write_text("") + before = scan_store.read_bytes() + snapshot = apply_mechanism_edits([MechanismEdit(action="admit", mechanism_id=a.id, source="loo")], candidates, scan_store) + admitted = MechanismStore(scan_store).load() + assert [r.id for r in admitted] == [a.id] and admitted[0].origin == "corpus" + apply_mechanism_edits([MechanismEdit(action="retract", mechanism_id=a.id, source="loo")], candidates, scan_store) + assert MechanismStore(scan_store).load() == () # retraction is a tombstone row; load() folds it away + snapshot.restore() + assert scan_store.read_bytes() == before # byte-for-byte revert of the whole round + assert b.id != a.id