feat(authorization-obligations): run the checker in MAP with config and manifest counters

Task 3.1. The obligations proposer runs after variant search in standard and
deep scans, reads the declared policy from the target, and never writes it.
Its findings are appended after every other finding and never enter the sandbox
candidate set. The manifest block carries the sibling sets, the under-populated
count, the operations examined, the findings by label, the policy version, the
stage degradations, and the sibling sets themselves.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

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

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")),
+    )

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/<title>')\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