feat(authorization-obligations): group sibling handlers and find consistency anomalies

Task 2.2. Handlers on the same resource under the same module are siblings;
when at least min_siblings of them discharge an obligation and one does not,
that one is the anomaly and the discharging siblings are its evidence (Req 3.2).
A set with fewer members than the minimum is under-populated and never reported
(Req 3.3). A public route among authenticated siblings is a missing path guard
with the mapper's access classification as evidence (Req 3.4).

The sibling key is the module path without its suffix, so two same-named files
in different directories are never merged. Witness validity (an identity
constraint counts only when it binds the authenticated context) is one helper
shared by operations, siblings and the checker.

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

src/openultrasast/ruleset/obligations/javascript.toml

@@ -12,7 +12,7 @@ sensitivity = "high"
 [[operation]]
 id = "orm-write"
 kind = "protected_write"
-calls = ["update", "updateOne", "updateMany", "deleteOne", "deleteMany", "delete", "create", "insert", "upsert", "save", "remove", "findByIdAndUpdate", "findByIdAndDelete"]
+calls = ["updateOne", "updateMany", "deleteOne", "deleteMany", "findByIdAndUpdate", "findByIdAndDelete", "upsert", "insertOne", "insertMany", "prisma.update", "prisma.delete", "prisma.create"]
 requires = ["identity_constraint", "ownership_check"]
 sensitivity = "high"
 
@@ -33,7 +33,7 @@ sensitivity = "high"
 [[operation]]
 id = "security-setting"
 kind = "security_setting"
-calls = ["cors", "cookie", "session", "helmet", "listen", "use", "setHeader", "createClient", "from"]
+calls = ["cors", "cookie", "session", "helmet", "app.listen", "setHeader", "createClient"]
 resource_arg = 0
 requires = ["non_permissive_value"]
 sensitivity = "medium"

src/openultrasast/ruleset/obligations/python.toml

@@ -12,7 +12,7 @@ sensitivity = "high"
 [[operation]]
 id = "orm-write"
 kind = "protected_write"
-calls = ["add", "delete", "update", "save", "commit", "insert_one", "update_one", "delete_one", "bulk_update"]
+calls = ["session.add", "session.delete", "session.merge", "query.update", "query.delete", "objects.update", "objects.create", "objects.filter", "insert_one", "update_one", "delete_one", "bulk_update"]
 requires = ["identity_constraint", "ownership_check"]
 sensitivity = "high"
 
@@ -33,7 +33,7 @@ sensitivity = "high"
 [[operation]]
 id = "security-setting"
 kind = "security_setting"
-calls = ["CORS", "cross_origin", "set_cookie", "Fernet", "run", "config.update", "TrustedHostMiddleware", "add_middleware", "set_debug"]
+calls = ["CORS", "cross_origin", "set_cookie", "app.run", "config.update", "TrustedHostMiddleware", "add_middleware", "set_debug"]
 resource_arg = 0
 requires = ["non_permissive_value"]
 sensitivity = "medium"
@@ -42,7 +42,7 @@ sensitivity = "medium"
 id = "route-guard"
 kind = "path_guard"
 decorators = ["login_required", "jwt_required", "auth_required", "roles_required", "permission_required", "admin_required", "requires_auth", "authenticated", "token_required"]
-calls = ["token_validator", "verify_jwt_in_request", "get_jwt_identity", "check_auth", "require_login", "authenticate", "abort", "login_required", "current_user_required"]
+calls = ["token_validator", "verify_jwt_in_request", "get_jwt_identity", "check_auth", "require_login", "authenticate", "login_required", "current_user_required"]
 
 [[discharger]]
 id = "identity-from-context"
