feat: add authentication and ACL primitives

app/security/acl.py

@@ -0,0 +1,87 @@
+"""Capability-driven ACL evaluation with token privilege separation."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from app.contracts.model import Permissions
+
+
+@dataclass(frozen=True, slots=True)
+class Realm:
+    name: str
+    kind: str
+
+
+@dataclass(frozen=True, slots=True)
+class Principal:
+    name: str
+    realm: str
+
+
+@dataclass(frozen=True, slots=True)
+class Role:
+    name: str
+    privileges: frozenset[str]
+
+
+@dataclass(frozen=True, slots=True)
+class AclEntry:
+    principal: str
+    path: str
+    privileges: frozenset[str]
+    propagate: bool = True
+
+
+def _ancestors(path: str) -> tuple[str, ...]:
+    parts = [part for part in path.split("/") if part]
+    return tuple(["/"] + ["/" + "/".join(parts[:index]) for index in range(1, len(parts) + 1)])
+
+
+def effective_privileges(
+    principal: str, path: str, entries: tuple[AclEntry, ...]
+) -> frozenset[str]:
+    privileges: set[str] = set()
+    for entry in entries:
+        if entry.principal != principal or entry.path not in _ancestors(path):
+            continue
+        if entry.path == path or entry.propagate:
+            privileges.update(entry.privileges)
+    return frozenset(privileges)
+
+
+def authorize(
+    principal: str,
+    path: str,
+    required: frozenset[str],
+    entries: tuple[AclEntry, ...],
+    *,
+    token_privileges: frozenset[str] | None = None,
+) -> bool:
+    privileges = effective_privileges(principal, path, entries)
+    if token_privileges is not None:
+        privileges &= token_privileges
+    return required <= privileges
+
+
+@dataclass(frozen=True, slots=True)
+class CapabilityRequirement:
+    path: str
+    privileges: frozenset[str]
+
+
+def requirement_from_contract(
+    permissions: Permissions | None, parameters: dict[str, str]
+) -> CapabilityRequirement | None:
+    if permissions is None or not permissions.expression:
+        return None
+    check = permissions.expression.get("check")
+    if not isinstance(check, list) or len(check) < 3 or check[0] != "perm":
+        return None
+    raw_path = str(check[1])
+    for name, value in parameters.items():
+        raw_path = raw_path.replace(f"{{{name}}}", value).replace(f"<{name}>", value)
+    raw_privileges = check[2]
+    if not isinstance(raw_privileges, list):
+        return None
+    return CapabilityRequirement(raw_path, frozenset(str(item) for item in raw_privileges))