"""Immutable capability registry (RBAC v2). A capability code is a behavior contract with the grammar:: .. The registry is the source of truth for capability metadata: label, description, risk level, prerequisites, and whether the capability is part of the general-membership baseline. Database Permission rows mirror the registry for role assignment/display; the registry (code) wins on conflicts and rows are validated against it in tests. Module catalogs are added incrementally per migration wave (plan section 20). Procurement is the worked example (plan section 8) and ships first. New resources always get a registry entry — never fold them into a broader code. """ from __future__ import annotations import re from dataclasses import dataclass, field CODE_RE = re.compile(r"^[a-z][a-z0-9_]*\.[a-z][a-z0-9_]*\.[a-z][a-z0-9_]*$") RISK_LOW = "low" RISK_MEDIUM = "medium" RISK_HIGH = "high" RISK_CRITICAL = "critical" RISK_LEVELS = (RISK_LOW, RISK_MEDIUM, RISK_HIGH, RISK_CRITICAL) @dataclass(frozen=True) class CapabilityDefinition: code: str label: str description: str risk: str = RISK_LOW baseline: bool = False prerequisites: tuple[str, ...] = field(default_factory=tuple) @property def module(self) -> str: return self.code.split(".", 2)[0] @property def resource(self) -> str: return self.code.split(".", 2)[1] @property def verb(self) -> str: return self.code.split(".", 2)[2] def _cap(code, label, description, risk=RISK_LOW, baseline=False, prerequisites=()): return CapabilityDefinition( code=code, label=label, description=description, risk=risk, baseline=baseline, prerequisites=tuple(prerequisites), ) # --- Procurement worked-example catalog (plan section 8, source of truth) --- _PROCUREMENT = [ _cap("procurement.requisition.read", "Requisitions: read", "List/read own scoped requisitions and safe status", RISK_LOW, baseline=True), _cap("procurement.requisition.create", "Requisitions: create", "Create a draft for an allowed department/project", RISK_MEDIUM, baseline=True), _cap("procurement.requisition.update", "Requisitions: update", "Edit own draft only", RISK_MEDIUM, baseline=True), _cap("procurement.requisition.submit", "Requisitions: submit", "Submit own draft into approval routing", RISK_MEDIUM, baseline=True), _cap("procurement.approval.read", "Approvals: read", "Read assigned/scoped approval work", RISK_MEDIUM), _cap("procurement.approval.decide", "Approvals: decide", "Approve/reject an assigned step", RISK_HIGH, prerequisites=("procurement.approval.read",)), _cap("procurement.vendor.read", "Vendors: read", "Read organization supplier directory, excluding bank details", RISK_MEDIUM), _cap("procurement.vendor.manage", "Vendors: manage", "Create/update supplier master data", RISK_HIGH, prerequisites=("procurement.vendor.read",)), _cap("procurement.vendor_bank.read", "Vendor bank details: read", "Read bank details", RISK_HIGH, prerequisites=("procurement.vendor.read",)), _cap("procurement.vendor_bank.verify", "Vendor bank details: verify", "Verify/change verification state", RISK_CRITICAL, prerequisites=("procurement.vendor_bank.read", "procurement.vendor.manage")), _cap("procurement.sourcing.read", "RFQ/RFP: read", "Read Sourcing Events and quotes", RISK_MEDIUM), _cap("procurement.sourcing.manage", "RFQ/RFP: manage", "Create/update Sourcing Events and quotes", RISK_HIGH, prerequisites=("procurement.sourcing.read",)), _cap("procurement.sourcing.award", "Sourcing: award", "Select the winning quote", RISK_CRITICAL, prerequisites=("procurement.sourcing.read",)), _cap("procurement.purchase_order.read", "Purchase Orders: read", "List/read POs in scope", RISK_MEDIUM), _cap("procurement.purchase_order.create", "Purchase Orders: create", "Create a draft PO from an approved award", RISK_HIGH, prerequisites=("procurement.purchase_order.read",)), _cap("procurement.purchase_order.update", "Purchase Orders: update", "Edit a draft PO", RISK_HIGH, prerequisites=("procurement.purchase_order.read",)), _cap("procurement.purchase_order.issue", "Purchase Orders: issue", "Issue PO and create budget commitment", RISK_CRITICAL, prerequisites=("procurement.purchase_order.read",)), _cap("procurement.purchase_order.cancel", "Purchase Orders: cancel", "Cancel/reverse an open PO", RISK_CRITICAL, prerequisites=("procurement.purchase_order.read",)), _cap("procurement.contract.read", "Contracts: read", "List/read contracts in scope", RISK_MEDIUM), _cap("procurement.contract.create", "Contracts: create", "Create a draft contract", RISK_HIGH, prerequisites=("procurement.contract.read",)), _cap("procurement.contract.update", "Contracts: update", "Edit a draft contract", RISK_HIGH, prerequisites=("procurement.contract.read",)), _cap("procurement.contract.sign", "Contracts: sign", "Sign and commit contract value", RISK_CRITICAL, prerequisites=("procurement.contract.read",)), _cap("procurement.contract.terminate", "Contracts: terminate", "Terminate and reverse commitment", RISK_CRITICAL, prerequisites=("procurement.contract.read",)), _cap("procurement.acceptance.read", "Goods Receipts: read", "List/read Acceptance records", RISK_MEDIUM), _cap("procurement.acceptance.record", "Goods Receipts: record", "Record pending goods/service acceptance", RISK_HIGH, prerequisites=("procurement.acceptance.read",)), _cap("procurement.acceptance.decide", "Goods Receipts: decide", "Confirm received/partial/rejected", RISK_CRITICAL, prerequisites=("procurement.acceptance.read",)), _cap("procurement.acceptance.generate_assets", "Goods Receipts: generate assets", "Create assets from accepted lines", RISK_HIGH, prerequisites=("procurement.acceptance.read",)), _cap("procurement.invoice.read", "Invoices: read", "List/read supplier invoices", RISK_HIGH), _cap("procurement.invoice.prepare", "Invoices: prepare", "Create/update a pending invoice", RISK_HIGH, prerequisites=("procurement.invoice.read",)), _cap("procurement.invoice.match", "Invoices: match", "Recompute three-way match", RISK_HIGH, prerequisites=("procurement.invoice.read",)), _cap("procurement.invoice.approve", "Invoices: approve", "Approve matched invoice and post accrual", RISK_CRITICAL, prerequisites=("procurement.invoice.read",)), _cap("procurement.payment.read", "Payments: read", "List/read prepared and released payments", RISK_CRITICAL), _cap("procurement.payment.prepare", "Payments: prepare", "Prepare/update a pending payment", RISK_CRITICAL, prerequisites=("procurement.payment.read",)), _cap("procurement.payment.approve_funds", "Payments: approve funds", "Approve project/department funding", RISK_CRITICAL, prerequisites=("procurement.payment.read",)), _cap("procurement.payment.release", "Payments: release", "Release cash and post payment", RISK_CRITICAL, prerequisites=("procurement.payment.read",)), _cap("procurement.report.read", "Reports: read", "Read scoped Procurement metrics", RISK_HIGH), _cap("procurement.report.export", "Reports: export", "Export Procurement report data", RISK_HIGH, prerequisites=("procurement.report.read",)), _cap("procurement.reconciliation.read", "Reconciliation: read", "Read budget commitment/GL drift", RISK_CRITICAL), ] # --- Access administration (plan section 7.15; needed for Roles/Users UI) --- _ACCESS = [ _cap("access.user.read", "Users: read", "Read organization user directory for access admin", RISK_MEDIUM), _cap("access.user.manage", "Users: manage", "Manage organization users", RISK_HIGH, prerequisites=("access.user.read",)), _cap("access.assignment.manage", "Role assignments: manage", "Assign/remove user roles", RISK_HIGH, prerequisites=("access.user.read",)), _cap("access.role.read", "Roles: read", "Read capability roles", RISK_MEDIUM), _cap("access.role.manage", "Roles: manage", "Create/edit capability roles", RISK_CRITICAL, prerequisites=("access.role.read",)), _cap("access.scope.manage", "Scope grants: manage", "Grant/revoke operational scopes", RISK_CRITICAL, prerequisites=("access.role.read",)), _cap("access.audit.read", "Access logs: read", "Read access audit log", RISK_HIGH), _cap("access.audit.export", "Access logs: export", "Export access audit log", RISK_HIGH, prerequisites=("access.audit.read",)), ] _ALL_DEFINITIONS = _PROCUREMENT + _ACCESS def _build_registry(definitions): registry: dict[str, CapabilityDefinition] = {} for definition in definitions: if not CODE_RE.match(definition.code): raise ValueError(f"Capability code violates grammar: {definition.code!r}") if definition.code in registry: raise ValueError(f"Duplicate capability code: {definition.code!r}") if definition.risk not in RISK_LEVELS: raise ValueError(f"Unknown risk level {definition.risk!r} on {definition.code!r}") if not definition.label or not definition.description: raise ValueError(f"Capability {definition.code!r} needs label and description") registry[definition.code] = definition for definition in registry.values(): for prereq in definition.prerequisites: if prereq not in registry: raise ValueError( f"Capability {definition.code!r} requires unknown prerequisite {prereq!r}" ) return registry _REGISTRY: dict[str, CapabilityDefinition] = _build_registry(_ALL_DEFINITIONS) def get_capability(code: str) -> CapabilityDefinition: return _REGISTRY[code] def is_registered(code: str) -> bool: return code in _REGISTRY def all_capabilities() -> tuple[CapabilityDefinition, ...]: return tuple(_REGISTRY.values()) def capabilities_for_module(module: str) -> tuple[CapabilityDefinition, ...]: return tuple(d for d in _REGISTRY.values() if d.module == module) def baseline_capability_codes() -> frozenset[str]: """Capabilities every active organization member holds, scoped to own records.""" return frozenset(d.code for d in _REGISTRY.values() if d.baseline) def validate_role_capability_codes(codes) -> list[str]: """Field-level validation for a role's capability set. Returns a list of human-readable problems (empty when valid): unknown codes, wildcards, and missing prerequisites. Baseline requisition capabilities are granted together via membership, so a role repeating one is allowed (it can be bound to a wider assignment scope). """ problems: list[str] = [] selected = set(codes) for code in sorted(selected): if "*" in code: problems.append(f"Wildcard capabilities are not allowed: {code}") continue if not is_registered(code): problems.append(f"Unknown capability code: {code}") continue for prereq in get_capability(code).prerequisites: if prereq not in selected: problems.append(f"{code} requires {prereq}") return problems