@@ -59,7 +59,7 @@ calls = ["check_owner", "is_owner", "assert_owner", "ensure_owner", "owns", "can
 [[discharger]]
 id = "safe-setting"
 kind = "non_permissive_value"
-calls = ["CORS", "cross_origin", "set_cookie", "run", "config.update", "add_middleware"]
+calls = ["CORS", "cross_origin", "set_cookie", "app.run", "config.update", "add_middleware"]
 permissive_values = ["'*'", "\"*\"", "True", "0.0.0.0", "None", "False", "allow_all", "disable"]
 
 [[discharger]]

src/openultrasast/ruleset/obligations/typescript.toml

@@ -12,7 +12,7 @@ sensitivity = "high"
 [[operation]]
 id = "orm-write"
 kind = "protected_write"
-calls = ["update", "updateOne", "updateMany", "deleteOne", "deleteMany", "delete", "create", "insert", "upsert", "save", "remove", "findByIdAndUpdate", "findByIdAndDelete"]
+calls = ["updateOne", "updateMany", "deleteOne", "deleteMany", "findByIdAndUpdate", "findByIdAndDelete", "upsert", "insertOne", "insertMany", "prisma.update", "prisma.delete", "prisma.create"]
 requires = ["identity_constraint", "ownership_check"]
 sensitivity = "high"
 
@@ -33,7 +33,7 @@ sensitivity = "high"
 [[operation]]
 id = "security-setting"
 kind = "security_setting"
-calls = ["cors", "cookie", "session", "helmet", "listen", "use", "setHeader", "createClient", "from"]
+calls = ["cors", "cookie", "session", "helmet", "app.listen", "setHeader", "createClient"]
 resource_arg = 0
 requires = ["non_permissive_value"]
 sensitivity = "medium"

src/openultrasast/semantic/obligations/__init__.py

@@ -6,6 +6,7 @@ ownership check, a non-permissive value, validated input). Facts are closed data
 findings never claim a proof rung.
 """
 
+from .check import ObligationFinding, ObligationResult, check_obligations, findings_to_static
 from .facts import (
     DISCHARGER_KINDS,
     OPERATION_KINDS,
@@ -25,10 +26,14 @@ __all__ = [
     "DischargerFact",
     "ObligationFacts",
     "ObligationFactsError",
+    "ObligationFinding",
+    "ObligationResult",
     "ObligationShape",
     "OperationFact",
+    "check_obligations",
     "derive_obligation",
     "explain_skip",
+    "findings_to_static",
     "load_obligation_facts",
     "obligation_mechanisms",
 ]

src/openultrasast/semantic/obligations/operations.py

@@ -103,6 +103,11 @@ def find_discharges(
     return tuple(_dedupe(witnesses))
 
 
+def valid_witness(witness: Discharge) -> bool:
+    """An identity constraint witnesses a discharge only when its value comes from the authenticated context."""
+    return witness.kind != "identity_constraint" or witness.provenance == "authenticated_context"
+
+
 def covers(witness: Discharge, operation: Operation) -> bool:
     """A witness discharges an operation when it sits in the same function, has a kind the operation requires, and, for an
     identity constraint, binds its value from the authenticated context. Ordering along the path is the Dominance protocol's
@@ -111,7 +116,7 @@ def covers(witness: Discharge, operation: Operation) -> bool:
         return False
     if witness.kind not in operation.requires:
         return False
-    if witness.kind == "identity_constraint" and witness.provenance != "authenticated_context":
+    if not valid_witness(witness):
         return False
     return not (witness.kind == "non_permissive_value" and witness.line != operation.line)
 
