feat(authorization-obligations): obligated operations and dischargers as closed facts

Task 1.1 (Req 1.1-1.5).

- semantic/obligations/facts.py: closed OPERATION_KINDS, DISCHARGER_KINDS
  and PROVENANCE_KINDS; OperationFact and DischargerFact (calls,
  decorators, identity sources, constraint params, request sources,
  permissive values); load_obligation_facts rejects unknown kinds and
  fields by name; per-language scoping.
- ruleset/obligations/{python,javascript,typescript}.toml beside (never
  inside) the flow facts directory. Protected reads and writes are
  discharged by an identity constraint or an ownership check; a path
  guard discharges privileged actions only.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
This commit is contained in:
Marius Ciepluch
2026-09-05 19:38:04 +02:00
co-authored by Claude Fable 5.1
parent c53eddd69d
commit f44dd2c512
6 changed files with 460 additions and 0 deletions
@@ -0,0 +1,68 @@
# Obligated operations and dischargers for JavaScript web code (authorization-obligations, Req 1). Closed data.
version = "1"
language = "javascript"
[[operation]]
id = "orm-read"
kind = "protected_read"
calls = ["findOne", "findById", "findMany", "findFirst", "findUnique", "findAll", "findByPk", "where", "aggregate"]
requires = ["identity_constraint", "ownership_check"]
sensitivity = "high"
[[operation]]
id = "orm-write"
kind = "protected_write"
calls = ["update", "updateOne", "updateMany", "deleteOne", "deleteMany", "delete", "create", "insert", "upsert", "save", "remove", "findByIdAndUpdate", "findByIdAndDelete"]
requires = ["identity_constraint", "ownership_check"]
sensitivity = "high"
[[operation]]
id = "sql-access"
kind = "protected_read"
calls = ["query", "execute", "raw", "run"]
requires = ["identity_constraint"]
sensitivity = "high"
[[operation]]
id = "privileged-action"
kind = "privileged_action"
calls = ["setRole", "grant", "revoke", "promote", "transfer", "payout", "deleteUser", "resetPassword", "unlink", "rm"]
requires = ["path_guard", "ownership_check"]
sensitivity = "high"
[[operation]]
id = "security-setting"
kind = "security_setting"
calls = ["cors", "cookie", "session", "helmet", "listen", "use", "setHeader", "createClient", "from"]
resource_arg = 0
requires = ["non_permissive_value"]
sensitivity = "medium"
[[discharger]]
id = "route-guard"
kind = "path_guard"
calls = ["authenticate", "requireAuth", "ensureAuthenticated", "isAuthenticated", "verifyToken", "checkAuth", "protect", "authorize", "requireUser", "getServerSession", "auth", "withAuth", "jwt.verify", "passport.authenticate"]
decorators = ["UseGuards", "Authorized", "Roles"]
[[discharger]]
id = "identity-from-context"
kind = "identity_constraint"
identity_sources = ["req.user", "request.user", "ctx.user", "session.user", "locals.user", "auth.uid", "auth().userId", "currentUser", "user.id", "token.sub", "decoded.sub", "session.userId", "getServerSession("]
constraint_params = ["userId", "user_id", "ownerId", "owner_id", "authorId", "author_id", "accountId", "tenantId", "createdBy", "user", "owner"]
request_sources = ["req.body", "req.query", "req.params", "request.body", "request.query", "req.headers", "req.cookies", "searchParams", "formData("]
[[discharger]]
id = "ownership-check"
kind = "ownership_check"
calls = ["isOwner", "checkOwner", "assertOwner", "canAccess", "can", "hasPermission", "assertUser", "ownsResource"]
[[discharger]]
id = "safe-setting"
kind = "non_permissive_value"
calls = ["cors", "cookie", "session", "helmet", "setHeader"]
permissive_values = ["'*'", "\"*\"", "true", "0.0.0.0", "null", "false", "origin: true", "secure: false", "httpOnly: false"]
[[discharger]]
id = "validated-input"
kind = "validated_input"
calls = ["parse", "safeParse", "validate", "validateSync", "sanitize", "escape", "validationResult", "z.object", "assert"]
@@ -0,0 +1,68 @@
# Obligated operations and dischargers for Python web code (authorization-obligations, Req 1). Closed data.
version = "1"
language = "python"
[[operation]]
id = "orm-read"
kind = "protected_read"
calls = ["filter_by", "query.filter", "objects.filter", "query.get", "objects.get", "get_or_404", "first_or_404", "find_one", "objects.all", "query.all"]
requires = ["identity_constraint", "ownership_check"]
sensitivity = "high"
[[operation]]
id = "orm-write"
kind = "protected_write"
calls = ["add", "delete", "update", "save", "commit", "insert_one", "update_one", "delete_one", "bulk_update"]
requires = ["identity_constraint", "ownership_check"]
sensitivity = "high"
[[operation]]
id = "sql-access"
kind = "protected_read"
calls = ["execute", "executemany", "fetchone", "fetchall"]
requires = ["identity_constraint"]
sensitivity = "high"
[[operation]]
id = "privileged-action"
kind = "privileged_action"
calls = ["set_password", "promote", "grant", "revoke", "set_role", "transfer", "payout", "remove", "rmtree", "send_file", "make_admin"]
requires = ["path_guard", "ownership_check"]
sensitivity = "high"
[[operation]]
id = "security-setting"
kind = "security_setting"
calls = ["CORS", "cross_origin", "set_cookie", "Fernet", "run", "config.update", "TrustedHostMiddleware", "add_middleware", "set_debug"]
resource_arg = 0
requires = ["non_permissive_value"]
sensitivity = "medium"
[[discharger]]
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"]
[[discharger]]
id = "identity-from-context"
kind = "identity_constraint"
identity_sources = ["request.user", "current_user", "g.user", "g.current_user", "get_jwt_identity(", "['sub']", "[\"sub\"]", "session['user_id']", "session[\"user_id\"]", "request.state.user", "token.user", "identity"]
constraint_params = ["user", "user_id", "owner", "owner_id", "author", "author_id", "account", "account_id", "tenant", "tenant_id", "created_by", "username"]
request_sources = ["request.get_json(", "request.json", "request.args", "request.form", "request.values", "request.data", "request.headers", "request.cookies", "request.view_args"]
[[discharger]]
id = "ownership-check"
kind = "ownership_check"
calls = ["check_owner", "is_owner", "assert_owner", "ensure_owner", "owns", "can_access", "has_permission", "authorize"]
[[discharger]]
id = "safe-setting"
kind = "non_permissive_value"
calls = ["CORS", "cross_origin", "set_cookie", "run", "config.update", "add_middleware"]
permissive_values = ["'*'", "\"*\"", "True", "0.0.0.0", "None", "False", "allow_all", "disable"]
[[discharger]]
id = "validated-input"
kind = "validated_input"
calls = ["validate", "schema.load", "parse_obj", "model_validate", "clean", "is_valid", "secure_filename", "safe_join"]
@@ -0,0 +1,68 @@
# Obligated operations and dischargers for TypeScript web code (same surface as JavaScript) (authorization-obligations, Req 1). Closed data.
version = "1"
language = "typescript"
[[operation]]
id = "orm-read"
kind = "protected_read"
calls = ["findOne", "findById", "findMany", "findFirst", "findUnique", "findAll", "findByPk", "where", "aggregate"]
requires = ["identity_constraint", "ownership_check"]
sensitivity = "high"
[[operation]]
id = "orm-write"
kind = "protected_write"
calls = ["update", "updateOne", "updateMany", "deleteOne", "deleteMany", "delete", "create", "insert", "upsert", "save", "remove", "findByIdAndUpdate", "findByIdAndDelete"]
requires = ["identity_constraint", "ownership_check"]
sensitivity = "high"
[[operation]]
id = "sql-access"
kind = "protected_read"
calls = ["query", "execute", "raw", "run"]
requires = ["identity_constraint"]
sensitivity = "high"
[[operation]]
id = "privileged-action"
kind = "privileged_action"
calls = ["setRole", "grant", "revoke", "promote", "transfer", "payout", "deleteUser", "resetPassword", "unlink", "rm"]
requires = ["path_guard", "ownership_check"]
sensitivity = "high"
[[operation]]
id = "security-setting"
kind = "security_setting"
calls = ["cors", "cookie", "session", "helmet", "listen", "use", "setHeader", "createClient", "from"]
resource_arg = 0
requires = ["non_permissive_value"]
sensitivity = "medium"
[[discharger]]
id = "route-guard"
kind = "path_guard"
calls = ["authenticate", "requireAuth", "ensureAuthenticated", "isAuthenticated", "verifyToken", "checkAuth", "protect", "authorize", "requireUser", "getServerSession", "auth", "withAuth", "jwt.verify", "passport.authenticate"]
decorators = ["UseGuards", "Authorized", "Roles"]
[[discharger]]
id = "identity-from-context"
kind = "identity_constraint"
identity_sources = ["req.user", "request.user", "ctx.user", "session.user", "locals.user", "auth.uid", "auth().userId", "currentUser", "user.id", "token.sub", "decoded.sub", "session.userId", "getServerSession("]
constraint_params = ["userId", "user_id", "ownerId", "owner_id", "authorId", "author_id", "accountId", "tenantId", "createdBy", "user", "owner"]
request_sources = ["req.body", "req.query", "req.params", "request.body", "request.query", "req.headers", "req.cookies", "searchParams", "formData("]
[[discharger]]
id = "ownership-check"
kind = "ownership_check"
calls = ["isOwner", "checkOwner", "assertOwner", "canAccess", "can", "hasPermission", "assertUser", "ownsResource"]
[[discharger]]
id = "safe-setting"
kind = "non_permissive_value"
calls = ["cors", "cookie", "session", "helmet", "setHeader"]
permissive_values = ["'*'", "\"*\"", "true", "0.0.0.0", "null", "false", "origin: true", "secure: false", "httpOnly: false"]
[[discharger]]
id = "validated-input"
kind = "validated_input"
calls = ["parse", "safeParse", "validate", "validateSync", "sanitize", "escape", "validationResult", "z.object", "assert"]
@@ -0,0 +1,29 @@
"""Absence bugs as obligations (authorization-obligations spec).
An operation carries an obligation (a protected read or write, a privileged action, a security-relevant setting) that
must be discharged on the way to it (a path guard, an identity constraint bound from the authenticated context, an
ownership check, a non-permissive value, validated input). Facts are closed data; shapes are learned from trusted pairs;
findings never claim a proof rung.
"""
from .facts import (
DISCHARGER_KINDS,
OPERATION_KINDS,
PROVENANCE_KINDS,
DischargerFact,
ObligationFacts,
ObligationFactsError,
OperationFact,
load_obligation_facts,
)
__all__ = [
"DISCHARGER_KINDS",
"OPERATION_KINDS",
"PROVENANCE_KINDS",
"DischargerFact",
"ObligationFacts",
"ObligationFactsError",
"OperationFact",
"load_obligation_facts",
]
@@ -0,0 +1,152 @@
"""Closed obligation facts: which operations carry an obligation and what discharges it (Req 1).
Facts live in ``ruleset/obligations/<language>.toml``, beside (never inside) the flow facts directory, so
``semantic.facts.load_facts`` never sees them. Every kind is checked against a closed set; an unknown kind or field is a
data bug and fails loud. Nothing detects an obligation that is not named here.
"""
from __future__ import annotations
import tomllib
from dataclasses import dataclass
from pathlib import Path
OPERATION_KINDS = ("protected_read", "protected_write", "privileged_action", "security_setting")
DISCHARGER_KINDS = ("path_guard", "identity_constraint", "ownership_check", "non_permissive_value", "validated_input")
PROVENANCE_KINDS = ("authenticated_context", "request_input", "constant", "unknown")
SENSITIVITIES = ("low", "medium", "high")
DEFAULT_OBLIGATION_FACTS_DIR = Path(__file__).resolve().parents[2] / "ruleset" / "obligations"
_LANGUAGE_ALIASES = {"c_cpp": "c", "cpp": "c", "python_web": "python", "node": "javascript", "ts": "typescript"}
_OPERATION_FIELDS = frozenset({"id", "kind", "calls", "resource_arg", "requires", "sensitivity"})
_DISCHARGER_FIELDS = frozenset(
{"id", "kind", "calls", "decorators", "identity_sources", "constraint_params", "request_sources", "permissive_values"}
)
class ObligationFactsError(ValueError):
"""Raised when an obligation facts file names a kind or a field outside the closed sets."""
@dataclass(frozen=True)
class OperationFact:
id: str
kind: str # OPERATION_KINDS
language: str
calls: tuple[str, ...] # trailing-name matches, same rule as sink facts
resource_arg: int | None # argument or receiver position naming the resource (model, table, path); None = receiver
requires: tuple[str, ...] # DISCHARGER_KINDS that discharge this operation
sensitivity: str # SENSITIVITIES
@dataclass(frozen=True)
class DischargerFact:
id: str
kind: str # DISCHARGER_KINDS
language: str
calls: tuple[str, ...] = () # guard calls, validators, setters
decorators: tuple[str, ...] = () # decorator names that guard a whole handler
identity_sources: tuple[str, ...] = () # text patterns whose value comes from the authenticated context
constraint_params: tuple[str, ...] = () # keyword or field names that constrain an operation by identity
request_sources: tuple[str, ...] = () # text patterns whose value comes from the request (identity that discharges nothing)
permissive_values: tuple[str, ...] = () # literal values that leave a security setting open (non_permissive_value facts)
@dataclass(frozen=True)
class ObligationFacts:
version: str
operations: tuple[OperationFact, ...]
dischargers: tuple[DischargerFact, ...]
def for_language(self, language: str) -> ObligationFacts:
key = _LANGUAGE_ALIASES.get(language, language)
return ObligationFacts(
version=self.version,
operations=tuple(item for item in self.operations if item.language == key),
dischargers=tuple(item for item in self.dischargers if item.language == key),
)
def load_obligation_facts(directory: Path | None = None) -> ObligationFacts:
root = directory if directory is not None else DEFAULT_OBLIGATION_FACTS_DIR
if not root.is_dir():
raise ObligationFactsError(f"obligation facts directory missing: {root}")
operations: list[OperationFact] = []
dischargers: list[DischargerFact] = []
versions: list[str] = []
for path in sorted(root.glob("*.toml")):
try:
payload = tomllib.loads(path.read_text())
except (OSError, tomllib.TOMLDecodeError) as exc:
raise ObligationFactsError(f"invalid obligation facts file {path}: {exc}") from exc
language = str(payload.get("language") or path.stem)
versions.append(str(payload.get("version", "1")))
for row in _rows(payload.get("operation"), path):
_only_fields(row, _OPERATION_FIELDS, path)
kind = str(row.get("kind", ""))
if kind not in OPERATION_KINDS:
raise ObligationFactsError(f"{path}: operation {row.get('id')!r} has unknown kind {kind!r}")
requires = _strings(row.get("requires"), "requires", path)
for item in requires:
if item not in DISCHARGER_KINDS:
raise ObligationFactsError(f"{path}: operation {row.get('id')!r} requires unknown discharger kind {item!r}")
sensitivity = str(row.get("sensitivity", "medium"))
if sensitivity not in SENSITIVITIES:
raise ObligationFactsError(f"{path}: operation {row.get('id')!r} has unknown sensitivity {sensitivity!r}")
resource_arg = row.get("resource_arg")
if resource_arg is not None and not isinstance(resource_arg, int):
raise ObligationFactsError(f"{path}: operation {row.get('id')!r} resource_arg must be an integer")
operations.append(
OperationFact(
id=str(row["id"]),
kind=kind,
language=language,
calls=_strings(row.get("calls"), "calls", path),
resource_arg=resource_arg,
requires=requires,
sensitivity=sensitivity,
)
)
for row in _rows(payload.get("discharger"), path):
_only_fields(row, _DISCHARGER_FIELDS, path)
kind = str(row.get("kind", ""))
if kind not in DISCHARGER_KINDS:
raise ObligationFactsError(f"{path}: discharger {row.get('id')!r} has unknown kind {kind!r}")
dischargers.append(
DischargerFact(
id=str(row["id"]),
kind=kind,
language=language,
calls=_strings(row.get("calls"), "calls", path),
decorators=_strings(row.get("decorators"), "decorators", path),
identity_sources=_strings(row.get("identity_sources"), "identity_sources", path),
constraint_params=_strings(row.get("constraint_params"), "constraint_params", path),
request_sources=_strings(row.get("request_sources"), "request_sources", path),
permissive_values=_strings(row.get("permissive_values"), "permissive_values", path),
)
)
return ObligationFacts(version=versions[0] if versions else "1", operations=tuple(operations), dischargers=tuple(dischargers))
def _rows(value: object, path: Path) -> 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 ObligationFactsError(f"{path}: fact tables must be arrays of tables")
return [dict(item) for item in value]
def _only_fields(row: dict[str, object], allowed: frozenset[str], path: Path) -> None:
extra = sorted(set(row) - allowed)
if extra:
raise ObligationFactsError(f"{path}: row {row.get('id')!r} has unknown field {extra[0]!r}")
if "id" not in row:
raise ObligationFactsError(f"{path}: every fact row needs an id")
def _strings(value: object, field: str, path: Path) -> tuple[str, ...]:
if value is None:
return ()
if not isinstance(value, list) or not all(isinstance(item, str) and item for item in value):
raise ObligationFactsError(f"{path} has invalid {field} list")
return tuple(str(item) for item in value)
+75
View File
@@ -0,0 +1,75 @@
"""authorization-obligations task 1.1: obligated operations and dischargers are closed facts (offline)."""
from __future__ import annotations
from pathlib import Path
import pytest
def test_default_obligation_facts_load_per_language_with_closed_kinds() -> None:
from openultrasast.semantic.obligations.facts import DISCHARGER_KINDS, OPERATION_KINDS, PROVENANCE_KINDS, load_obligation_facts
assert set(OPERATION_KINDS) == {"protected_read", "protected_write", "privileged_action", "security_setting"}
assert set(DISCHARGER_KINDS) == {"path_guard", "identity_constraint", "ownership_check", "non_permissive_value", "validated_input"}
assert {"authenticated_context", "request_input", "constant", "unknown"} <= set(PROVENANCE_KINDS)
facts = load_obligation_facts()
py = facts.for_language("python")
reads = [op for op in py.operations if op.kind == "protected_read"]
assert any("filter_by" in op.calls or "query" in op.calls for op in reads) # ORM reads
assert any("execute" in op.calls for op in py.operations) # raw SQL access is an obligated operation too
assert all(
op.kind in OPERATION_KINDS and all(r in DISCHARGER_KINDS for r in op.requires) and op.sensitivity in {"low", "medium", "high"}
for op in facts.operations
)
guards = [d for d in py.dischargers if d.kind == "path_guard"]
assert any("login_required" in d.decorators for d in guards) and any(
"token_validator" in d.calls or "jwt_required" in d.decorators for d in guards
)
identity = [d for d in py.dischargers if d.kind == "identity_constraint"]
assert identity and any("user" in d.constraint_params and "owner_id" in d.constraint_params for d in identity)
assert any("request.user" in d.identity_sources or "['sub']" in d.identity_sources for d in identity)
assert facts.for_language("javascript").operations and facts.for_language("typescript").dischargers
assert facts.for_language("cobol").operations == ()
def test_loader_rejects_kinds_and_fields_outside_the_closed_sets(tmp_path: Path) -> None:
from openultrasast.semantic.obligations.facts import ObligationFactsError, load_obligation_facts
good = (
'version = "1"\nlanguage = "python"\n\n[[operation]]\nid = "orm-read"\nkind = "protected_read"\n'
'calls = ["filter_by"]\nrequires = ["identity_constraint"]\nsensitivity = "high"\n'
)
(tmp_path / "python.toml").write_text(good)
facts = load_obligation_facts(tmp_path)
assert [op.id for op in facts.operations] == ["orm-read"] and facts.operations[0].resource_arg is None
(tmp_path / "python.toml").write_text(good.replace('kind = "protected_read"', 'kind = "maybe_guard"'))
with pytest.raises(ObligationFactsError, match="maybe_guard"):
load_obligation_facts(tmp_path)
(tmp_path / "python.toml").write_text(good + 'regex = "filter_.*"\n')
with pytest.raises(ObligationFactsError, match="regex"):
load_obligation_facts(tmp_path)
(tmp_path / "python.toml").write_text(good.replace('requires = ["identity_constraint"]', 'requires = ["vibes"]'))
with pytest.raises(ObligationFactsError, match="vibes"):
load_obligation_facts(tmp_path)
(tmp_path / "python.toml").write_text(
good + '\n[[discharger]]\nid = "g"\nkind = "path_guard"\ndecorators = ["login_required"]\nmood = "strict"\n'
)
with pytest.raises(ObligationFactsError, match="mood"):
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)
(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:
"""Obligation facts live beside the flow facts, never inside their directory: `load_facts()` must not see them."""
from openultrasast.semantic.facts import load_facts
from openultrasast.semantic.obligations.facts import DEFAULT_OBLIGATION_FACTS_DIR
assert DEFAULT_OBLIGATION_FACTS_DIR.is_dir() and DEFAULT_OBLIGATION_FACTS_DIR.name == "obligations"
flow = load_facts()
assert not any(source.language == "obligations" for source in flow.sources)