diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index b3202d7..fabf5e2 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -7,6 +7,7 @@ from collections.abc import Sequence from dataclasses import dataclass, replace from pathlib import Path from time import perf_counter +from typing import Any from . import tool_hunter from .benchmark import ( @@ -31,7 +32,7 @@ from .calibration import ( from .complexity import map as complexity_map from .complexity.ledger import persist_verdicts from .complexity.map import Hotspot -from .config import load_config, load_dotenv +from .config import ObligationsConfig, load_config, load_dotenv from .findings import StaticFinding, quick_scan_findings, write_findings from .fusion import FusionDecision, fuse_findings_dispatch from .gate import FALSE_POSITIVE_CEILING, RECALL_FLOOR @@ -60,7 +61,7 @@ from .policy import assert_rules_resolve, load_policy from .preprocess import FileTarget, preprocess_repository, write_preprocess_artifact from .provenance import fingerprint from .provider.openrouter import OpenRouterEmbeddingClient, OpenRouterError -from .rank import rank_targets, write_rankings +from .rank import rank_obligations, rank_targets, write_rankings from .regress import TRIGGERABLE, CandidateVerdict, run_regression, write_verdicts from .regress.candidate import hotspot_from_finding from .reports import scan_exit_code, write_manifest, write_markdown_report, write_sarif_report @@ -428,6 +429,8 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut variants_payload: dict[str, object] | None = None variant_findings: list[StaticFinding] = [] mechanisms_cited: dict[str, dict[str, object]] = {} + obligations_payload: dict[str, object] | None = None + obligations_cited: dict[str, dict[str, object]] = {} if Stage.MAP in plan.requested: built_map = runtime.run_stage( "map", @@ -473,6 +476,19 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut findings = findings + variant_findings write_findings(findings, findings_path) write_overlay(overlay_records, overlay_path) + if config.obligations.enabled: + obligation_findings, obligations_payload, obligations_cited = _check_obligations( + root=run.target, + targets=targets, + entries=entry_points, + facts=facts_or_error, + settings=config.obligations, + runtime=runtime, + hunter_model=hunter_model, + ) + if obligation_findings: + findings = rank_obligations(findings + obligation_findings) + write_findings(findings, findings_path) plan = record_completed(plan, Stage.MAP) if not hunter_model: runtime.state["degradations"].append(skip_as_degradation(Stage.MAP, "hunter_model_unavailable")) @@ -553,6 +569,11 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut ) write_verdicts(records, verdicts_path) persist_verdicts(run.target / CALIBRATION_DIR / "complexity_ledger.json", records) + # Req 6.4: proof rungs live on verdicts, not on StaticFinding, so obligations re-rank once the sandbox has spoken. + proven_ids = {fid for record in records if record.verdict == TRIGGERABLE for fid in record.inventory_finding_ids} + if proven_ids and any(finding.finding_id.startswith("obligation:") for finding in findings): + findings = rank_obligations(findings, proven_ids=proven_ids) + write_findings(findings, findings_path) _append_proven_mechanisms(store, records, overlay_records, findings, run.target) verdict_records = records wrote_verdicts = True @@ -569,6 +590,8 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut verdicts=verdict_records if wrote_verdicts else None, overlay=overlay_records if wrote_overlay else None, mechanisms=mechanisms_cited or None, + obligations=obligations_cited or None, + obligations_summary=obligations_payload or None, ), ) runtime.run_stage( @@ -579,6 +602,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut sarif_path, overlay=overlay_records if wrote_overlay else None, mechanisms=mechanisms_cited or None, + obligations=obligations_cited or None, ), ) runtime.run_stage( @@ -608,6 +632,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut worth_fixing=worth_fixing_payload, provenance=provenance.to_dict(), variants=variants_payload, + obligations=obligations_payload, ), ) runtime.finish(status="succeeded") @@ -949,6 +974,95 @@ def _search_variants( return findings, records, payload, cited +def _check_obligations( + *, + root: Path, + targets: list[FileTarget], + entries: Sequence[object], + facts: SemanticFacts | FactLoadError, + settings: ObligationsConfig, + runtime: HarnessRuntime, + hunter_model: str | None = None, +) -> tuple[list[StaticFinding], dict[str, object], dict[str, dict[str, object]]]: + """Fourth MAP proposer (authorization-obligations Req 3-6): obligations without a dominating discharger, at suspicion. + + Function-local until path records exist; the declared policy is read from the target; obligation findings never + enter the sandbox candidate set. + """ + from .semantic.ir import parse_file + from .semantic.obligations import load_obligation_facts, obligation_mechanisms + from .semantic.obligations.check import check_obligations, findings_to_static + from .semantic.obligations.dominance import OrderDominance + from .semantic.obligations.policy import PolicyError, load_declared_policy + + payload: dict[str, object] = {"sibling_sets": 0, "under_populated": 0, "operations": 0, "findings_by_label": {}, "policy_version": None} + if isinstance(facts, FactLoadError): + runtime.state["degradations"].append({"stage": "obligations", "reason": "facts_unavailable"}) + return [], payload, {} + irs: dict[str, tuple[Any, str]] = {} + texts: dict[str, str] = {} + for target in targets: + try: + text = (root / target.path).read_text(errors="ignore") + except OSError: + continue + texts[target.path] = text + irs[target.path] = (parse_file(target.path, text, target.language), text) + policy = None + policy_path = root / settings.policy_path + try: + policy = load_declared_policy(policy_path) + except PolicyError as exc: + runtime.state["degradations"].append({"stage": "obligations", "reason": "policy_invalid", "detail": str(exc)[:200]}) + store = MechanismStore(root / CALIBRATION_DIR / "mechanisms.jsonl") + result = runtime.run_stage( + "obligations", + lambda: check_obligations( + irs=irs, + entries=entries, + facts=load_obligation_facts(), + flow_facts=facts, + policy=policy, + paths=(), + dominance=OrderDominance(texts=texts), + store_shapes=obligation_mechanisms(store.load()), + min_siblings=settings.min_siblings, + ), + ) + from .semantic.obligations.intent import adjudicate_intent + + result = adjudicate_intent(result, client=tool_hunter.resolve_hunter_client(), model=hunter_model or "", texts=texts) + for degradation in result.degradations: + runtime.state["degradations"].append(degradation) + by_label: dict[str, int] = {} + for finding in result.findings: + by_label[finding.label] = by_label.get(finding.label, 0) + 1 + payload = { + "sibling_sets": len(result.sibling_sets), + "under_populated": len(result.under_populated), + "operations": result.operations, + "findings_by_label": by_label, + "policy_version": result.policy_version, + "degradations": [dict(item) for item in result.degradations], + "sets": [{"module": group.key[0], "resource": group.key[1], "handlers": len(group.handlers)} for group in result.sibling_sets], + } + statics = findings_to_static(result) + cited: dict[str, dict[str, object]] = {} + for finding, static in zip(result.findings, statics, strict=True): + cited[static.finding_id] = { + "obligation": finding.operation.kind, + "resource": finding.operation.resource, + "missing": finding.missing, + "provenance": finding.provenance, + "label": finding.label, + "evidence": list(finding.evidence), + "known_fix": finding.known_fix, + "intent": finding.intent, + "intent_rationale": finding.intent_rationale, + } + return statics, payload, cited + + def _hotspot_from_variant(finding: StaticFinding) -> Hotspot: return hotspot_from_finding(finding, rationale="variant of a known mechanism (suspicion; sandbox may raise)") diff --git a/src/openultrasast/config.py b/src/openultrasast/config.py index 2a42343..59ca250 100644 --- a/src/openultrasast/config.py +++ b/src/openultrasast/config.py @@ -135,6 +135,14 @@ class VariantsConfig: max_mechanisms: int = 500 +@dataclass(frozen=True) +class ObligationsConfig: + # authorization-obligations: the obligation checker in MAP (standard/deep only). + enabled: bool = True + min_siblings: int = 3 + policy_path: str = ".openultrasast/obligations.toml" + + @dataclass(frozen=True) class ResolvedConfig: models: ModelConfig = ModelConfig() @@ -151,6 +159,7 @@ class ResolvedConfig: complexity: ComplexityConfig = ComplexityConfig() regress: RegressConfig = RegressConfig() variants: VariantsConfig = VariantsConfig() + obligations: ObligationsConfig = ObligationsConfig() runs_dir: str = ".openultrasast/runs" @@ -176,6 +185,7 @@ def load_config(config_path: Path | None = None) -> ResolvedConfig: complexity=_load_complexity(data.get("complexity", {})), regress=_load_regress(data.get("regress", {})), variants=_load_variants(data.get("variants", {})), + obligations=_load_obligations(data.get("obligations", {})), runs_dir=os.environ.get("OPENULTRASAST_RUNS_DIR", ".openultrasast/runs"), ) @@ -345,3 +355,12 @@ def _load_variants(value: object) -> VariantsConfig: enabled=bool(data.get("enabled", True)), max_mechanisms=_int_value(data.get("max_mechanisms"), 500), ) + + +def _load_obligations(value: object) -> ObligationsConfig: + data = _section(value) + return ObligationsConfig( + enabled=bool(data.get("enabled", True)), + min_siblings=_int_value(data.get("min_siblings"), 3), + policy_path=str(data.get("policy_path", ".openultrasast/obligations.toml")), + ) diff --git a/tests/test_obligation_scan.py b/tests/test_obligation_scan.py new file mode 100644 index 0000000..2a9e214 --- /dev/null +++ b/tests/test_obligation_scan.py @@ -0,0 +1,110 @@ +"""authorization-obligations task 3.1: the checker runs in MAP with config, manifest counters and no sandbox candidacy.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openultrasast.cli import main + +APP = ( + "from flask import request\n\n\n" + "@app.route('/books/')\n@login_required\ndef guarded(title):\n" + " return Book.query.filter_by(book_title=title).first()\n\n\n" + "@app.route('/books/mine/<title>')\ndef constrained(title):\n" + " user_id = request.user.id\n" + " return Book.query.filter_by(user_id=user_id, book_title=title).first()\n\n\n" + "@app.route('/books/shared/<title>')\ndef constrained_too(title):\n" + " owner = current_user.id\n" + " return Book.query.filter_by(owner=owner, book_title=title).first()\n\n\n" + "@app.route('/books/any/<title>')\ndef leaky(title):\n" + " owner = request.args.get('owner')\n" + " return Book.query.filter_by(user_id=owner, book_title=title).first()\n" +) + + +def _repo(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text(APP) + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "0") + return repo + + +def _run_dir(repo: Path) -> Path: + return sorted((repo / ".runs").iterdir())[-1] + + +def test_standard_scan_emits_obligation_findings_and_the_manifest_block(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = _repo(tmp_path, monkeypatch) + (repo / "openultrasast.toml").write_text("[obligations]\nmin_siblings = 2\n") + assert main(["scan", str(repo), "--mode", "standard", "--config", str(repo / "openultrasast.toml")]) == 0 + run_dir = _run_dir(repo) + findings = json.loads((run_dir / "findings.json").read_text())["findings"] + obligations = [f for f in findings if f["finding_id"].startswith("obligation:")] + leaky = [f for f in obligations if f["function_name"] == "leaky" and "discharger:identity_constraint" in f["tags"]] + assert leaky and leaky[0]["evidence_level"] == "suspicion" and "obligation_evidence:consistency_violation" in leaky[0]["tags"] + assert "app.py::constrained" in leaky[0]["rationale"] + manifest = json.loads((run_dir / "manifest.json").read_text()) + block = manifest["obligations"] + assert block["sibling_sets"] == 1 and block["operations"] == 4 and block["policy_version"] is None + assert block["findings_by_label"]["consistency_violation"] >= 1 and block["under_populated"] == 0 + assert any(d["reason"] == "obligations_function_local" for d in manifest["degradations"]) + (local,) = [d for d in block["degradations"] if d["reason"] == "obligations_function_local"] # design: the block carries them + assert local["language"] == "python" and local["files"] == ["app.py"] + assert block["sets"] == [{"module": "app", "resource": "book", "handlers": 4}] + report = (run_dir / "report.md").read_text() + assert "Sibling sets evaluated: 1 (under-populated: 0)" in report and "Policy version: none" in report + assert "- `app` / `book`: 4 handlers" in report + + +def test_declared_policy_is_read_from_the_target_and_versioned_in_the_manifest(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = _repo(tmp_path, monkeypatch) + (repo / "openultrasast.toml").write_text("[obligations]\nmin_siblings = 2\n") + (repo / ".openultrasast").mkdir() + (repo / ".openultrasast" / "obligations.toml").write_text( + 'version = 1\n[[resource]]\nname = "book"\nsensitivity = "high"\nidentity_field = "user_id"\n\n' + '[[route]]\npath = "/books/any/*"\naccess = "public"\n' + ) + assert main(["scan", str(repo), "--mode", "standard", "--config", str(repo / "openultrasast.toml")]) == 0 + run_dir = _run_dir(repo) + manifest = json.loads((run_dir / "manifest.json").read_text()) + assert len(manifest["obligations"]["policy_version"]) == 40 + findings = json.loads((run_dir / "findings.json").read_text())["findings"] + leaky = [f for f in findings if f["finding_id"].startswith("obligation:") and f["function_name"] == "leaky"] + assert any("obligation_evidence:declared_policy_violation" in f["tags"] for f in leaky) + assert not any("discharger:path_guard" in f["tags"] for f in leaky) # declared public: the guard obligation is waived + + +def test_quick_scan_and_disabled_config_skip_obligations(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = _repo(tmp_path, monkeypatch) + assert main(["scan", str(repo), "--mode", "quick"]) == 0 + manifest = json.loads((_run_dir(repo) / "manifest.json").read_text()) + assert "obligations" not in manifest + (repo / "openultrasast.toml").write_text("[obligations]\nenabled = false\n") + assert main(["scan", str(repo), "--mode", "standard", "--config", str(repo / "openultrasast.toml")]) == 0 + manifest = json.loads((_run_dir(repo) / "manifest.json").read_text()) + assert "obligations" not in manifest + findings = json.loads((_run_dir(repo) / "findings.json").read_text())["findings"] + assert not any(f["finding_id"].startswith("obligation:") for f in findings) + + +def test_deep_scan_never_offers_obligation_findings_to_the_sandbox(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast import cli + from openultrasast.sandbox import FakeSandboxRunner, SandboxResult + + repo = _repo(tmp_path, monkeypatch) + (repo / "openultrasast.toml").write_text("[obligations]\nmin_siblings = 2\n") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1") + fake = FakeSandboxRunner(SandboxResult(exit_code=0, stdout="", stderr="", timed_out=False)) + monkeypatch.setattr(cli, "resolve_sandbox_runner", lambda: fake) + assert main(["scan", str(repo), "--mode", "deep", "--config", str(repo / "openultrasast.toml")]) == 0 + run_dir = _run_dir(repo) + verdicts = json.loads((run_dir / "verdicts.json").read_text())["verdicts"] + offered = {finding_id for item in verdicts for finding_id in item["inventory_finding_ids"]} + assert not any(finding_id.startswith("obligation:") for finding_id in offered) + findings = json.loads((run_dir / "findings.json").read_text())["findings"] + assert any(f["finding_id"].startswith("obligation:") for f in findings) # still reported