From f04955c36784942eb5110b68cd8dcce720962bab Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 13 Sep 2026 17:45:16 +0200 Subject: [PATCH] feat(pre-push-safety-net): execute full Git push transactions --- .../implementation-group-7.md | 37 +++++ .kiro/specs/pre-push-safety-net/spec.json | 19 ++- .kiro/specs/pre-push-safety-net/tasks.md | 2 +- src/openultrasast/cli.py | 36 ++-- src/openultrasast/push/report.py | 2 + src/openultrasast/push/runner.py | 157 ++++++++++++++++-- tests/test_push_interface.py | 150 +++++++++++++++++ 7 files changed, 370 insertions(+), 33 deletions(-) create mode 100644 .kiro/specs/pre-push-safety-net/implementation-group-7.md create mode 100644 tests/test_push_interface.py diff --git a/.kiro/specs/pre-push-safety-net/implementation-group-7.md b/.kiro/specs/pre-push-safety-net/implementation-group-7.md new file mode 100644 index 0000000..7324ee0 --- /dev/null +++ b/.kiro/specs/pre-push-safety-net/implementation-group-7.md @@ -0,0 +1,37 @@ +# Group 7 implementation and verification + +The maintainer authorized all of group 7 on 2026-09-13. Fresh agent dispatch failed +with `agent thread limit reached`; kiro-impl's explicit manual fallback applies. +Reviews here are main-context reviews, not independent reviewer evidence. +Rollout remains NO-GO and the default capability registry remains empty. + +## Task 7.1 brief + +Requirements 1.1, 1.3, 1.4, 4.1, 5.1–5.5 and 7.1; design: Snapshot Adapter, +Push runner, Result and Hook Adapter. Consume every Git update and remote identity, +resolve exact comparisons using the existing adapter, share one absolute budget, +retain independent finding/coverage/enforcement statuses and all comparison evidence. +Verify real stdin delivery, immutable readable snapshots, deduplication, missing bases, +deletions, no work, policy combinations and deadline ownership. + +## Status Report +- STATUS: READY_FOR_REVIEW +- TASK: 7.1 +- RED_PHASE_OUTPUT: `/tmp/group7-1-red.log`: four failures with the interface switch OFF. ON: 37 focused tests passed; switch removed, final focused run: 29 passed (including 11 transaction controls). +- TESTS_RUN: full suite 1274 passed, 9 skipped in 109.09s; final targeted tests cover the subsequently added five policy combinations and retained resolved identity on preparation failure. Ruff and mypy pass. +- CONCERNS: synthetic admission controls prove enforcement, not detector eligibility. No capability enabled. + +## Review Verdict +- VERDICT: APPROVED +- TASK: 7.1 +- MECHANICAL_RESULTS: actual diff reviewed; full suite and final targeted checks pass; no new placeholders or credentials; boundary restricted to CLI, push runner and report. +- FINDINGS: fixed child stdin closure by duplicating Git's descriptor before forking; remote URL is hashed rather than storing possible credentials; unresolved comparison identity survives preparation failure. +- RED phase: VERIFIED. +- SUMMARY: One transaction retains every resolved ref and completed scan without resetting the analysis deadline. + +## Verification Result +- STATUS: VERIFIED +- CLAIM_TYPE: TASK +- CLAIM: Task 7.1 interface and policy integration implemented. +- EVIDENCE: `/tmp/group7-1-full.log`, `/tmp/group7-1-final-focused.log`; ruff, mypy and diff checks exit 0. +- GAPS: Representative multi-ref performance and real capability admission remain group 8 work. diff --git a/.kiro/specs/pre-push-safety-net/spec.json b/.kiro/specs/pre-push-safety-net/spec.json index 931f8ce..c5ceda0 100644 --- a/.kiro/specs/pre-push-safety-net/spec.json +++ b/.kiro/specs/pre-push-safety-net/spec.json @@ -1,7 +1,7 @@ { "feature_name": "pre-push-safety-net", "created_at": "2026-09-12T15:35:00Z", - "updated_at": "2026-09-13T15:06:56.856952+00:00", + "updated_at": "2026-09-13T15:45:15.902603+00:00", "language": "en", "phase": "implementation", "approvals": { @@ -19,7 +19,7 @@ } }, "ready_for_implementation": true, - "implementation_approval_note": "Groups 1\u20136 implemented and verified (20/27); group 6 authorized as a whole. Main-context Kiro review fallback after fresh dispatch thread limit. Valid reuse is demonstrated; real capability registry remains empty and rollout gates remain unmet. Groups 7\u20138 pending.", + "implementation_approval_note": "Group 7 authorized as a whole; 7.1 verified, 7.2\u20137.3 in progress. Default capability registry empty; rollout NO-GO.", "depends_on": [ "contributor-scan" ], @@ -70,6 +70,11 @@ "at": "2026-09-13T13:58:52.336506+00:00", "source": "continue with group 6 as a whole. do we need to amend the spec to get a go for the rollout", "action": "Authorized group 6 (6.1\u20136.4). Clarified complete preparation/discovery reuse after measured group 5 bottleneck; rollout gates remain unchanged. Manual Kiro review fallback remains necessary after a fresh agent dispatch hit the thread limit." + }, + { + "at": "2026-09-13T15:45:15.902603+00:00", + "source": "ok, continue with group 7 as a whole", + "action": "Authorized tasks 7.1\u20137.3; manual Kiro fallback after fresh dispatch hit thread limit. Group 8 and rollout remain pending." } ], "task_review": { @@ -102,17 +107,19 @@ "6.1", "6.2", "6.3", - "6.4" + "6.4", + "7.1" ], - "remaining_tasks": 7, - "verification_record": "implementation-group-6.md", + "remaining_tasks": 6, + "verification_record": "implementation-group-7.md", "verification_records": [ "implementation-group-1.md", "implementation-group-2.md", "implementation-group-3.md", "implementation-group-4.md", "implementation-group-5.md", - "implementation-group-6.md" + "implementation-group-6.md", + "implementation-group-7.md" ], "next_group": 7, "blocked_tasks": [], diff --git a/.kiro/specs/pre-push-safety-net/tasks.md b/.kiro/specs/pre-push-safety-net/tasks.md index 9b77162..46e7fa5 100644 --- a/.kiro/specs/pre-push-safety-net/tasks.md +++ b/.kiro/specs/pre-push-safety-net/tasks.md @@ -191,7 +191,7 @@ independent capability qualification. Groups 7–8 remain pending; rollout remai - _Requirements: 1.1, 2.2, 4.1, 4.2, 4.3, 4.4, 7.3_ - [ ] 7. Connect the real push interface without adding noise -- [ ] 7.1 Execute Git's full push input and enforcement policy +- [x] 7.1 Execute Git's full push input and enforcement policy - Extend the replay command to consume all update lines and remote arguments, aggregating per-ref comparisons under one deadline and preserving independent coverage and finding statuses. - Apply advisory, opt-in blocking and separate incomplete-coverage policy; keep external services disabled by default and normal alerts limited to admitted capabilities. - Done when controlled multi-ref push invocations exercise all allow/block/status combinations, including deletion-only, missing-base and no-work cases, without resetting budgets. diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index de143fc..509658b 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse import json +import os import tempfile from collections.abc import Mapping, Sequence from dataclasses import asdict, dataclass, replace @@ -110,10 +111,12 @@ def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="ousast") subparsers = parser.add_subparsers(dest="command", required=True) - replay_parser = subparsers.add_parser("pre-push", help="replay explicit local base/head revisions (experimental)") + replay_parser = subparsers.add_parser("pre-push", help="check Git push input or replay local revisions (experimental)") replay_parser.add_argument("path", type=Path) - replay_parser.add_argument("--base", required=True) - replay_parser.add_argument("--head", required=True) + replay_parser.add_argument("--base") + replay_parser.add_argument("--head") + replay_parser.add_argument("--remote", nargs=2, metavar=("NAME", "URL")) + replay_parser.add_argument("--comparison-base") replay_parser.add_argument("--artifact", type=Path, required=True) replay_parser.add_argument( "--cache-dir", type=Path, help="reuse compatible local artifacts in a private directory outside the repository" @@ -205,24 +208,31 @@ def main(argv: list[str] | None = None) -> int: args = parser.parse_args(argv) if args.command == "pre-push": from .config import PushConfig - from .push.runner import replay + from .push.runner import push, replay try: + if bool(args.base) != bool(args.head) or bool(args.base) == bool(args.remote): + parser.error("provide --base and --head together, or --remote NAME URL for Git stdin") settings = PushConfig( + comparison_base=args.comparison_base, deadline_seconds=args.deadline, cancellation_allowance_seconds=args.cancellation_allowance, mode=args.mode, incomplete_coverage_policy=args.incomplete_coverage, ) - delivery = replay( - args.path, - base=args.base, - head=args.head, - artifact=args.artifact, - config=settings, - max_regions=args.max_regions, - cache_dir=args.cache_dir, - ) + options: dict[str, Any] = dict(artifact=args.artifact, config=settings, max_regions=args.max_regions, cache_dir=args.cache_dir) + if args.base: + delivery = replay(args.path, base=args.base, head=args.head, **options) + else: + # multiprocessing closes sys.stdin in its child; duplicate the Git pipe first. + with os.fdopen(os.dup(0), encoding="utf-8", errors="strict") as stream: + delivery = push( + args.path, + updates=lambda: stream.read(1024 * 1024 + 1), + remote_name=args.remote[0], + remote_url=args.remote[1], + **options, + ) except ValueError as error: parser.error(str(error)) print(delivery.text, end="") diff --git a/src/openultrasast/push/report.py b/src/openultrasast/push/report.py index b0ba43b..34c76c5 100644 --- a/src/openultrasast/push/report.py +++ b/src/openultrasast/push/report.py @@ -38,6 +38,7 @@ class PushReport: resolution: PushResolution | None = None snapshots: tuple[SnapshotManifest, ...] = () change_context: ChangeContext | None = None + change_contexts: tuple[ChangeContext, ...] = () def __post_init__(self) -> None: object.__setattr__(self, "provenance", MappingProxyType(dict(self.provenance))) @@ -111,6 +112,7 @@ def _write_artifact(report: PushReport, target: Path, temporary: Path, connectio with os.fdopen(fd, "w", encoding="utf-8") as stream: payload = { "schema_version": 1, + "change_contexts": [item.to_payload() for item in report.change_contexts], "snapshots": [item.to_payload() for item in report.snapshots], "change_context": report.change_context.to_payload() if report.change_context else None, "result": report.result.to_payload(), diff --git a/src/openultrasast/push/runner.py b/src/openultrasast/push/runner.py index 67e1987..69a01ef 100644 --- a/src/openultrasast/push/runner.py +++ b/src/openultrasast/push/runner.py @@ -25,8 +25,9 @@ from openultrasast.model.scan import ScanBudget, scan_repository from openultrasast.model.shipped import declared_sources from openultrasast.preprocess import build_file_target, enumerate_source_files from openultrasast.push.cache import ArtifactCache, SemanticKeys -from openultrasast.push.contracts import ComparisonAnalysis, SnapshotManifest +from openultrasast.push.contracts import ComparisonAnalysis, PushComparison, PushResult, SnapshotManifest from openultrasast.push.policy import ( + ActionableDefect, AdmissionCandidate, AdmissionResult, CapabilityKey, @@ -162,7 +163,7 @@ def _discovery_identity() -> str: return digest_value([(p.relative_to(root).as_posix(), hashlib.sha256(p.read_bytes()).hexdigest()) for p in sorted(root.rglob("*.py"))]) -def replay( +def _analyze( repository: Path, *, base: str, @@ -172,7 +173,9 @@ def replay( backend: Any = None, max_regions: int = 500, cache_dir: Path | None = None, -) -> ReportDelivery: + execution_budget: ExecutionBudget, + resolved: PushComparison | None = None, +) -> PushReport: """Run a single explicit comparison; exit policy never stands for coverage.""" config = config or PushConfig() if max_regions < 1: @@ -180,7 +183,7 @@ def replay( if cache_dir is not None and cache_dir.resolve().is_relative_to(repository.resolve()): raise ValueError("cache directory must be outside the analyzed repository") started = time.monotonic() - budget = ExecutionBudget(started + config.deadline_seconds, config.cancellation_allowance_seconds) + budget = execution_budget scan_budget = ScanBudget(max_model_calls=0, max_regions=max_regions, order_by_evidence=True) if backend is None: backend = JoernBackend() @@ -191,7 +194,7 @@ def replay( manifests: list[SnapshotManifest] = [] reasons: list[str] = [] admission = AdmissionResult((), (), ()) - comparison = None + comparison = resolved context = None timings: dict[str, float] = { "deadline_seconds": config.deadline_seconds, @@ -203,7 +206,7 @@ def replay( stage_started = started try: adapter = SnapshotAdapter(repository, execution_budget=budget) - comparison = adapter.resolve_replay(base, head) + comparison = resolved or adapter.resolve_replay(base, head) timings[stage + "_seconds"] = time.monotonic() - stage_started stage, stage_started = "provenance", time.monotonic() provenance.update(_prepare(lambda: _provenance(repository, scan_budget), budget)) @@ -354,11 +357,139 @@ def replay( result = admission_push_result(admission, analyses, config=config) timings["total_seconds"] = time.monotonic() - started report = PushReport(result, admission, tuple(records), provenance, timings, snapshots=tuple(manifests), change_context=context) - try: - artifact.resolve().relative_to(repository.resolve()) - except ValueError: - pass - else: - write_error = "artifact_inside_repository" - return ReportDelivery(result, render_report(report, artifact=None, error=write_error), None, write_error) + return report + + +def _deliver(repository: Path, report: PushReport, artifact: Path, budget: ExecutionBudget) -> ReportDelivery: + if artifact.resolve().is_relative_to(repository.resolve()): + error = "artifact_inside_repository" + return ReportDelivery(report.result, render_report(report, artifact=None, error=error), None, error) return deliver_report(report, artifact, execution_budget=budget) + + +def replay( + repository: Path, + *, + base: str, + head: str, + artifact: Path, + config: PushConfig | None = None, + backend: Any = None, + max_regions: int = 500, + cache_dir: Path | None = None, +) -> ReportDelivery: + config = config or PushConfig() + budget = ExecutionBudget(time.monotonic() + config.deadline_seconds, config.cancellation_allowance_seconds) + report = _analyze( + repository, + base=base, + head=head, + artifact=artifact, + config=config, + backend=backend, + max_regions=max_regions, + cache_dir=cache_dir, + execution_budget=budget, + ) + return _deliver(repository, report, artifact, budget) + + +def push( + repository: Path, + *, + updates: str | Callable[[], str], + remote_name: str, + remote_url: str, + artifact: Path, + config: PushConfig | None = None, + backend: Any = None, + max_regions: int = 500, + cache_dir: Path | None = None, +) -> ReportDelivery: + """Consume one Git transaction. Resolution, every comparison and reporting share a deadline.""" + config = config or PushConfig() + if max_regions < 1: + raise ValueError("max_regions must be positive") + if cache_dir is not None and cache_dir.resolve().is_relative_to(repository.resolve()): + raise ValueError("cache directory must be outside the analyzed repository") + started = time.monotonic() + budget = ExecutionBudget(started + config.deadline_seconds, config.cancellation_allowance_seconds) + reports: list[PushReport] = [] + analyses: list[ComparisonAnalysis] = [] + reasons: list[str] = [] + resolution = None + try: + data = _prepare(updates, budget) if callable(updates) else updates + if len(data.encode("utf-8")) > 1024 * 1024: + raise ValueError("push_input_limit") + adapter = SnapshotAdapter(repository, comparison_base=config.comparison_base, execution_budget=budget) + resolution = adapter.resolve_updates(data) + reasons.extend("resolution:" + row.disposition for row in resolution.updates if row.disposition not in ("ready", "deleted")) + for comparison in resolution.comparisons: + if comparison.base_oid is None or time.monotonic() >= budget.deadline_monotonic: + analyses.append(ComparisonAnalysis(comparison, (), (), "unavailable")) + reasons.append("base_comparison_unavailable" if comparison.base_oid is None else "deadline_exhausted") + continue + report = _analyze( + repository, + base=comparison.base_oid, + head=comparison.head_oid, + artifact=artifact, + config=config, + backend=backend, + max_regions=max_regions, + cache_dir=cache_dir, + execution_budget=budget, + resolved=comparison, + ) + reports.append(report) + analyses.extend(report.result.analyses) + except Exception as error: + reasons.append("push_input_failed:" + type(error).__name__) + if time.monotonic() >= budget.deadline_monotonic: + reasons.append("deadline_exhausted") + defects: dict[str, ActionableDefect] = {} + for report in reports: + for defect in report.admission.defects: + previous = defects.get(defect.defect_id) + if previous is None: + defects[defect.defect_id] = defect + else: + fields = ("witnesses", "locations", "comparisons", "refs", "change_evidence", "consequences", "repairs", "evaluation_ids") + defects[defect.defect_id] = replace( + previous, + **cast(dict[str, Any], {key: tuple(dict.fromkeys((*getattr(previous, key), *getattr(defect, key)))) for key in fields}), + ) + admission = AdmissionResult( + tuple(defects.values()), + tuple(d for r in reports for d in r.admission.dispositions), + tuple(dict.fromkeys((*reasons, *(g for r in reports for g in r.admission.coverage_reasons)))), + ) + result = admission_push_result(admission, tuple(analyses), config=config) + if resolution is not None and not resolution.comparisons and not reasons: + result = PushResult((), "none", "not_applicable", "allow", ()) + provenance = dict(reports[0].provenance) if reports else {key: "not_run" for key in ("engine", "facts", "queries", "policy")} + # Remote URLs can contain credentials. Retain identity without persisting secrets. + provenance.update( + repository=str(repository.absolute()), + model="disabled", + remote_name=remote_name or "unnamed", + remote_url_sha256=hashlib.sha256(remote_url.encode()).hexdigest(), + ) + timings = {f"comparison_{i}_{k}": v for i, r in enumerate(reports) for k, v in r.timings.items()} + timings.update( + total_seconds=time.monotonic() - started, + deadline_seconds=config.deadline_seconds, + cancellation_allowance_seconds=config.cancellation_allowance_seconds, + ) + report = PushReport( + result, + admission, + tuple(s for r in reports for s in r.scans), + provenance, + timings, + resolution=resolution, + snapshots=tuple(s for r in reports for s in r.snapshots), + change_contexts=tuple(r.change_context for r in reports if r.change_context is not None), + ) + return _deliver(repository, report, artifact, budget) diff --git a/tests/test_push_interface.py b/tests/test_push_interface.py new file mode 100644 index 0000000..0571fd8 --- /dev/null +++ b/tests/test_push_interface.py @@ -0,0 +1,150 @@ +import json +import time + +import pytest +from test_push_runner import history + +from openultrasast.config import PushConfig +from openultrasast.cpg.backend import NullBackend +from openultrasast.push.runner import push + + +def invoke(tmp_path, root, updates, **kwargs): + return push( + root, + updates=updates, + remote_name="origin", + remote_url="https://secret@example.invalid/repo", + artifact=tmp_path / "push.json", + backend=NullBackend(), + **kwargs, + ) + + +@pytest.mark.parametrize("updates", ["", "(delete) {zero} refs/heads/old {base}\n"]) +def test_no_work(tmp_path, updates): + root, base, head, _ = history(tmp_path) + result = invoke(tmp_path, root, updates.format(zero="0" * 40, base=base)) + assert result.result.coverage_status == "not_applicable" + assert result.exit_code == 0 + + +def test_all_refs_deduplicated_and_missing_base_retained(tmp_path): + root, base, head, _ = history(tmp_path) + lines = "".join(f"refs/heads/local {head} refs/heads/{ref} {oid}\n" for ref, oid in [("one", base), ("two", base), ("new", "0" * 40)]) + result = invoke(tmp_path, root, lines, config=PushConfig(mode="blocking", incomplete_coverage_policy="block")) + data = json.loads((tmp_path / "push.json").read_text()) + assert result.exit_code == 1 + assert len(data["resolution"]["updates"]) == 3 + assert len(result.result.analyses) == 2 + assert result.result.analyses[0].comparison.refs == ("refs/heads/one", "refs/heads/two") + assert data["snapshots"][0]["bytes_read"] > 0 + assert len(data["change_contexts"]) == 1 + assert "secret@example.invalid" not in json.dumps(data) + + +def test_input_wait_is_inside_deadline(tmp_path): + root, _, _, _ = history(tmp_path) + + def delayed(): + time.sleep(5) + return "" + + start = time.monotonic() + result = invoke(tmp_path, root, delayed, config=PushConfig(deadline_seconds=0.1, cancellation_allowance_seconds=0.5)) + assert time.monotonic() - start < 0.8 + assert result.result.coverage_status == "unavailable" + assert result.exit_code == 0 + + +def test_distinct_comparisons_share_deadline(tmp_path, monkeypatch): + import openultrasast.push.runner as runner + + root, base, head, _ = history(tmp_path) + original = runner._analyze + budgets = [] + + def analyze(*args, **kwargs): + budgets.append(kwargs["execution_budget"]) + return original(*args, **kwargs) + + monkeypatch.setattr(runner, "_analyze", analyze) + result = invoke(tmp_path, root, f"refs/heads/a {head} refs/heads/a {base}\nrefs/heads/b {base} refs/heads/b {head}\n") + assert len(result.result.analyses) == 2 + assert len(budgets) == 2 and budgets[0] is budgets[1] + assert len(json.loads((tmp_path / "push.json").read_text())["change_contexts"]) == 2 + + +def test_cli_reads_git_stdin(tmp_path): + import subprocess + import sys + + root, base, _, _ = history(tmp_path) + result = subprocess.run( + [ + sys.executable, + "-m", + "openultrasast.cli", + "pre-push", + str(root), + "--remote", + "origin", + "/tmp/remote", + "--artifact", + str(tmp_path / "cli.json"), + ], + input=f"(delete) {'0' * 40} refs/heads/old {base}\n", + text=True, + capture_output=True, + timeout=10, + ) + assert result.returncode == 0, result.stderr + data = json.loads((tmp_path / "cli.json").read_text()) + assert data["result"]["coverage_status"] == "not_applicable" + assert len(data["resolution"]["updates"]) == 1 + + +@pytest.mark.parametrize( + "mode,strict,admitted,exit_code", + [ + ("advisory", "allow", True, 0), + ("blocking", "allow", True, 1), + ("advisory", "block", False, 0), + ("blocking", "allow", False, 0), + ("blocking", "block", False, 1), + ], +) +def test_transaction_enforcement_and_dedup(tmp_path, monkeypatch, mode, strict, admitted, exit_code): + from dataclasses import replace + + from test_push_report import report_with + + import openultrasast.push.runner as runner + + root, base, head, _ = history(tmp_path) + + def analyze(*args, **kwargs): + report = report_with(admitted=admitted) + comparison = kwargs["resolved"] + admission = replace( + report.admission, + defects=tuple(replace(d, comparisons=(comparison,), refs=comparison.refs) for d in report.admission.defects), + dispositions=tuple(replace(d, candidate=replace(d.candidate, comparison=comparison)) for d in report.admission.dispositions), + ) + analyses = (replace(report.result.analyses[0], comparison=comparison),) + result = runner.admission_push_result(admission, analyses, config=kwargs["config"]) + return replace(report, admission=admission, result=result, scans=tuple(replace(s, comparison=comparison) for s in report.scans)) + + monkeypatch.setattr(runner, "_analyze", analyze) + result = invoke( + tmp_path, + root, + f"refs/heads/a {head} refs/heads/a {base}\nrefs/heads/b {base} refs/heads/b {head}\n", + config=PushConfig(mode=mode, incomplete_coverage_policy=strict), + ) + assert result.exit_code == exit_code + assert result.result.finding_status == ("actionable" if admitted else "none") + data = json.loads((tmp_path / "push.json").read_text()) + if admitted: + assert len(data["admission"]["defects"]) == 1 + assert len(data["admission"]["defects"][0]["comparisons"]) == 2