feat(authorization-obligations): declared policy with a closed schema

Task 2.3 (Req 4.1, 4.2). load_declared_policy: closed top-level,
resource and route fields; access public|authenticated|role; identity
source restricted to the facts' authenticated-context sources; comment-
stripped version hash; trailing-* route globs; unknown fields and values
rejected by name; the tool never writes the file.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

src/openultrasast/semantic/obligations/policy.py

@@ -0,0 +1,136 @@
+"""The declared policy: project-owned, versioned, closed (authorization-obligations, Req 4).
+
+The tool never writes this file. Unknown fields or values are rejected by name; a missing file means no declared policy.
+"""
+
+from __future__ import annotations
+
+import fnmatch
+import hashlib
+import tomllib
+from dataclasses import dataclass
+from pathlib import Path
+
+from .facts import SENSITIVITIES, load_obligation_facts
+
+ROUTE_ACCESS = ("public", "authenticated", "role")
+_TOP_FIELDS = frozenset({"version", "identity_source", "resource", "route"})
+_RESOURCE_FIELDS = frozenset({"name", "sensitivity", "identity_field"})
+_ROUTE_FIELDS = frozenset({"path", "access", "roles"})
+DEFAULT_POLICY_RELPATH = Path(".openultrasast") / "obligations.toml"
+
+
+class PolicyError(ValueError):
+    """Raised when a declared policy names a field or value outside the closed schema."""
+
+
+@dataclass(frozen=True)
+class PolicyResource:
+    name: str
+    sensitivity: str
+    identity_field: str | None
+
+
+@dataclass(frozen=True)
+class PolicyRoute:
+    path: str  # exact or with a trailing `*`
+    access: str  # ROUTE_ACCESS
+    roles: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class DeclaredPolicy:
+    version: int
+    identity_source: str | None
+    resources: dict[str, PolicyResource]
+    routes: tuple[PolicyRoute, ...]
+    version_hash: str
+
+    def route_access(self, path: str) -> str | None:
+        for route in self.routes:
+            if fnmatch.fnmatchcase(path, route.path) or path == route.path:
+                return route.access
+        return None
+
+    def sensitivity(self, resource: str | None) -> str | None:
+        return self.resources[resource].sensitivity if resource in self.resources else None
+
+
+def load_declared_policy(path: Path) -> DeclaredPolicy | None:
+    if not path.is_file():
+        return None
+    text = path.read_text()
+    try:
+        payload = tomllib.loads(text)
+    except tomllib.TOMLDecodeError as exc:
+        raise PolicyError(f"{path}: invalid TOML: {exc}") from exc
+    extra = sorted(set(payload) - _TOP_FIELDS)
+    if extra:
+        raise PolicyError(f"{path}: unknown field {extra[0]!r}")
+    version = payload.get("version", 1)
+    if not isinstance(version, int):
+        raise PolicyError(f"{path}: version must be an integer")
+    identity_source = payload.get("identity_source")
+    if identity_source is not None:
+        allowed = {source for fact in load_obligation_facts().dischargers for source in fact.identity_sources} | {"token"}
+        if not isinstance(identity_source, str) or identity_source not in allowed:
+            raise PolicyError(f"{path}: identity_source {identity_source!r} is not a known authenticated-context source")
+    resources: dict[str, PolicyResource] = {}
+    for row in _rows(payload.get("resource"), path, "resource"):
+        _only(row, _RESOURCE_FIELDS, path, "resource")
+        sensitivity = str(row.get("sensitivity", "medium"))
+        if sensitivity not in SENSITIVITIES:
+            raise PolicyError(f"{path}: resource {row.get('name')!r} has unknown sensitivity {sensitivity!r}")
+        name = str(row["name"])
+        identity_field = row.get("identity_field")
+        resources[name] = PolicyResource(name=name, sensitivity=sensitivity, identity_field=str(identity_field) if identity_field else None)
+    routes: list[PolicyRoute] = []
+    for row in _rows(payload.get("route"), path, "route"):
+        _only(row, _ROUTE_FIELDS, path, "route")
+        access = str(row.get("access", ""))
+        if access not in ROUTE_ACCESS:
+            raise PolicyError(f"{path}: route {row.get('path')!r} has unknown access {access!r}")
+        roles = row.get("roles", [])
+        if not isinstance(roles, list) or not all(isinstance(item, str) for item in roles):
+            raise PolicyError(f"{path}: route {row.get('path')!r} roles must be a list of strings")
+        routes.append(PolicyRoute(path=str(row["path"]), access=access, roles=tuple(roles)))
+    return DeclaredPolicy(
+        version=version,
+        identity_source=identity_source,
+        resources=resources,
+        routes=tuple(routes),
+        version_hash=_version_hash(text),
+    )
+
+
+def _version_hash(text: str) -> str:
+    normalized = "\n".join(line.split("#", 1)[0].rstrip() for line in text.splitlines() if line.split("#", 1)[0].strip())
+    return hashlib.sha1(normalized.encode()).hexdigest()
+
+
+def _rows(value: object, path: Path, table: str) -> list[dict[str, object]]:
+    if value is None:
+        return []
+    if not isinstance(value, list) or not all(isinstance(item, dict) for item in value):
+        raise PolicyError(f"{path}: {table} must be an array of tables")
+    return [dict(item) for item in value]
+
+
+def _only(row: dict[str, object], allowed: frozenset[str], path: Path, table: str) -> None:
+    extra = sorted(set(row) - allowed)
+    if extra:
+        raise PolicyError(f"{path}: {table} {row.get('name', row.get('path'))!r} has unknown field {extra[0]!r}")
+    required = "name" if table == "resource" else "path"
+    if required not in row:
+        raise PolicyError(f"{path}: every {table} needs a {required}")
+
+
+__all__ = [
+    "DEFAULT_POLICY_RELPATH",
+    "ROUTE_ACCESS",
+    "DeclaredPolicy",
+    "PolicyError",
+    "PolicyResource",
+    "PolicyRoute",
+    "load_declared_policy",
+]

