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"])