From cec1c4a42811d4a619c501233fdab6dff2c5d7ce Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Wed, 16 Sep 2026 19:59:12 +0200 Subject: [PATCH] feat(pre-push-safety-net): render the experimental evaluation explanation (M1b 10.10) The M1b exit run established the comparison but could not show a developer-readable consequence or repair direction, because both belong to an evaluated capability declaration and the installed registry is deliberately empty. The experimental evaluation output now renders, for each change-attributed finding, its family, exact location, witness, change attribution, provenance, engine method and operation, together with a consequence and repair direction produced from a supplied declaration through the same grounded-template path the admitted case uses. No prose is invented: when no declaration matches, or a template is not grounded in the operation and context, or the declaration does not cover the operation symbol, the explanation is absent and the reason is stated. Declarations reach this rendering only. The runner never passes them to admission, so nothing in that file can enable a capability or emit an alert however it is written, and a control asserts that even a declaration claiming PASS admits nothing. The flag requires explicit base/head replay and the recorded-veto flag. The evaluation instrument generates its own unreviewed declaration bound to the run's analysis semantics, asserting that it is neither enabled nor PASS, so the demonstration cannot drift into a qualified claim. 1,377 tests passed, nine skipped. Rollout stays NO-GO. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo --- benchmarks/push/m1a_record_vetoes.py | 47 +++++++++++++ src/openultrasast/cli.py | 17 ++++- src/openultrasast/push/report.py | 7 +- src/openultrasast/push/runner.py | 19 +++++- src/openultrasast/push/vetoes.py | 98 ++++++++++++++++++++++++++++ tests/test_push_vetoes.py | 85 +++++++++++++++++++++++- 6 files changed, 266 insertions(+), 7 deletions(-) diff --git a/benchmarks/push/m1a_record_vetoes.py b/benchmarks/push/m1a_record_vetoes.py index 326d13d..4b9423a 100644 --- a/benchmarks/push/m1a_record_vetoes.py +++ b/benchmarks/push/m1a_record_vetoes.py @@ -27,6 +27,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parent)) from measure import git, pin_snapshot # noqa: E402 from openultrasast.model.scan import ScanBudget # noqa: E402 +from openultrasast.push.policy import CapabilityAdmission, CapabilityKey # noqa: E402 from openultrasast.push.runner import _provenance # noqa: E402 from openultrasast.push.vetoes import FLAG, LABEL # noqa: E402 from openultrasast.push_inputs import validate_manifest # noqa: E402 @@ -61,6 +62,42 @@ PLAN: tuple[dict[str, Any], ...] = ( ) +# Unreviewed evaluation input for the M1b explanation rendering. It is bound to the run's own +# analysis semantics, marked experimental and never enabled, so it cannot qualify a capability or +# emit an alert: the runner keeps it out of admission entirely. The wording describes exactly the +# mechanism the graph established, evaluation of a request field as JavaScript, and the repair the +# pinned source documents immediately below the eval calls. +DECLARATION = { + "evaluation_id": "unreviewed-m1b-demonstration-v1", + "evaluation_artifact": "experimental://m1b-recorded-vetoes", + "verdict": "experimental", + "consequence_template": ( + "A request field reaches {operation}, where it is evaluated as JavaScript, so an attacker " + "controlling that field can execute arbitrary code in the server process ({context} capability)." + ), + "repair_template": ( + "Replace the evaluation at {operation} with a numeric conversion such as Number(...) or " + "parseInt(...), then validate the result before use ({context} capability)." + ), + "operation_symbols": ("eval", "const preTax = eval", "const afterTax = eval", "const roth = eval"), +} + + +def declarations_for(semantics: str) -> list[dict[str, Any]]: + key = CapabilityKey("javascript", "unspecified", "unspecified", "injection", "taint", "unreviewed", semantics) + declaration = CapabilityAdmission( + key, + DECLARATION["evaluation_id"], + DECLARATION["evaluation_artifact"], + "experimental", + DECLARATION["consequence_template"], + DECLARATION["repair_template"], + operation_symbols=tuple(DECLARATION["operation_symbols"]), + ) + assert not declaration.enabled and declaration.verdict != "PASS", "the demonstration declaration must stay unqualified" + return [declaration.to_payload()] + + def write(path: Path, data: object) -> None: path.write_text(json.dumps(data, indent=2, allow_nan=False) + "\n") @@ -108,6 +145,7 @@ def summarize(case: dict[str, Any], artifact: dict[str, Any]) -> dict[str, Any]: "production_finding_status": artifact["result"]["finding_status"], "production_coverage_status": artifact["result"]["coverage_status"], "recorded_status": experimental.get("status"), + "evaluation": experimental.get("evaluation", []), "recorded_target_findings": [ { "site": f["site"], @@ -169,9 +207,16 @@ def main() -> int: pins = {sid: pin_snapshot(bare, recipe.commit, snapshots[sid]) for sid in needed} settings = ScanBudget(max_model_calls=0, max_regions=500, order_by_evidence=True) installed = _provenance(bare, settings) + declarations_path = out / "experimental-declarations.json" + write(declarations_path, declarations_for(installed["semantics"])) record: dict[str, Any] = { "schema_version": 1, "label": LABEL, + "declarations": declarations_path.name, + "declaration_note": ( + "Unreviewed experimental declaration, bound to this run's semantics, used only to render the " + "evaluation explanation. It is not enabled and not PASS, and the runner keeps it out of admission." + ), "milestone": "M1a", "flag": FLAG, "development_budget_seconds": args.deadline, @@ -212,6 +257,8 @@ def main() -> int: "--cache-dir", str(out / "cache"), FLAG, + "--experimental-declarations", + str(declarations_path), ] started = time.monotonic() process = subprocess.run(command, capture_output=True, text=True, timeout=args.deadline + 30) diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index c096828..a7e76ce 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -132,6 +132,12 @@ def main(argv: list[str] | None = None) -> int: replay_parser.add_argument("--max-regions", type=int, default=500) replay_parser.add_argument("--mode", choices=("advisory", "blocking"), default="advisory") replay_parser.add_argument("--incomplete-coverage", choices=("allow", "block"), default="allow") + replay_parser.add_argument( + "--experimental-declarations", + type=Path, + help="explicit --base/--head replay only: unreviewed capability declarations used to render the experimental " + "evaluation explanation; never consulted by admission and cannot enable any capability", + ) replay_parser.add_argument( "--experimental-record-vetoes", action="store_true", @@ -243,9 +249,18 @@ def main(argv: list[str] | None = None) -> int: parser.error("--prior-hook requires Git stdin mode") if args.experimental_record_vetoes and not args.base: parser.error("--experimental-record-vetoes requires explicit --base/--head replay; it is never a hook capability") + if args.experimental_declarations and not args.experimental_record_vetoes: + parser.error("--experimental-declarations requires --experimental-record-vetoes") prior_data = None if args.base: - delivery = replay(args.path, base=args.base, head=args.head, record_vetoes=args.experimental_record_vetoes, **options) + delivery = replay( + args.path, + base=args.base, + head=args.head, + record_vetoes=args.experimental_record_vetoes, + declarations=args.experimental_declarations, + **options, + ) else: # multiprocessing closes sys.stdin in its child; duplicate the Git pipe first. with os.fdopen(os.dup(0), "rb") as stream: diff --git a/src/openultrasast/push/report.py b/src/openultrasast/push/report.py index 9340f4c..4f4a31b 100644 --- a/src/openultrasast/push/report.py +++ b/src/openultrasast/push/report.py @@ -202,9 +202,12 @@ def _experimental_line(experimental: Mapping[str, object]) -> str: vetoes = sum(len(f.get("vetoes", [])) for f in findings if isinstance(f, dict)) if isinstance(findings, list) else 0 base = experimental.get("base_only_findings") quiet = len(base) if isinstance(base, list) else 0 + evaluation = experimental.get("evaluation") + explained = len(evaluation) if isinstance(evaluation, list) else 0 return ( - f"recorded-veto evaluation retained {count} raw finding(s) with {vetoes} recorded veto(es) and " - f"{quiet} base-only finding(s) in the detailed result; recorded, not applied; not an alert." + f"recorded-veto evaluation retained {count} raw finding(s) with {vetoes} recorded veto(es), " + f"{quiet} base-only finding(s) and {explained} change-attributed evaluation finding(s) in the detailed " + f"result; recorded, not applied; not an alert." ) diff --git a/src/openultrasast/push/runner.py b/src/openultrasast/push/runner.py index ce1175f..f0b0020 100644 --- a/src/openultrasast/push/runner.py +++ b/src/openultrasast/push/runner.py @@ -42,7 +42,7 @@ from openultrasast.push.policy import ( from openultrasast.push.report import PushReport, ReportDelivery, ScanRecord, _completed_scan, deliver_report, render_report from openultrasast.push.reuse import ReusingBackend, declaration_identity, discovery from openultrasast.push.snapshot import SnapshotAdapter -from openultrasast.push.vetoes import FLAG, LABEL, record_vetoes +from openultrasast.push.vetoes import FLAG, LABEL, load_declarations, record_vetoes _T = TypeVar("_T") @@ -179,9 +179,15 @@ def _record( reasons: Sequence[str], admission: AdmissionResult, provenance: dict[str, str], + declarations: Path | None = None, ) -> dict[str, object]: - """Experimental M1a: the same policy over lifted evidence, reported only, never admitted.""" + """Experimental M1a: the same policy over lifted evidence, reported only, never admitted. + + `declarations` supplies templates for the experimental evaluation rendering ONLY. They are + never passed to admission, so nothing read from that file can enable a capability. + """ registry = load_registry(current=provenance) + supplied = load_declarations(declarations) if declarations is not None else () report = record_vetoes( head=head, base=delta.base_scan, @@ -192,6 +198,7 @@ def _record( production_admission={d.candidate.delta.defect_id: d.reasons for d in admission.dispositions}, coverage_reasons=tuple(dict.fromkeys(reasons)), capabilities=registry.capabilities, + declarations=supplied, ) return report.to_payload() @@ -216,6 +223,7 @@ def _analyze( execution_budget: ExecutionBudget, resolved: PushComparison | None = None, record_vetoes: bool = False, + declarations: Path | None = None, ) -> PushReport: """Run a single explicit comparison; exit policy never stands for coverage. @@ -380,7 +388,8 @@ def _analyze( stage, stage_started = "recorded_vetoes", time.monotonic() assert context is not None recorded = _prepare( - lambda: _record(head_scan, delta, context, comparison, tuple(reasons), admission, provenance), budget + lambda: _record(head_scan, delta, context, comparison, tuple(reasons), admission, provenance, declarations), + budget, ) experimental = {**recorded, "status": "evaluated"} timings[stage + "_seconds"] = time.monotonic() - stage_started @@ -447,8 +456,11 @@ def replay( cache_dir: Path | None = None, model_config: Path | None = None, record_vetoes: bool = False, + declarations: Path | None = None, ) -> ReportDelivery: config = config or PushConfig() + if declarations is not None and not record_vetoes: + raise ValueError("experimental declarations require the recorded-veto evaluation flag") budget = ExecutionBudget(time.monotonic() + config.deadline_seconds, config.cancellation_allowance_seconds) report = _analyze( repository, @@ -461,6 +473,7 @@ def replay( cache_dir=cache_dir, execution_budget=budget, record_vetoes=record_vetoes, + declarations=declarations, ) from openultrasast.push.assistance import assist diff --git a/src/openultrasast/push/vetoes.py b/src/openultrasast/push/vetoes.py index 1593537..a869a45 100644 --- a/src/openultrasast/push/vetoes.py +++ b/src/openultrasast/push/vetoes.py @@ -18,6 +18,7 @@ import json from collections import Counter from collections.abc import Sequence from dataclasses import dataclass, replace +from pathlib import Path from openultrasast.contracts import Contract from openultrasast.model.contracts import ChangeContext, ExecutionBudget, QuestionIdentity @@ -33,6 +34,7 @@ from openultrasast.push.policy import ( _compare_evidence, _context_boundaries, _counterpart, + _grounded_template, admit_candidates, operation_provenance, ) @@ -134,6 +136,36 @@ class BaseOnlyFinding(Contract): head_counterparts: tuple[QuestionReceipt, ...] +@dataclass(frozen=True) +class EvaluationFinding(Contract): + """A developer-readable rendering of one evaluation finding, for reading, never for alerting. + + Consequence and repair come from a supplied declaration's templates through the same + grounded-template path the admitted case uses, so no prose is invented here. When no + declaration matches, both are absent with the reason stated. This record is produced outside + admission and cannot become an alert. + """ + + family: str + path: str + line: int + novelty: str + reason: str + witness: str + change_evidence: str + provenance: str + question_function: str | None + engine_method: str | None + operation: str + operation_symbol: str + consequence: str | None + repair: str | None + unavailable: tuple[str, ...] + declaration: str | None + declaration_verdict: str | None + declaration_enabled: bool + + @dataclass(frozen=True) class RecordedVetoReport(Contract): label: str @@ -145,6 +177,7 @@ class RecordedVetoReport(Contract): base_only_findings: tuple[BaseOnlyFinding, ...] lifted_coverage_reasons: tuple[str, ...] remaining_coverage_reasons: tuple[str, ...] + evaluation: tuple[EvaluationFinding, ...] = () def boundary_kind(boundary: str) -> str | None: @@ -314,6 +347,67 @@ def _finding_vetoes( return tuple(vetoes) +def load_declarations(path: Path) -> tuple[CapabilityAdmission, ...]: + """Read experimental declarations for the evaluation rendering only. + + These never reach admission. The installed registry remains the sole source of eligibility, + so a declaration read here cannot enable a capability or emit an alert however it is written. + """ + content = path.read_bytes()[: 1024**2 + 1] + if len(content) > 1024**2: + raise ValueError("declaration_size_limit") + payload = json.loads(content) + if not isinstance(payload, list): + raise ValueError("declarations must be a list") + return tuple(CapabilityAdmission.from_payload(item) for item in payload) + + +def _evaluation(delta: CandidateDelta, declarations: Sequence[CapabilityAdmission], semantics: str) -> EvaluationFinding | None: + """Render one established finding. Returns None for anything not attributed to the change.""" + op = delta.head_operation + if op is None or delta.novelty not in ("new", "worsened") or delta.witness is None or not delta.change_evidence: + return None + key = CapabilityKey(op.question.language, "unspecified", "unspecified", delta.family, op.mechanism, "unreviewed", semantics) + matching = [declaration for declaration in declarations if declaration.key == key] + symbol = op.operation.partition("(")[0].strip() + unavailable: list[str] = [] + consequence = repair = None + declaration = matching[0] if len(matching) == 1 else None + if not matching: + unavailable.append("no_matching_declaration") + elif len(matching) > 1: + unavailable.append("ambiguous_declaration") + if declaration is not None: + consequence = _grounded_template(declaration.consequence_template, op, key.context) + repair = _grounded_template(declaration.repair_template, op, key.context) + if consequence is None: + unavailable.append("consequence_template_ungrounded") + if repair is None: + unavailable.append("repair_template_ungrounded") + if symbol not in declaration.operation_symbols: + unavailable.append("declaration_does_not_cover_operation_symbol") + return EvaluationFinding( + delta.family, + op.path, + op.line, + delta.novelty, + delta.reason, + delta.witness, + delta.change_evidence[0], + operation_provenance(op), + op.question.function, + op.method, + op.operation, + symbol, + consequence, + repair, + tuple(unavailable), + declaration.evaluation_id if declaration else None, + declaration.verdict if declaration else None, + bool(declaration.enabled) if declaration else False, + ) + + def record_vetoes( *, head: ModelScanResult, @@ -325,6 +419,7 @@ def record_vetoes( production_admission: dict[str, tuple[str, ...]], coverage_reasons: Sequence[str], capabilities: Sequence[CapabilityAdmission] = (), + declarations: Sequence[CapabilityAdmission] = (), execution_budget: ExecutionBudget | None = None, ) -> RecordedVetoReport: """Re-run the production comparison and admission over lifted evidence and list what was lifted. @@ -431,4 +526,7 @@ def record_vetoes( tuple(base_only), tuple(dict.fromkeys((*lifted_reasons, *lifted_boundaries))), remaining, + # Rendered from the PRODUCTION comparison, so the demonstration shows what the production + # rules established, not what the lifted view would have. + tuple(item for delta in production for item in (_evaluation(delta, declarations, semantics),) if item is not None), ) diff --git a/tests/test_push_vetoes.py b/tests/test_push_vetoes.py index 7c8e0d5..a8c3b42 100644 --- a/tests/test_push_vetoes.py +++ b/tests/test_push_vetoes.py @@ -211,9 +211,11 @@ def test_report_requires_label_and_renders_experimental_line(tmp_path): base = report_with() with pytest.raises(ValueError, match="experimental label"): replace(base, experimental={"status": "evaluated"}) - labeled = replace(base, experimental={"label": LABEL, "flag": FLAG, "status": "evaluated", "findings": [{"vetoes": [{}, {}]}]}) + section = {"label": LABEL, "flag": FLAG, "status": "evaluated", "findings": [{"vetoes": [{}, {}]}], "evaluation": [{}]} + labeled = replace(base, experimental=section) text = render_report(labeled, artifact=None) assert "Experimental: recorded-veto evaluation retained 1 raw finding(s) with 2 recorded veto(es)" in text + assert "1 change-attributed evaluation finding(s)" in text assert "not an alert" in text pending = replace(base, experimental={"label": LABEL, "flag": FLAG, "status": "not_evaluated", "reason": "head_failed:TimeoutError"}) assert "was not produced (head_failed:TimeoutError)" in render_report(pending, artifact=None) @@ -264,3 +266,84 @@ def test_cli_refuses_the_flag_outside_explicit_replay(tmp_path, capsys): ] ) assert code == 0 and "Experimental:" in capsys.readouterr().out + + +def declaration(*, symbols=("eval",), consequence="Input reaching {operation} is evaluated in {context}.", verdict="experimental"): + from openultrasast.push.policy import CapabilityAdmission, CapabilityKey + + key = CapabilityKey("javascript", "unspecified", "unspecified", "injection", "taint", "unreviewed", "same") + return CapabilityAdmission( + key, + "unreviewed-demonstration-v1", + "experimental://unreviewed", + verdict, + consequence, + "At {operation}, convert the value before use in {context}.", + operation_symbols=symbols, + ) + + +def rendered(*, declarations=(), base=None): + head = scan() + base = base if base is not None else scan(rows=[], findings=False) + applied, admission = production(head, base, context()) + return record_vetoes( + head=head, + base=base, + context=context(), + comparison=COMPARISON, + semantics="same", + production=applied, + production_admission=admission, + coverage_reasons=(), + declarations=declarations, + ) + + +def test_evaluation_renders_consequence_and_repair_from_a_declaration(): + """M1b 10.10: the explanation comes from a declaration's templates, never from invented prose.""" + report = rendered(declarations=(declaration(),)) + item = report.evaluation[0] + assert item.novelty == "new" and item.family == "injection" and (item.path, item.line) == ("a.js", 4) + assert item.witness == "req.body -> exec(cmd) in a.js (line 4, 2 steps)" + assert item.change_evidence and item.provenance == "exact" and item.engine_method == "handler" + assert item.consequence == "Input reaching exec(cmd) is evaluated in unreviewed." + assert item.repair == "At exec(cmd), convert the value before use in unreviewed." + assert item.declaration == "unreviewed-demonstration-v1" + assert item.declaration_verdict == "experimental" and item.declaration_enabled is False + + +def test_evaluation_states_why_an_explanation_is_unavailable(): + report = rendered() + item = report.evaluation[0] + assert item.consequence is None and item.repair is None + assert item.unavailable == ("no_matching_declaration",) + + +def test_an_ungrounded_template_is_refused_rather_than_rendered(): + report = rendered(declarations=(declaration(consequence="Sanitize your input."),)) + item = report.evaluation[0] + assert item.consequence is None and "consequence_template_ungrounded" in item.unavailable + assert item.repair is not None + + +def test_a_declaration_not_covering_the_operation_symbol_says_so(): + report = rendered(declarations=(declaration(symbols=("query",)),)) + assert "declaration_does_not_cover_operation_symbol" in report.evaluation[0].unavailable + assert report.evaluation[0].operation_symbol == "exec" + + +def test_only_a_change_attributed_finding_is_rendered(): + # The anchor maps head line 4 to base line 5, so this base carries the same operation. + unchanged = rendered(declarations=(declaration(),), base=scan(line=5)) + assert unchanged.findings[0].production_novelty == "unchanged" + assert unchanged.evaluation == () + + +def test_an_enabled_passing_declaration_here_still_admits_nothing(): + """The rendering path is outside admission, so no supplied declaration can emit an alert.""" + report = rendered(declarations=(declaration(verdict="PASS"),)) + assert report.evaluation[0].declaration_verdict == "PASS" + payload = report.to_payload() + assert "defects" not in payload + assert all("admitted" not in item for item in payload["evaluation"])