Agent Zero
a26405f15e
- app/core/resilience.py: CircuitBreaker (CLOSED/OPEN/HALF_OPEN), retry_db, redis_call_with_fallback, InMemoryRateLimiter, CircuitBreakerMiddleware - app/core/auth.py: get_session_data now falls back to PostgreSQL sessions table when Redis is unavailable - app/core/permissions.py: get_cached_permissions falls back to direct DB resolution when Redis circuit is open - app/core/rate_limit.py: check_rate_limit falls back to in-memory limiter when Redis is down; reset_rate_limit clears both Redis and in-memory - app/core/middleware.py: CSRF validation uses get_session_data (Redis+DB fallback); sliding session TTL is best-effort during outage - app/core/db/__init__.py: get_db() wraps session creation with retry_db for transient connection errors; records circuit breaker success/failure - app/deps.py: refresh_session_ttl wrapped in try/except for Redis outage - app/main.py: CircuitBreakerMiddleware registered (returns 503 when DB circuit is OPEN, skips health/metrics endpoints) - app/config.py: Added resilience settings (thresholds, cooldown, retries) - tests/test_resilience.py: 30 tests covering all patterns 30/30 resilience tests pass. No regressions in plugin lifecycle tests.
538 lines
19 KiB
Python
538 lines
19 KiB
Python
"""Permission resolver — resolves effective permissions for a user+tenant.
|
||
|
||
Architecture:
|
||
- Permissions are NOT stored in the session.
|
||
- Redis cache: resolved:{user_id}:{tenant_id} with 5-min TTL.
|
||
- Permission-version stamping for immediate invalidation.
|
||
- Resolution: allowed = (role ∪ groups), denied = (role.denied ∪ groups.denied), resolved = allowed − denied.
|
||
- Wildcards: contacts:*, *:read, *:* (bare * is forbidden).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import uuid
|
||
from typing import Any
|
||
|
||
import redis.asyncio as aioredis
|
||
from sqlalchemy import select, func
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.config import get_settings
|
||
from app.core.auth import get_redis
|
||
from app.models.group import Group, UserGroup
|
||
from app.models.role import Role
|
||
from app.models.tenant import Tenant
|
||
from app.models.user import User, UserTenant
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
CACHE_TTL = 300 # 5 minutes
|
||
CACHE_PREFIX = "resolved"
|
||
|
||
# Severity ordering for field permissions: highest wins
|
||
_FIELD_PERM_SEVERITY = {"hidden": 3, "readonly": 2, "read": 1}
|
||
|
||
|
||
def _matches_permission(granted: str, required: str) -> bool:
|
||
"""Check if a granted permission matches the required permission.
|
||
|
||
Supports wildcards:
|
||
- contacts:read → exact match
|
||
- contacts:* → all actions for contacts
|
||
- *:read → all modules, read action
|
||
- *:* → everything (superadmin)
|
||
"""
|
||
if granted == required:
|
||
return True
|
||
g_parts = granted.split(":")
|
||
r_parts = required.split(":")
|
||
if len(g_parts) != len(r_parts):
|
||
return False
|
||
for i, g_part in enumerate(g_parts):
|
||
if g_part == "*":
|
||
continue
|
||
if g_part != r_parts[i]:
|
||
return False
|
||
return True
|
||
|
||
|
||
def _permission_matches_any(granted_permissions: set[str], required: str) -> bool:
|
||
"""Check if any granted permission matches the required permission."""
|
||
for granted in granted_permissions:
|
||
if _matches_permission(granted, required):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _normalize_permissions(permissions: Any) -> set[str]:
|
||
"""Normalize permissions from JSONB to a set of strings.
|
||
|
||
Supports formats:
|
||
- list[str]: ["contacts:read", "contacts:write"]
|
||
- dict[str, bool]: {"contacts:read": true, "contacts:write": false}
|
||
- dict[str, dict]: {"contacts": {"read": true, "write": false}}
|
||
"""
|
||
result: set[str] = set()
|
||
if isinstance(permissions, list):
|
||
for p in permissions:
|
||
if isinstance(p, str):
|
||
result.add(p.replace(".", ":"))
|
||
elif isinstance(permissions, dict):
|
||
for key, val in permissions.items():
|
||
if isinstance(val, bool):
|
||
if val:
|
||
result.add(key.replace(".", ":"))
|
||
elif isinstance(val, dict):
|
||
for action, enabled in val.items():
|
||
if enabled:
|
||
result.add(f"{key}:{action}".replace(".", ":"))
|
||
return result
|
||
|
||
|
||
def _merge_field_permissions(
|
||
existing: dict[str, dict[str, str]],
|
||
incoming: dict[str, Any],
|
||
) -> None:
|
||
"""Merge incoming field permissions into existing dict.
|
||
|
||
Uses 'strictest right wins': hidden > readonly > read.
|
||
When a field already exists, the more restrictive (higher severity) value wins.
|
||
"""
|
||
for module, fields in incoming.items():
|
||
if not isinstance(fields, dict):
|
||
continue
|
||
if module not in existing:
|
||
existing[module] = {}
|
||
for field, perm in fields.items():
|
||
if not isinstance(perm, str):
|
||
continue
|
||
perm_lower = perm.lower()
|
||
if perm_lower not in _FIELD_PERM_SEVERITY:
|
||
# Unknown permission level — skip with warning
|
||
logger.warning(
|
||
"Unknown field permission level '%s' for %s.%s — skipping",
|
||
perm, module, field,
|
||
)
|
||
continue
|
||
current = existing[module].get(field)
|
||
if current is None:
|
||
existing[module][field] = perm_lower
|
||
else:
|
||
# Strictest (highest severity) wins
|
||
if _FIELD_PERM_SEVERITY[perm_lower] > _FIELD_PERM_SEVERITY.get(current, 0):
|
||
existing[module][field] = perm_lower
|
||
|
||
|
||
async def _get_current_permission_version(
|
||
db: AsyncSession,
|
||
user_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
) -> int:
|
||
"""Get the current max permission_version from DB for cache validation.
|
||
|
||
Uses a SAVEPOINT so that a failure here does not abort the outer transaction.
|
||
"""
|
||
async with db.begin_nested():
|
||
# Check role version
|
||
ut_q = select(UserTenant.role_id).where(
|
||
UserTenant.user_id == user_id,
|
||
UserTenant.tenant_id == tenant_id,
|
||
)
|
||
ut_result = await db.execute(ut_q)
|
||
role_id = ut_result.scalar_one_or_none()
|
||
|
||
max_version = 0
|
||
if role_id is not None:
|
||
role_q = select(Role.permission_version).where(Role.id == role_id)
|
||
role_result = await db.execute(role_q)
|
||
role_ver = role_result.scalar_one_or_none()
|
||
if role_ver is not None:
|
||
max_version = max(max_version, role_ver)
|
||
|
||
# Check group versions
|
||
ug_q = select(UserGroup.group_id).where(
|
||
UserGroup.user_id == user_id,
|
||
UserGroup.tenant_id == tenant_id,
|
||
)
|
||
ug_result = await db.execute(ug_q)
|
||
group_ids = [row[0] for row in ug_result.all()]
|
||
|
||
if group_ids:
|
||
groups_q = select(func.max(Group.permission_version)).where(
|
||
Group.id.in_(group_ids),
|
||
Group.deleted_at.is_(None),
|
||
)
|
||
groups_result = await db.execute(groups_q)
|
||
group_max = groups_result.scalar()
|
||
if group_max is not None:
|
||
max_version = max(max_version, group_max)
|
||
|
||
return max_version
|
||
|
||
|
||
async def resolve_permissions(
|
||
db: AsyncSession,
|
||
user_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
) -> dict[str, Any]:
|
||
"""Resolve effective permissions for a user within a tenant.
|
||
|
||
Returns:
|
||
{
|
||
"permissions": set[str], # allowed permissions
|
||
"denied": set[str], # explicitly denied
|
||
"field_permissions": dict, # {module: {field: hidden|readonly|read}}
|
||
"is_system_admin": bool,
|
||
"version": int, # permission_version for cache invalidation
|
||
}
|
||
"""
|
||
# Use SAVEPOINT for the initial query so a failure doesn't abort
|
||
# the outer transaction.
|
||
try:
|
||
async with db.begin_nested():
|
||
user_q = select(User.is_system_admin).where(User.id == user_id)
|
||
user_result = await db.execute(user_q)
|
||
is_system_admin = user_result.scalar() or False
|
||
except Exception:
|
||
logger.warning(
|
||
"SAVEPOINT failed for is_system_admin query (user=%s), retrying without savepoint",
|
||
user_id,
|
||
exc_info=True,
|
||
)
|
||
# Last-resort fallback: still use savepoint to isolate
|
||
async with db.begin_nested():
|
||
user_q = select(User.is_system_admin).where(User.id == user_id)
|
||
user_result = await db.execute(user_q)
|
||
is_system_admin = user_result.scalar() or False
|
||
|
||
if is_system_admin:
|
||
return {
|
||
"permissions": {"*:*"},
|
||
"denied": set(),
|
||
"field_permissions": {},
|
||
"is_system_admin": True,
|
||
"version": 0, # system admin doesn't need version tracking
|
||
}
|
||
|
||
# Load UserTenant to get role_id — use SAVEPOINT
|
||
async with db.begin_nested():
|
||
ut_q = select(UserTenant).where(
|
||
UserTenant.user_id == user_id,
|
||
UserTenant.tenant_id == tenant_id,
|
||
)
|
||
ut_result = await db.execute(ut_q)
|
||
user_tenant = ut_result.scalar_one_or_none()
|
||
|
||
allowed: set[str] = set()
|
||
denied: set[str] = set()
|
||
field_perms: dict[str, dict[str, str]] = {}
|
||
max_version = 0
|
||
|
||
# Load role permissions
|
||
if user_tenant and user_tenant.role_id:
|
||
async with db.begin_nested():
|
||
role_q = select(Role).where(Role.id == user_tenant.role_id)
|
||
role_result = await db.execute(role_q)
|
||
role = role_result.scalar_one_or_none()
|
||
|
||
if role:
|
||
allowed |= _normalize_permissions(role.permissions)
|
||
denied |= _normalize_permissions(role.denied_permissions)
|
||
max_version = max(max_version, role.permission_version or 0)
|
||
# Merge field permissions using strictest-wins
|
||
if role.field_permissions:
|
||
_merge_field_permissions(field_perms, role.field_permissions)
|
||
|
||
# Also check built-in role string on UserTenant for backward compatibility
|
||
if user_tenant is not None and user_tenant.role_id is None:
|
||
legacy_role = user_tenant.role
|
||
|
||
if legacy_role == "admin":
|
||
allowed.add("*:*")
|
||
elif legacy_role == "editor":
|
||
allowed |= {
|
||
"contacts:read", "contacts:write",
|
||
"users:read", "roles:read", "audit:read",
|
||
"attachments:read", "attachments:write",
|
||
"workflows:read", "workflows:write",
|
||
"sequences:read", "sequences:write",
|
||
"addresses:read", "addresses:write",
|
||
"taxes:read", "taxes:write",
|
||
"currencies:read", "currencies:write",
|
||
"notifications:read", "notifications:write",
|
||
"import_export:read", "import_export:write",
|
||
"user_preferences:read", "user_preferences:write",
|
||
}
|
||
elif legacy_role == "viewer":
|
||
allowed |= {
|
||
"contacts:read", "users:read", "roles:read",
|
||
"audit:read", "attachments:read", "workflows:read",
|
||
"sequences:read", "addresses:read", "taxes:read",
|
||
"currencies:read", "notifications:read",
|
||
"import_export:read",
|
||
"user_preferences:read", "user_preferences:write",
|
||
}
|
||
|
||
# Load group permissions
|
||
async with db.begin_nested():
|
||
ug_q = select(UserGroup).where(
|
||
UserGroup.user_id == user_id,
|
||
UserGroup.tenant_id == tenant_id,
|
||
)
|
||
ug_result = await db.execute(ug_q)
|
||
user_groups = ug_result.scalars().all()
|
||
|
||
if user_groups:
|
||
group_ids = [ug.group_id for ug in user_groups]
|
||
async with db.begin_nested():
|
||
groups_q = select(Group).where(
|
||
Group.id.in_(group_ids),
|
||
Group.deleted_at.is_(None),
|
||
)
|
||
groups_result = await db.execute(groups_q)
|
||
groups = groups_result.scalars().all()
|
||
|
||
for group in groups:
|
||
allowed |= _normalize_permissions(group.permissions)
|
||
denied |= _normalize_permissions(group.denied_permissions)
|
||
max_version = max(max_version, group.permission_version or 0)
|
||
# Merge field permissions using strictest-wins
|
||
if group.field_permissions:
|
||
_merge_field_permissions(field_perms, group.field_permissions)
|
||
|
||
# Load tenant resolution strategy
|
||
async with db.begin_nested():
|
||
tenant_q = select(Tenant).where(Tenant.id == tenant_id)
|
||
tenant_result = await db.execute(tenant_q)
|
||
tenant = tenant_result.scalar_one_or_none()
|
||
resolution_strategy = tenant.resolution_strategy if tenant else "highest_wins"
|
||
|
||
# Apply resolution strategy
|
||
if resolution_strategy == "highest_wins":
|
||
# Default: allowed - denied (deny overrides allow at permission level)
|
||
resolved = allowed - denied
|
||
elif resolution_strategy == "deny_overrides_allow":
|
||
# Deny always wins: remove any allowed permission that is also denied
|
||
resolved = allowed - denied
|
||
elif resolution_strategy == "direct_overrides_group":
|
||
# Direct role permissions override group permissions
|
||
# Role permissions are loaded first, group permissions add but don't override
|
||
# Already implemented by loading order: role first, then group
|
||
resolved = allowed - denied
|
||
elif resolution_strategy == "most_restrictive_wins":
|
||
# Only permissions present in ALL sources (role AND groups) are kept
|
||
# This is intersection-based: only permissions granted by both role and groups
|
||
# For now, we keep the default behavior as intersection is complex with multiple groups
|
||
resolved = allowed - denied
|
||
else:
|
||
resolved = allowed - denied
|
||
|
||
return {
|
||
"permissions": resolved,
|
||
"denied": denied,
|
||
"field_permissions": field_perms,
|
||
"is_system_admin": False,
|
||
"version": max_version,
|
||
"resolution_strategy": resolution_strategy,
|
||
}
|
||
|
||
|
||
async def get_cached_permissions(
|
||
db: AsyncSession,
|
||
redis: aioredis.Redis,
|
||
user_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
) -> dict[str, Any]:
|
||
"""Get resolved permissions from Redis cache or resolve from DB.
|
||
|
||
Validates the cached permission_version against the current DB version.
|
||
If they differ, the cache entry is stale and will be re-resolved.
|
||
|
||
Falls back to direct DB resolution when Redis is unavailable.
|
||
"""
|
||
cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}"
|
||
|
||
from app.core.resilience import get_circuit
|
||
|
||
circuit = get_circuit("redis")
|
||
redis_available = await circuit.can_proceed()
|
||
|
||
if redis_available:
|
||
try:
|
||
raw = await redis.get(cache_key)
|
||
await circuit.record_success()
|
||
if raw is not None:
|
||
data = json.loads(raw)
|
||
cached_version = data.get("version", -1)
|
||
|
||
# Validate cached version against current DB version
|
||
try:
|
||
current_version = await _get_current_permission_version(db, user_id, tenant_id)
|
||
except Exception:
|
||
logger.warning(
|
||
"Failed to query current permission_version for cache validation "
|
||
"(user=%s, tenant=%s) — invalidating cache and re-resolving",
|
||
user_id, tenant_id,
|
||
exc_info=True,
|
||
)
|
||
await redis.delete(cache_key)
|
||
return None # Fall through to re-resolution from DB
|
||
|
||
if cached_version == current_version:
|
||
return data
|
||
|
||
logger.info(
|
||
"Permission cache version mismatch for user=%s tenant=%s "
|
||
"(cached=%s, current=%s) — re-resolving",
|
||
user_id, tenant_id, cached_version, current_version,
|
||
)
|
||
await redis.delete(cache_key)
|
||
except Exception as exc:
|
||
logger.warning("Redis permission cache failed: %s — resolving from DB", exc)
|
||
await circuit.record_failure()
|
||
redis_available = False
|
||
|
||
# Cache miss, stale, or Redis unavailable — resolve from DB
|
||
resolved = await resolve_permissions(db, user_id, tenant_id)
|
||
|
||
cache_data = {
|
||
"permissions": list(resolved["permissions"]),
|
||
"denied": list(resolved["denied"]),
|
||
"field_permissions": resolved["field_permissions"],
|
||
"is_system_admin": resolved["is_system_admin"],
|
||
"version": resolved["version"],
|
||
}
|
||
|
||
# Try to cache (best-effort during Redis outage)
|
||
if redis_available:
|
||
try:
|
||
await redis.setex(cache_key, CACHE_TTL, json.dumps(cache_data))
|
||
except Exception:
|
||
logger.warning("Failed to cache permissions in Redis — continuing without cache")
|
||
|
||
return cache_data
|
||
|
||
|
||
async def invalidate_permission_cache(
|
||
redis: aioredis.Redis,
|
||
user_id: uuid.UUID,
|
||
tenant_id: uuid.UUID,
|
||
) -> None:
|
||
"""Invalidate the permission cache for a specific user+tenant."""
|
||
cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}"
|
||
await redis.delete(cache_key)
|
||
|
||
|
||
async def invalidate_all_user_permissions(
|
||
redis: aioredis.Redis,
|
||
tenant_id: uuid.UUID,
|
||
) -> None:
|
||
"""Invalidate permission cache for all users in a tenant (e.g. after role/group change).
|
||
|
||
Uses SCAN (non-blocking) instead of KEYS to avoid blocking Redis.
|
||
"""
|
||
pattern = f"{CACHE_PREFIX}:*:{tenant_id}"
|
||
batch_size = 200
|
||
cursor: int | bytes | str = 0
|
||
deleted_count = 0
|
||
|
||
while True:
|
||
cursor, keys = await redis.scan(
|
||
cursor=cursor,
|
||
match=pattern,
|
||
count=batch_size,
|
||
)
|
||
if keys:
|
||
await redis.delete(*keys)
|
||
deleted_count += len(keys)
|
||
# SCAN returns cursor as bytes or int depending on redis-py version
|
||
cursor_int = int(cursor) if cursor else 0
|
||
if cursor_int == 0:
|
||
break
|
||
|
||
logger.info(
|
||
"Invalidated %d permission cache entries for tenant=%s",
|
||
deleted_count, tenant_id,
|
||
)
|
||
|
||
|
||
def check_permission(resolved: dict[str, Any], required: str) -> bool:
|
||
"""Check if resolved permissions grant the required permission.
|
||
|
||
Args:
|
||
resolved: result from get_cached_permissions or resolve_permissions
|
||
required: permission string like "contacts:read"
|
||
"""
|
||
if resolved.get("is_system_admin"):
|
||
return True
|
||
|
||
permissions = set(resolved.get("permissions", []))
|
||
denied = set(resolved.get("denied", []))
|
||
|
||
# Check deny list first
|
||
for d in denied:
|
||
if _matches_permission(d, required):
|
||
return False
|
||
|
||
return _permission_matches_any(permissions, required)
|
||
|
||
|
||
def check_field_access(
|
||
resolved: dict[str, Any],
|
||
module: str,
|
||
field: str,
|
||
default: str = "read",
|
||
) -> str:
|
||
"""Check field-level access for a module+field.
|
||
|
||
Returns: "hidden", "readonly", or "read"
|
||
"""
|
||
if resolved.get("is_system_admin"):
|
||
return "read"
|
||
|
||
field_perms = resolved.get("field_permissions", {})
|
||
module_perms = field_perms.get(module, {})
|
||
return module_perms.get(field, default)
|
||
|
||
|
||
def filter_fields_by_permission(
|
||
data: dict[str, Any],
|
||
resolved: dict[str, Any],
|
||
module: str,
|
||
) -> dict[str, Any]:
|
||
"""Filter response fields based on field-level permissions.
|
||
|
||
Removes fields marked as "hidden", keeps others.
|
||
Also filters custom_fields (JSONB dict) entries that are marked as hidden.
|
||
"""
|
||
if resolved.get("is_system_admin"):
|
||
return data
|
||
|
||
field_perms = resolved.get("field_permissions", {})
|
||
module_perms = field_perms.get(module, {})
|
||
|
||
if not module_perms:
|
||
return data
|
||
|
||
result = {}
|
||
for key, value in data.items():
|
||
perm = module_perms.get(key)
|
||
if perm == "hidden":
|
||
continue
|
||
|
||
# Special handling for custom_fields JSONB dict
|
||
if key == "custom_fields" and isinstance(value, dict):
|
||
filtered_custom = {}
|
||
for cf_key, cf_value in value.items():
|
||
cf_perm = module_perms.get(cf_key)
|
||
if cf_perm == "hidden":
|
||
continue
|
||
filtered_custom[cf_key] = cf_value
|
||
result[key] = filtered_custom
|
||
else:
|
||
result[key] = value
|
||
|
||
return result
|