tests/test_obligation_policy.py

@@ -0,0 +1,48 @@
+"""authorization-obligations task 2.3: the declared policy has a closed schema (offline)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+GOOD = (
+    'version = 1\nidentity_source = "request.user"\n\n'
+    '[[resource]]\nname = "book"\nsensitivity = "high"\nidentity_field = "user_id"\n\n'
+    '[[route]]\npath = "/health"\naccess = "public"\n\n[[route]]\npath = "/admin/*"\naccess = "role"\nroles = ["admin"]\n'
+)
+
+
+def test_valid_policy_loads_with_a_version_hash(tmp_path: Path) -> None:
+    from openultrasast.semantic.obligations.policy import load_declared_policy
+
+    path = tmp_path / "obligations.toml"
+    path.write_text(GOOD)
+    policy = load_declared_policy(path)
+    assert policy is not None and policy.version == 1 and policy.identity_source == "request.user"
+    assert policy.resources["book"].sensitivity == "high" and policy.resources["book"].identity_field == "user_id"
+    assert (
+        policy.route_access("/health") == "public"
+        and policy.route_access("/admin/users") == "role"
+        and policy.route_access("/books/1") is None
+    )
+    assert len(policy.version_hash) == 40 and policy.version_hash == load_declared_policy(path).version_hash  # type: ignore[union-attr]
+    path.write_text(GOOD + "\n# comment\n")
+    assert load_declared_policy(path).version_hash == policy.version_hash  # type: ignore[union-attr]  # normalized: comments do not change the version
+    assert load_declared_policy(tmp_path / "missing.toml") is None
+
+
+def test_unknown_fields_and_values_are_rejected_by_name(tmp_path: Path) -> None:
+    from openultrasast.semantic.obligations.policy import PolicyError, load_declared_policy
+
+    path = tmp_path / "obligations.toml"
+    for bad, needle in (
+        (GOOD.replace('access = "public"', 'access = "sometimes"'), "sometimes"),
+        (GOOD + '\n[[resource]]\nname = "x"\nsensitivity = "high"\nmood = "strict"\n', "mood"),
+        (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"),
+    ):
+        path.write_text(bad)
+        with pytest.raises(PolicyError, match=needle):
+            load_declared_policy(path)