From 2249de8d2ed551a2e85fa2ff2b71328fcbcd822c Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 6 Sep 2026 07:24:13 +0200 Subject: [PATCH] feat(authorization-obligations): the obligation checker, path-aware and function-local Task 2.5. An obligated operation that is reached and has no dominating discharger is reported at suspicion, labeled by why the obligation exists: a declared policy clause, the practice of sibling handlers, or, in the degraded no-path mode, the operation fact alone. With path records and neither a clause nor a sibling norm nothing justifies the obligation and nothing is reported. A declared public route waives the path guard wherever it is required, not only where siblings practice it (Req 4.4). A malformed store row is recorded as a degradation naming the record instead of being skipped silently. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- .../semantic/obligations/check.py | 308 ++++++++++++++++++ tests/test_obligation_check.py | 221 +++++++++++++ 2 files changed, 529 insertions(+) create mode 100644 src/openultrasast/semantic/obligations/check.py create mode 100644 tests/test_obligation_check.py diff --git a/src/openultrasast/semantic/obligations/check.py b/src/openultrasast/semantic/obligations/check.py new file mode 100644 index 0000000..cd08b6a --- /dev/null +++ b/src/openultrasast/semantic/obligations/check.py @@ -0,0 +1,308 @@ +"""The obligation checker (authorization-obligations, Req 3–6). + +An obligated operation is reported when it is reached (a path record ends at it, or, without path records, its function is +an entry point) and none of the dischargers it accepts dominates it. The label says why the obligation exists: a declared +policy clause (``declared_policy_violation``), the practice of sibling handlers (``consistency_violation``), or, in the +degraded mode without path records, the operation fact alone (``function_local``). With path records and neither a clause +nor a sibling norm, nothing justifies the obligation and nothing is reported (design gating 3). Findings are ``suspicion``; +the obligation, the missing discharger, the evidence and the known fix travel as data. +""" + +from __future__ import annotations + +import re +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field + +from ...findings import StaticFinding +from ..facts import SemanticFacts +from ..ir import FileIR +from .dominance import Dominance +from .facts import ObligationFacts +from .operations import Discharge, Operation, find_discharges, find_operations, valid_witness +from .policy import DeclaredPolicy +from .siblings import Anomaly, SiblingSet, consistency_anomalies, sibling_sets + +LABELS = ("declared_policy_violation", "consistency_violation", "function_local") +_ROUTE_LITERAL = re.compile(r"@\w[\w.]*\.(?:route|get|post|put|delete|patch|api_route|websocket)\(\s*['\"]([^'\"]+)['\"]") + + +@dataclass(frozen=True) +class ObligationFinding: + operation: Operation + missing: str # discharger kind + provenance: str | None # when a discharger of that kind exists but binds the wrong provenance + label: str # LABELS + evidence: tuple[str, ...] # sibling handler ids or the policy clause + known_fix: str | None # mechanism record id whose obligation shape matches + sensitivity: str + route: str | None = None + intent: str | None = None # hunter adjudication (task 3.3): public | protected | unknown + intent_rationale: str | None = None + + +@dataclass(frozen=True) +class ObligationResult: + findings: tuple[ObligationFinding, ...] + discharges: tuple[Discharge, ...] + sibling_sets: tuple[SiblingSet, ...] + under_populated: tuple[tuple[str, str], ...] = () + operations: int = 0 + degradations: tuple[dict[str, object], ...] = field(default=()) + policy_version: str | None = None + + +def check_obligations( + *, + irs: Mapping[str, tuple[FileIR, str]], + entries: Sequence[object], + facts: ObligationFacts, + flow_facts: SemanticFacts, + policy: DeclaredPolicy | None, + paths: Sequence[object], + dominance: Dominance, + store_shapes: Sequence[object], + min_siblings: int, + extra_witnesses: Sequence[Discharge] = (), +) -> ObligationResult: + operations: list[Operation] = [] + witnesses: list[Discharge] = list(extra_witnesses) + texts: dict[str, str] = {} + degradations: list[dict[str, object]] = [] + unsupported: set[str] = set() + for path, (ir, text) in irs.items(): + texts[path] = text + if not ir.parse_ok: + if ir.reason == "language_unsupported" and ir.language not in unsupported: + unsupported.add(ir.language) + degradations.append({"stage": "obligations", "reason": "obligations_language_unsupported", "language": ir.language}) + continue + scoped = facts.for_language(ir.language) + flow = flow_facts.for_language(ir.language) + operations.extend(find_operations(ir, scoped, text=text)) + witnesses.extend(find_discharges(ir, scoped, flow, text=text, entries=entries)) + sets = sibling_sets(entries, operations, witnesses) + anomalies, under = consistency_anomalies(sets, min_siblings=min_siblings) + anomaly_index: dict[tuple[str, str], Anomaly] = {(a.handler, a.missing): a for a in anomalies} + entry_handlers = {f"{getattr(e, 'path', '')}::{getattr(e, 'function_name', '')}" for e in entries} + reached = _reached(operations, paths, entry_handlers) + if not paths and operations: + for language in sorted({irs[op.path][0].language for op in operations if op.path in irs}): + degradations.append( + { + "stage": "obligations", + "reason": "obligations_function_local", + "language": language, + "files": sorted({op.path for op in operations if op.path in irs and irs[op.path][0].language == language}), + "detail": "no path records; obligations evaluated within entry-point handlers only", + } + ) + findings: list[ObligationFinding] = [] + invalid_rows = tuple(sorted(str(getattr(row, "id", "")) for row in store_shapes if not _usable_shape(row))) + if invalid_rows: + degradations.append({"stage": "obligations", "reason": "obligations_store_row_invalid", "records": list(invalid_rows)}) + for operation, path_record in reached: + handler = f"{operation.path}::{operation.function}" + route = _route_path(operation, texts.get(operation.path, "")) + declared_access = policy.route_access(route) if policy is not None and route else None + same = [w for w in witnesses if _relevant(w, operation)] + dominated_kinds = {w.kind for w in same if valid_witness(w) and dominance.dominates(w, operation, path_record)} + # (a) the operation's own obligation: discharged when any accepted discharger dominates. A declared public route + # waives the path guard here too (Req 4.4), so the operation falls through to the next kind it accepts. + required = [kind for kind in operation.requires if not (kind == "path_guard" and declared_access == "public")] + if required and not (set(required) & dominated_kinds): + missing = next( + (kind for kind in required if (handler, kind) in anomaly_index), + required[0], + ) + finding = _finding( + operation, missing, same, anomaly_index.get((handler, missing)), policy, route, declared_access, store_shapes, paths, facts + ) + if finding is not None: + findings.append(finding) + # (b) consistency anomalies on kinds the siblings practice beyond the operation's own requirement (a path guard, say) + for kind in sorted({a.missing for (h, _), a in anomaly_index.items() if h == handler} - set(operation.requires)): + if kind in dominated_kinds: + continue + if kind == "path_guard" and declared_access == "public": + continue # declared public: the guard obligation is waived, the identity one is not + finding = _finding( + operation, kind, same, anomaly_index[(handler, kind)], policy, route, declared_access, store_shapes, paths, facts + ) + if finding is not None: + findings.append(finding) + return ObligationResult( + findings=tuple(_dedupe(findings)), + discharges=tuple(witnesses), + sibling_sets=sets, + under_populated=under, + operations=len(operations), + degradations=tuple(degradations), + policy_version=policy.version_hash if policy is not None else None, + ) + + +def findings_to_static(result: ObligationResult) -> list[StaticFinding]: + statics: list[StaticFinding] = [] + for finding in result.findings: + op = finding.operation + tags = [f"obligation:{op.kind}", f"discharger:{finding.missing}", f"obligation_evidence:{finding.label}"] + if op.resource: + tags.append(f"resource:{op.resource}") + if finding.known_fix: + tags.append(f"mechanism:{finding.known_fix}") + why = "; ".join(finding.evidence) if finding.evidence else "the operation fact alone (function-local mode)" + resource = op.resource or "a resource" + provenance = f" (a {finding.missing} is present but bound from {finding.provenance})" if finding.provenance else "" + statics.append( + StaticFinding( + finding_id=f"obligation:{op.kind}:{op.path}:{op.line}:{finding.missing}", + path=op.path, + title=f"Obligation not discharged: {op.kind} on {resource} without {finding.missing}", + severity={"high": "high", "medium": "medium", "low": "low"}.get(finding.sensitivity, "medium"), + confidence="low", + evidence_level="suspicion", + rationale=( + f"{op.kind} on {resource} in {op.function} is reached without a dominating {finding.missing}{provenance}. " + f"Why the obligation exists: {why}. Label: {finding.label}." + + (f" Known fix learned from the corpus: {finding.known_fix}." if finding.known_fix else "") + ), + line=op.line, + function_name=op.function, + reachability_status="unknown", + reachability_evidence=[], + reachability_conditions=[], + tags=tags, + ranking_priority=0.0, + ) + ) + return statics + + +def _finding( + operation: Operation, + missing: str, + same: Sequence[Discharge], + anomaly: Anomaly | None, + policy: DeclaredPolicy | None, + route: str | None, + declared_access: str | None, + store_shapes: Sequence[object], + paths: Sequence[object], + facts: ObligationFacts, +) -> ObligationFinding | None: + wrong = next((w for w in same if w.kind == missing and not valid_witness(w)), None) + declared = policy.resources.get(operation.resource or "") if policy is not None else None + # A declared clause covers the obligation it speaks about: a protected resource covers identity and ownership + # dischargers; a route declared authenticated or role-restricted covers the path guard. + resource_clause = declared is not None and missing in {"identity_constraint", "ownership_check"} + route_clause = missing == "path_guard" and declared_access in {"authenticated", "role"} + if resource_clause and declared is not None: + label = "declared_policy_violation" + clause = f"resource {declared.name} declared sensitivity {declared.sensitivity}" + if declared.identity_field: + clause += f", identity field {declared.identity_field}" + evidence: tuple[str, ...] = (clause,) + if anomaly is not None: + evidence = evidence + anomaly.siblings + elif route_clause: + label = "declared_policy_violation" + evidence = (f"route {route} declared {declared_access}",) + (anomaly.siblings if anomaly is not None else ()) + elif anomaly is not None: + label = "consistency_violation" + evidence = anomaly.siblings + anomaly.evidence[len(anomaly.siblings) :] + elif paths: + return None # path-aware mode: without a clause or a sibling norm, nothing justifies the obligation (gating 3) + else: + label = "function_local" # degraded mode: the operation fact alone, recorded as a degradation + evidence = () + sensitivity = (declared.sensitivity if declared is not None else None) or _fact_sensitivity(operation, facts) + return ObligationFinding( + operation=operation, + missing=missing, + provenance=wrong.provenance if wrong is not None else None, + label=label, + evidence=evidence, + known_fix=_known_fix(operation, missing, store_shapes), + sensitivity=sensitivity, + route=route, + ) + + +def _reached(operations: Sequence[Operation], paths: Sequence[object], entry_handlers: set[str]) -> list[tuple[Operation, object | None]]: + if not paths: + return [(op, None) for op in operations if f"{op.path}::{op.function}" in entry_handlers] + reached: list[tuple[Operation, object | None]] = [] + for op in operations: + for record in paths: + hops = getattr(record, "hops", None) or () + if not hops: + continue + last = hops[-1][0] if isinstance(hops[-1], (tuple, list)) else str(hops[-1]) + if str(last) == f"{op.path}::{op.function}": + reached.append((op, record)) + break + return reached + + +def _relevant(witness: Discharge, operation: Operation) -> bool: + return witness.scope == "hop" or (witness.path, witness.function) == (operation.path, operation.function) + + +def _usable_shape(record: object) -> bool: + """A store row is usable when its shape dict names both an operation kind and a discharger kind.""" + shape = getattr(record, "shape", None) + return isinstance(shape, dict) and bool(shape.get("operation_kind")) and bool(shape.get("discharger_kind")) + + +def _known_fix(operation: Operation, missing: str, store_shapes: Sequence[object]) -> str | None: + for record in store_shapes: + if not _usable_shape(record): + continue + shape = getattr(record, "shape", None) or {} + if isinstance(shape, dict) and shape.get("operation_kind") == operation.kind and shape.get("discharger_kind") == missing: + return str(getattr(record, "id", "")) or None + return None + + +def _fact_sensitivity(operation: Operation, facts: ObligationFacts) -> str: + for fact in facts.operations: + if fact.id == operation.fact_id: + return fact.sensitivity + return "medium" + + +def _route_path(operation: Operation, text: str) -> str | None: + lines = text.splitlines() + index = _def_index(lines, operation.function) + if index is None: + return None + cursor = index - 1 + while cursor >= 0 and lines[cursor].strip().startswith("@"): + match = _ROUTE_LITERAL.search(lines[cursor].strip()) + if match: + return match.group(1) + cursor -= 1 + return None + + +def _def_index(lines: Sequence[str], function: str) -> int | None: + pattern = re.compile(rf"^\s*(?:async\s+)?(?:def|function)\s+{re.escape(function)}\b") + for index, line in enumerate(lines): + if pattern.match(line): + return index + return None + + +def _dedupe(findings: Sequence[ObligationFinding]) -> list[ObligationFinding]: + seen: set[tuple[str, int, str, str]] = set() + out: list[ObligationFinding] = [] + for finding in findings: + key = (finding.operation.path, finding.operation.line, finding.operation.kind, finding.missing) + if key not in seen: + seen.add(key) + out.append(finding) + return out + + +__all__ = ["LABELS", "ObligationFinding", "ObligationResult", "check_obligations", "findings_to_static"] diff --git a/tests/test_obligation_check.py b/tests/test_obligation_check.py new file mode 100644 index 0000000..3344034 --- /dev/null +++ b/tests/test_obligation_check.py @@ -0,0 +1,221 @@ +"""authorization-obligations task 2.5: the checker, function-local and path-aware (offline).""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent)) +from test_obligation_operations import APP, ENTRIES, _entry # noqa: E402 + +from openultrasast.semantic.facts import load_facts # noqa: E402 +from openultrasast.semantic.ir import parse_file # noqa: E402 + +POLICY = ( + 'version = 1\nidentity_source = "request.user"\n\n[[resource]]\nname = "book"\nsensitivity = "high"\nidentity_field = "user_id"\n\n' + '[[route]]\npath = "/books/any/*"\naccess = "public"\n' +) + + +@dataclass(frozen=True) +class _Path: + hops: tuple[tuple[str, int], ...] + + +def _check( + *, policy_text: str | None = None, paths=(), dominance=None, min_siblings: int = 2, store_shapes=(), tmp_path: Path | None = None +): + from openultrasast.semantic.obligations import load_obligation_facts + from openultrasast.semantic.obligations.check import check_obligations + from openultrasast.semantic.obligations.dominance import OrderDominance + from openultrasast.semantic.obligations.policy import load_declared_policy + + policy = None + if policy_text is not None and tmp_path is not None: + (tmp_path / "obligations.toml").write_text(policy_text) + policy = load_declared_policy(tmp_path / "obligations.toml") + ir = parse_file("app.py", APP, "python") + return check_obligations( + irs={"app.py": (ir, APP)}, + entries=ENTRIES, + facts=load_obligation_facts(), + flow_facts=load_facts(), + policy=policy, + paths=paths, + dominance=dominance or OrderDominance(texts={"app.py": APP}), + store_shapes=store_shapes, + min_siblings=min_siblings, + ) + + +def test_function_local_run_reports_the_consistency_violation_with_siblings_and_a_degradation() -> None: + result = _check() + identity = [f for f in result.findings if f.missing == "identity_constraint"] + assert {f.operation.function for f in identity} == {"leaky", "guarded"} + leaky = next(f for f in identity if f.operation.function == "leaky") + assert leaky.label == "consistency_violation" and leaky.provenance == "request_input" + assert "app.py::constrained" in leaky.evidence and "app.py::constrained_too" in leaky.evidence # names both siblings + assert leaky.operation.resource == "book" and leaky.operation.kind == "protected_read" + assert not [f for f in result.findings if f.operation.function == "constrained" and f.missing == "identity_constraint"] + (degradation,) = [d for d in result.degradations if d["reason"] == "obligations_function_local"] + assert degradation["language"] == "python" and degradation["files"] == ["app.py"] # Req 5.3: names the file + assert result.sibling_sets and result.discharges + + +def test_declared_policy_outranks_consistency_and_public_routes_drop_only_the_guard(tmp_path: Path) -> None: + result = _check(policy_text=POLICY, tmp_path=tmp_path) + leaky = [f for f in result.findings if f.operation.function == "leaky"] + identity = next(f for f in leaky if f.missing == "identity_constraint") + assert identity.label == "declared_policy_violation" and any("resource book" in e for e in identity.evidence) + assert not [f for f in leaky if f.missing == "path_guard"] # /books/any/* is declared public + constrained_guard = [f for f in result.findings if f.operation.function == "constrained_too" and f.missing == "path_guard"] + assert constrained_guard and constrained_guard[0].label == "consistency_violation" # not declared public: the guard anomaly stands + + +def test_path_records_gate_reachability_and_a_dominating_hop_discharges() -> None: + from openultrasast.semantic.obligations.operations import Discharge + + class _HopAware: + def __init__(self) -> None: + self.calls = 0 + + def dominates(self, witness: Discharge, operation, path) -> bool: # type: ignore[no-untyped-def] + self.calls += 1 + return witness.scope == "hop" + + reached = (_Path(hops=(("api.py::gateway", 3), ("app.py::leaky", 25))),) + result = _check(paths=reached) + assert {f.operation.function for f in result.findings} == {"leaky"} # only the operation a path reaches + assert not any(d["reason"] == "obligations_function_local" for d in result.degradations) + dominance = _HopAware() + hop_guard = Discharge( + path="api.py", line=1, function="gateway", kind="identity_constraint", fact_id="x", provenance="authenticated_context", scope="hop" + ) + covered = _check(paths=reached, dominance=dominance, store_shapes=(), min_siblings=2) + assert dominance.calls == 0 # leaky's only witness is request-bound, so there is nothing valid to ask dominance about + from openultrasast.semantic.obligations import load_obligation_facts + from openultrasast.semantic.obligations.check import check_obligations + + ir = parse_file("app.py", APP, "python") + with_hop = check_obligations( + irs={"app.py": (ir, APP)}, + entries=ENTRIES, + facts=load_obligation_facts(), + flow_facts=load_facts(), + policy=None, + paths=reached, + dominance=dominance, + store_shapes=(), + min_siblings=2, + extra_witnesses=(hop_guard,), + ) + assert dominance.calls > 0 # the hop witness is valid, so the checker consulted the injected Dominance implementation + assert not [f for f in with_hop.findings if f.operation.function == "leaky" and f.missing == "identity_constraint"] + assert covered.findings # without the hop witness the finding stands + + +def test_findings_become_suspicion_static_findings_with_known_fix_and_labels(tmp_path: Path) -> None: + from openultrasast.semantic.mechanisms import MechanismStore, append_from_pair + from openultrasast.semantic.obligations.check import findings_to_static + from openultrasast.semantic.obligations.shapes import ObligationShape, obligation_mechanisms + + store = MechanismStore(tmp_path / "m.jsonl") + shape = ObligationShape( + language="python", + operation_kind="protected_read", + discharger_kind="identity_constraint", + provenance="authenticated_context", + resource_class="owned", + mechanism="missing_auth_guard", + ) + record = append_from_pair(store, shape, summary="s", cwe="CWE-639", pair="vampi", provenance="human", tier="seeded") + result = _check(store_shapes=obligation_mechanisms(store.load())) + leaky = next(f for f in result.findings if f.operation.function == "leaky" and f.missing == "identity_constraint") + assert leaky.known_fix == record.id + statics = findings_to_static(result) + ids = {s.finding_id for s in statics} + guarded = next(f for f in result.findings if f.operation.function == "guarded" and f.missing == "identity_constraint") + assert f"obligation:protected_read:app.py:{leaky.operation.line}:identity_constraint" in ids # exact id contract + assert f"obligation:protected_read:app.py:{guarded.operation.line}:identity_constraint" in ids + static = next(s for s in statics if s.line == leaky.operation.line and "discharger:identity_constraint" in s.tags) + assert static.evidence_level == "suspicion" + assert { + "obligation:protected_read", + "discharger:identity_constraint", + "obligation_evidence:consistency_violation", + f"mechanism:{record.id}", + } <= set(static.tags) + assert "app.py::constrained" in static.rationale + + +def test_with_path_records_an_unjustified_obligation_yields_no_finding() -> None: + """Round-2 finding: `function_local` is the degraded mode's label only; with paths and no sibling or clause, nothing.""" + reached = (_Path(hops=(("api.py::gateway", 3), ("app.py::leaky", 25))),) + result = _check(paths=reached, min_siblings=5) # every set under-populated: no consistency evidence, no policy + assert result.findings == () and result.under_populated == (("app", "book"),) + assert not any(d["reason"] == "obligations_function_local" for d in result.degradations) + assert all(f.label != "consistency_violation" or f.evidence for f in _check(paths=reached, min_siblings=2).findings) + + +ADMIN = ( + "from flask import request\n\n\n" + "@app.route('/admin/reset/')\ndef reset(name):\n" + " set_password(name, 'temporary')\n" + " return Book.query.filter_by(book_title=name).first()\n" +) +ADMIN_ENTRIES = [_entry("reset", 4, 6, "public", [])] +ADMIN_POLICY = ( + 'version = 1\nidentity_source = "request.user"\n\n[[resource]]\nname = "book"\nsensitivity = "high"\n\n' + '[[route]]\npath = "/admin/*"\naccess = "public"\n' +) + + +def _check_admin(tmp_path: Path, policy_text: str | None): + from openultrasast.semantic.obligations import load_obligation_facts + from openultrasast.semantic.obligations.check import check_obligations + from openultrasast.semantic.obligations.dominance import OrderDominance + from openultrasast.semantic.obligations.policy import load_declared_policy + + policy = None + if policy_text is not None: + (tmp_path / "obligations.toml").write_text(policy_text) + policy = load_declared_policy(tmp_path / "obligations.toml") + ir = parse_file("app.py", ADMIN, "python") + return check_obligations( + irs={"app.py": (ir, ADMIN)}, + entries=ADMIN_ENTRIES, + facts=load_obligation_facts(), + flow_facts=load_facts(), + policy=policy, + paths=(), + dominance=OrderDominance(texts={"app.py": ADMIN}), + store_shapes=(), + min_siblings=2, + ) + + +def test_a_declared_public_route_never_reports_its_missing_path_guard_but_keeps_the_resource_obligation(tmp_path: Path) -> None: + """Round-3 finding (Req 4.4): the waiver must also cover the operation's own requirement, not only sibling anomalies.""" + without = _check_admin(tmp_path, None) + assert [(f.operation.kind, f.missing) for f in without.findings if f.operation.kind == "privileged_action"] == [ + ("privileged_action", "path_guard") + ] + declared = _check_admin(tmp_path, ADMIN_POLICY) + assert not [f for f in declared.findings if f.missing == "path_guard"] # /admin/* is declared public: no guard obligation + identity = [f for f in declared.findings if f.operation.kind == "protected_read" and f.missing == "identity_constraint"] + assert identity and identity[0].label == "declared_policy_violation" # the declared resource still owes its identity constraint + + +def test_a_malformed_store_row_is_a_degradation_not_a_crash() -> None: + """Design Error Handling: store row with malformed obligation shape -> `obligations_store_row_invalid`.""" + + @dataclass(frozen=True) + class _Row: + id: str + shape: object + + result = _check(store_shapes=(_Row(id="corpus:bad", shape="garbage"), _Row(id="corpus:worse", shape={"family": "obligation"}))) + (degradation,) = [d for d in result.degradations if d["reason"] == "obligations_store_row_invalid"] + assert degradation["records"] == ["corpus:bad", "corpus:worse"] + assert all(f.known_fix is None for f in result.findings)