Agent Zero
7c8f2a2222
- Integrate ABAC policies into apply_visibility_filter() (allow/deny with priority) - Add field whitelist (ABAC_ALLOWED_FIELDS) for build_sql_condition() security - Add request-level ContextVar for user principals (group_ids, role_id) - Set principals in deps.py (session + bearer auth) - Use ContextVar in visibility.py and permission_resolver.py (N+1 fix) - Add version validation to get_cached_visible_ids() (cache strategy unification) - Deactivate delegation route (parked — not integrated into resolve_permissions) - Add 7 ABAC integration tests All 70 tests pass (7 ABAC + 63 existing). No regressions.
43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
"""Request-level caching for user principals (group_ids + role_id).
|
|
|
|
Avoids N+1 queries by loading principals once per request in deps.py
|
|
and reusing them in visibility.py and permission_resolver.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from contextvars import ContextVar
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass
|
|
class UserPrincipals:
|
|
"""Cached user principals for the current request."""
|
|
user_id: uuid.UUID
|
|
tenant_id: uuid.UUID
|
|
group_ids: list[uuid.UUID]
|
|
role_id: uuid.UUID | None
|
|
|
|
|
|
# ContextVar — async-safe, isolated per request
|
|
_principals_ctx: ContextVar[UserPrincipals | None] = ContextVar(
|
|
"user_principals", default=None
|
|
)
|
|
|
|
|
|
def set_principals(principals: UserPrincipals) -> None:
|
|
"""Set principals for the current request context."""
|
|
_principals_ctx.set(principals)
|
|
|
|
|
|
def get_principals() -> UserPrincipals | None:
|
|
"""Get principals for the current request context, or None if not set."""
|
|
return _principals_ctx.get()
|
|
|
|
|
|
def clear_principals() -> None:
|
|
"""Clear principals at end of request (optional — ContextVar is
|
|
isolated per task, but explicit cleanup is good practice)."""
|
|
_principals_ctx.set(None)
|