Kairo-Jinkendo/backend/capabilities.py
Lars 910b1d7e07
Some checks failed
Deploy Development / deploy (push) Successful in 33s
Test Suite / pytest-backend (push) Failing after 4s
Test Suite / k6 /api/health Baseline (push) Has been skipped
Test Suite / playwright-smoke (push) Has been skipped
Test Suite / lint-backend (push) Successful in 1s
Test Suite / compose-smoke (push) Has been skipped
AP0.3: Capability Registry, Entitlements-Snapshot und require_capability.
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-04 22:54:58 +02:00

67 lines
2.1 KiB
Python

"""Central capability resolution and FastAPI enforcement dependencies."""
from __future__ import annotations
import os
from typing import Any, Callable
from fastapi import Depends, HTTPException
from rights_registry import load_grants_for_roles
from services.audit import log_audit
from tenant_context import TenantContext, require_tenant_context
CAPABILITY_ENFORCE_ENV = "CAPABILITY_ENFORCE"
def capability_enforcement_mode() -> str:
raw = os.getenv(CAPABILITY_ENFORCE_ENV, "probe").strip().lower()
return "enforce" if raw in ("1", "true", "enforce", "yes") else "probe"
def resolve_capabilities(*, portal_role: str, tenant_role: str | None) -> frozenset[str]:
return frozenset(load_grants_for_roles(portal_role=portal_role, tenant_role=tenant_role))
def has_capability(ctx: TenantContext, capability_key: str) -> bool:
return capability_key in ctx.capabilities
def check_capability(ctx: TenantContext, capability_key: str) -> dict[str, Any]:
allowed = has_capability(ctx, capability_key)
return {
"capability_key": capability_key,
"allowed": allowed,
"reason": None if allowed else "missing_grant",
}
def require_capability(capability_key: str) -> Callable[..., TenantContext]:
"""FastAPI dependency — blocks in enforce mode, probes in probe mode."""
def _dependency(ctx: TenantContext = Depends(require_tenant_context)) -> TenantContext:
result = check_capability(ctx, capability_key)
if result["allowed"]:
return ctx
log_audit(
"capability.denied",
user_id=ctx.user_id,
tenant_id=ctx.tenant_id,
details={
"capability_key": capability_key,
"mode": capability_enforcement_mode(),
"portal_role": ctx.portal_role,
"tenant_role": ctx.tenant_role,
},
)
if capability_enforcement_mode() == "enforce":
raise HTTPException(
status_code=403,
detail=f"Capability fehlt: {capability_key}",
)
return ctx
return _dependency