From aaee724ebda6822d761ed7404e4be0dcdb55e70e Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sat, 5 Sep 2026 19:38:04 +0200 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- .../semantic/obligations/policy.py | 136 ++++++++++++++++++ tests/test_obligation_policy.py | 48 +++++++ 2 files changed, 184 insertions(+) create mode 100644 src/openultrasast/semantic/obligations/policy.py create mode 100644 tests/test_obligation_policy.py diff --git a/src/openultrasast/semantic/obligations/policy.py b/src/openultrasast/semantic/obligations/policy.py new file mode 100644 index 0000000..2a040fa --- /dev/null +++ b/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", +] diff --git a/tests/test_obligation_policy.py b/tests/test_obligation_policy.py new file mode 100644 index 0000000..bc3abb0 --- /dev/null +++ b/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)