@@ -130,7 +135,7 @@ def _decorator_witnesses(path: str, function: FunctionIR, fact: DischargerFact,
     out: list[Discharge] = []
     for decorator in decorators:
         name = _decorator_name(decorator)
-        if any(candidate == name or candidate in decorator for candidate in fact.decorators):
+        if any(candidate == name for candidate in fact.decorators):  # the parsed decorator name only, never free text
             out.append(
                 Discharge(
                     path=path,
@@ -245,4 +250,4 @@ def _dedupe(witnesses: Sequence[Discharge]) -> list[Discharge]:
     return out
 
 
-__all__ = ["SCOPES", "Discharge", "Operation", "covers", "find_discharges", "find_operations"]
+__all__ = ["SCOPES", "Discharge", "Operation", "covers", "find_discharges", "find_operations", "valid_witness"]

src/openultrasast/semantic/obligations/siblings.py

@@ -0,0 +1,125 @@
+"""Sibling handler sets and consistency anomalies (authorization-obligations, Req 3).
+
+Handlers that operate on the same resource under the same router or module are siblings. When at least the configured
+minimum number of siblings discharge an obligation and one does not (Req 3.2), that one is the anomaly; its evidence is
+the siblings that discharge. A set with fewer members than the minimum is counted as under-populated and never produces
+a finding (Req 3.3). Nothing here touches other findings.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from .facts import DISCHARGER_KINDS
+from .operations import Discharge, Operation, valid_witness
+
+
+@dataclass(frozen=True)
+class SiblingSet:
+    key: tuple[str, str]  # (module path without suffix, resource token)
+    handlers: tuple[str, ...]  # "path::function"
+    discharging: dict[str, tuple[str, ...]]  # discharger kind -> handlers that carry a valid witness of that kind
+    operation_kinds: tuple[str, ...]
+    access_levels: dict[str, str] = field(default_factory=dict)  # handler -> entry-point access classification
+
+
+@dataclass(frozen=True)
+class Anomaly:
+    handler: str
+    missing: str  # discharger kind the siblings practice and this handler lacks
+    operation_kind: str
+    siblings: tuple[str, ...]  # handlers that discharge
+    key: tuple[str, str]
+    access_level: str = "unknown"
+    evidence: tuple[str, ...] = ()
+
+
+def sibling_sets(entries: Sequence[object], operations: Sequence[Operation], discharges: Sequence[Discharge]) -> tuple[SiblingSet, ...]:
+    """Group entry-point handlers by (module, resource) over the operations they perform; record who discharges what."""
+    by_handler_ops: dict[str, list[Operation]] = {}
+    for operation in operations:
+        by_handler_ops.setdefault(f"{operation.path}::{operation.function}", []).append(operation)
+    access: dict[str, str] = {}
+    for entry in entries:
+        name = getattr(entry, "function_name", None)
+        path = getattr(entry, "path", None)
+        if name and path:
+            access[f"{path}::{name}"] = str(getattr(entry, "access_level", "unknown"))
+    valid: dict[str, set[str]] = {}
+    for witness in discharges:
+        if valid_witness(witness):
+            valid.setdefault(f"{witness.path}::{witness.function}", set()).add(witness.kind)
+    groups: dict[tuple[str, str], dict[str, list[Operation]]] = {}
+    for handler, ops in by_handler_ops.items():
+        if handler not in access:
+            continue  # only entry-point handlers are siblings; helpers are reached through them
+        module = _module_of(handler.split("::", 1)[0])
+        for operation in ops:
+            if operation.resource is None:
+                continue
+            groups.setdefault((module, operation.resource), {}).setdefault(handler, []).append(operation)
+    sets: list[SiblingSet] = []
+    for key, members in sorted(groups.items()):
+        handlers = tuple(sorted(members))
+        discharging = {
+            kind: tuple(sorted(handler for handler in handlers if kind in valid.get(handler, set()))) for kind in DISCHARGER_KINDS
+        }
+        kinds = tuple(sorted({operation.kind for ops in members.values() for operation in ops}))
+        sets.append(
+            SiblingSet(
+                key=key,
+                handlers=handlers,
+                discharging={kind: who for kind, who in discharging.items() if who},
+                operation_kinds=kinds,
+                access_levels={handler: access.get(handler, "unknown") for handler in handlers},
+            )
+        )
+    return tuple(sets)
+
+
+def consistency_anomalies(sets: Sequence[SiblingSet], *, min_siblings: int) -> tuple[tuple[Anomaly, ...], tuple[tuple[str, str], ...]]:
+    """Anomalies where at least ``min_siblings`` siblings discharge a kind another sibling lacks; sets with fewer members than
+    ``min_siblings`` are returned as under-populated instead. One threshold, read as Req 3.2 states it."""
+    anomalies: list[Anomaly] = []
+    under: list[tuple[str, str]] = []
+    for group in sets:
+        if len(group.handlers) < min_siblings:
+            under.append(group.key)
+            continue
+        for kind, dischargers in sorted(group.discharging.items()):
+            if len(dischargers) < min_siblings:
+                continue  # one sibling does not set a norm
+            for handler in group.handlers:
+                if handler in dischargers:
+                    continue
+                access = group.access_levels.get(handler, "unknown")
+                evidence: list[str] = [f"{sibling} discharges {kind}" for sibling in dischargers]
+                if kind == "path_guard":
+                    evidence.append(
+                        f"route access: {access}; siblings: "
+                        + ", ".join(sorted({group.access_levels.get(s, "unknown") for s in dischargers}))
+                    )
+                for operation_kind in group.operation_kinds:
+                    anomalies.append(
+                        Anomaly(
+                            handler=handler,
+                            missing=kind,
+                            operation_kind=operation_kind,
+                            siblings=dischargers,
+                            key=group.key,
+                            access_level=access,
+                            evidence=tuple(evidence),
+                        )
+                    )
+    return tuple(anomalies), tuple(under)
+
+
+def _module_of(path: str) -> str:
+    """The module path without its suffix: `api/v1/books.py` -> `api/v1/books`. Two same-named files in different
+    directories are different modules, so their handlers are never siblings."""
+    return Path(path).with_suffix("").as_posix()
+
+
+__all__ = ["Anomaly", "SiblingSet", "consistency_anomalies", "sibling_sets"]

tests/test_obligation_facts.py

@@ -60,9 +60,6 @@ def test_loader_rejects_kinds_and_fields_outside_the_closed_sets(tmp_path: Path)
     (tmp_path / "python.toml").write_text(good + '\n[[discharger]]\nid = "g"\nkind = "maybe_guard"\ndecorators = ["login_required"]\n')
     with pytest.raises(ObligationFactsError, match="maybe_guard"):
         load_obligation_facts(tmp_path)
-    (tmp_path / "python.toml").write_text(good + '\n[[discharger]]\nid = "g"\nkind = "maybe_guard"\ndecorators = ["login_required"]\n')
-    with pytest.raises(ObligationFactsError, match="maybe_guard"):
-        load_obligation_facts(tmp_path)
 
 
 def test_flow_facts_loader_ignores_the_obligation_facts_directory() -> None:

tests/test_obligation_operations.py

@@ -10,9 +10,12 @@ 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"
+    "@app.route('/books/mine/<title>')\n@login_required\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\n\n"
@@ -39,9 +42,10 @@ def _entry(function: str, line: int, end: int, access: str, evidence: list[str])
 
 ENTRIES = [
     _entry("guarded", 6, 7, "authenticated", ["@login_required"]),
-    _entry("constrained", 11, 13, "public", []),
-    _entry("leaky", 17, 19, "public", []),
-    _entry("health", 23, 24, "public", []),
+    _entry("constrained", 12, 14, "authenticated", ["@login_required"]),
+    _entry("constrained_too", 18, 20, "public", []),
+    _entry("leaky", 24, 26, "public", []),
+    _entry("health", 30, 31, "public", []),
 ]
 
 
@@ -54,6 +58,7 @@ def test_find_operations_names_kind_fact_function_and_resource() -> None:
     assert [(op.function, op.kind, op.resource) for op in ops] == [
         ("guarded", "protected_read", "book"),
         ("constrained", "protected_read", "book"),
+        ("constrained_too", "protected_read", "book"),
         ("leaky", "protected_read", "book"),
     ]
     assert all(op.path == "app.py" and op.fact_id == "orm-read" and op.line > 0 for op in ops)
@@ -73,6 +78,7 @@ def test_find_discharges_reports_witness_kind_provenance_and_scope() -> None:
         by_function.setdefault(w.function, []).append((w.kind, w.provenance, w.scope))
     assert ("path_guard", "authenticated_context", "decorator") in by_function["guarded"]
     assert ("identity_constraint", "authenticated_context", "statement") in by_function["constrained"]
+    assert ("identity_constraint", "authenticated_context", "statement") in by_function["constrained_too"]
     assert ("identity_constraint", "request_input", "statement") in by_function["leaky"]  # present, but discharges nothing
     assert "health" not in by_function
     assert all(w.path == "app.py" and w.fact_id for w in witnesses)
@@ -87,6 +93,6 @@ def test_discharge_witness_covers_operation_only_when_kind_and_provenance_fit()
     ops = {op.function: op for op in find_operations(ir, facts, text=APP)}
     witnesses = find_discharges(ir, facts, load_facts().for_language("python"), text=APP, entries=ENTRIES)
     covering = {op.function: [w.kind for w in witnesses if covers(w, op)] for op in ops.values()}
-    assert covering["constrained"] == ["identity_constraint"]
+    assert covering["constrained"] == ["identity_constraint"] and covering["constrained_too"] == ["identity_constraint"]
     assert covering["leaky"] == []  # request-bound identity does not cover
     assert covering["guarded"] == []  # a path guard does not cover a protected read (only privileged actions)

tests/test_obligation_policy.py

@@ -42,6 +42,7 @@ def test_unknown_fields_and_values_are_rejected_by_name(tmp_path: Path) -> None:
         (GOOD.replace('sensitivity = "high"', 'sensitivity = "critical"'), "critical"),
         (GOOD + "\nallow_everything = true\n", "allow_everything"),
         (GOOD.replace('identity_source = "request.user"', 'identity_source = "whatever.works"'), "identity_source"),
+        (GOOD + '\n[[route]]\npath = "/x"\naccess = "public"\ncolor = "blue"\n', "color"),
     ):
         path.write_text(bad)
         with pytest.raises(PolicyError, match=needle):

tests/test_obligation_siblings.py

@@ -0,0 +1,111 @@
+"""authorization-obligations task 2.2: sibling handler sets and consistency anomalies (offline)."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+from test_obligation_operations import APP, ENTRIES  # noqa: E402
+
+from openultrasast.semantic.facts import load_facts  # noqa: E402
+from openultrasast.semantic.ir import parse_file  # noqa: E402
+
+
+def _inputs():
+    from openultrasast.semantic.obligations import load_obligation_facts
+    from openultrasast.semantic.obligations.operations import find_discharges, find_operations
+
+    ir = parse_file("app.py", APP, "python")
+    facts = load_obligation_facts().for_language("python")
+    ops = find_operations(ir, facts, text=APP)
+    wit = find_discharges(ir, facts, load_facts().for_language("python"), text=APP, entries=ENTRIES)
+    return ops, wit
+
+
+def test_sibling_sets_group_by_router_and_resource_and_record_who_discharges() -> None:
+    from openultrasast.semantic.obligations.siblings import sibling_sets
+
+    ops, wit = _inputs()
+    sets = sibling_sets(ENTRIES, ops, wit)
+    assert len(sets) == 1
+    (books,) = sets
+    assert books.key == ("app", "book")
+    assert set(books.handlers) == {"app.py::guarded", "app.py::constrained", "app.py::constrained_too", "app.py::leaky"}
+    assert books.discharging["identity_constraint"] == (
+        "app.py::constrained",
+        "app.py::constrained_too",
+    )  # leaky's request-bound one does not count
+    assert books.discharging["path_guard"] == ("app.py::constrained", "app.py::guarded")
+    assert books.operation_kinds == ("protected_read",)
+
+
+def test_consistency_anomalies_respect_the_minimum_and_name_the_siblings() -> None:
+    from openultrasast.semantic.obligations.siblings import consistency_anomalies, sibling_sets
+
+    ops, wit = _inputs()
+    sets = sibling_sets(ENTRIES, ops, wit)
+    anomalies, under = consistency_anomalies(sets, min_siblings=2)
+    by_handler = {(a.handler, a.missing): a for a in anomalies}
+    # tasks.md observable: two siblings constrain by owner -> one anomaly per unconstrained handler naming both siblings
+    assert by_handler[("app.py::leaky", "identity_constraint")].siblings == ("app.py::constrained", "app.py::constrained_too")
+    assert by_handler[("app.py::guarded", "identity_constraint")].siblings == ("app.py::constrained", "app.py::constrained_too")
+    assert by_handler[("app.py::leaky", "identity_constraint")].operation_kind == "protected_read"
+    assert ("app.py::constrained", "identity_constraint") not in by_handler
+    assert under == ()
+    # Req 3.2: the configured minimum applies to how many siblings discharge, not to one sibling setting the norm
+    anomalies_three, _ = consistency_anomalies(sets, min_siblings=3)
+    assert not [a for a in anomalies_three if a.missing == "identity_constraint"]  # only two discharge
+    anomalies_strict, under_strict = consistency_anomalies(sets, min_siblings=5)
+    assert anomalies_strict == () and under_strict == (("app", "book"),)  # Req 3.3: under-populated, never reported
+
+
+def test_public_route_among_authenticated_siblings_is_a_missing_path_guard_with_access_evidence() -> None:
+    from openultrasast.semantic.obligations.siblings import consistency_anomalies, sibling_sets
+
+    ops, wit = _inputs()
+    sets = sibling_sets(ENTRIES, ops, wit)
+    anomalies, _ = consistency_anomalies(sets, min_siblings=2)
+    guard_anomalies = [a for a in anomalies if a.missing == "path_guard"]
+    assert {a.handler for a in guard_anomalies} == {"app.py::constrained_too", "app.py::leaky"}
+    assert all(a.access_level == "public" and "authenticated" in " ".join(a.evidence) for a in guard_anomalies)
+    assert all(a.siblings == ("app.py::constrained", "app.py::guarded") for a in guard_anomalies)
+
+
+def test_same_named_files_in_different_modules_are_not_siblings() -> None:
+    """Round-2 finding: the key must be the module path, not the file's basename."""
+    from openultrasast.mapping import EntryPointRecord
+    from openultrasast.semantic.obligations.operations import Operation
+    from openultrasast.semantic.obligations.siblings import sibling_sets
+
+    def entry(path: str, function: str) -> EntryPointRecord:
+        return EntryPointRecord(
+            path=path,
+            line=1,
+            end_line=2,
+            function_name=function,
+            name=function,
+            kind="route",
+            access_level="public",
+            trust_boundary="http_request",
+            access_evidence=[],
+            conditions=[],
+            provenance="t",
+            rationale="",
+        )  # type: ignore[arg-type]
+
+    def op(path: str, function: str) -> Operation:
+        return Operation(
+            path=path,
+            line=2,
+            function=function,
+            kind="protected_read",
+            fact_id="orm-read",
+            resource="book",
+            requires=("identity_constraint",),
+        )
+
+    entries = [entry("api/v1/books.py", "list_a"), entry("admin/legacy/books.py", "list_b")]
+    sets = sibling_sets(entries, [op("api/v1/books.py", "list_a"), op("admin/legacy/books.py", "list_b")], [])
+    assert {group.key for group in sets} == {("api/v1/books", "book"), ("admin/legacy/books", "book")}
+    assert all(len(group.handlers) == 1 for group in sets)