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.
465 lines
16 KiB
Python
465 lines
16 KiB
Python
"""FastAPI dependencies: auth, db, tenant context, RBAC."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from typing import Any
|
|
|
|
import redis.asyncio as aioredis
|
|
from fastapi import Depends, HTTPException, Request, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import get_settings
|
|
from app.core.auth import get_redis, get_session_data, refresh_session_ttl
|
|
from app.core.db import get_db, set_tenant_context, set_user_context
|
|
from app.models.guest_user import GuestUser
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Known write-permission modules — used by require_write() to check
|
|
# specific permissions instead of broad wildcards like *:write
|
|
_WRITE_PERMISSIONS = [
|
|
"contacts:write",
|
|
"contacts:create",
|
|
"users:write",
|
|
"roles:write",
|
|
"audit:write",
|
|
"attachments:write",
|
|
"workflows:write",
|
|
"sequences:write",
|
|
"addresses:write",
|
|
"taxes:write",
|
|
"currencies:write",
|
|
"notifications:write",
|
|
"import_export:write",
|
|
"user_preferences:write",
|
|
]
|
|
|
|
|
|
async def get_redis_dep() -> aioredis.Redis:
|
|
"""FastAPI dependency for Redis client."""
|
|
return get_redis()
|
|
|
|
|
|
async def get_current_guest(
|
|
request: Request,
|
|
redis: aioredis.Redis = Depends(get_redis_dep),
|
|
) -> dict[str, Any]:
|
|
"""Get the current guest user from guest session cookie.
|
|
|
|
Returns session data dict with guest_user_id, tenant_id, email, name.
|
|
Used for guest-specific endpoints (guest login, guest contacts).
|
|
"""
|
|
session_id = request.cookies.get("guest_session")
|
|
if not session_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"detail": "Not authenticated", "code": "not_authenticated"},
|
|
)
|
|
|
|
import json
|
|
|
|
raw = await redis.get(f"guest_session:{session_id}")
|
|
if raw is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"detail": "Session expired or invalid", "code": "session_invalid"},
|
|
)
|
|
|
|
session_data = json.loads(raw)
|
|
|
|
# Extend TTL on each request (sliding session)
|
|
await redis.expire(f"guest_session:{session_id}", 1800)
|
|
|
|
return session_data
|
|
|
|
|
|
async def get_current_user(
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
redis: aioredis.Redis = Depends(get_redis_dep),
|
|
) -> dict[str, Any]:
|
|
"""Get the current authenticated user from session cookie.
|
|
|
|
Returns session data dict with user_id, tenant_id, email, name, role,
|
|
and resolved permissions from Redis cache.
|
|
"""
|
|
settings = get_settings()
|
|
|
|
session_id = request.cookies.get(settings.session_cookie_name)
|
|
if not session_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"detail": "Not authenticated", "code": "not_authenticated"},
|
|
)
|
|
|
|
session_data = await get_session_data(redis, session_id)
|
|
if session_data is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"detail": "Session expired or invalid", "code": "session_invalid"},
|
|
)
|
|
|
|
# Sliding session: extend TTL on each authenticated request
|
|
# Best-effort during Redis outage — session still valid from DB fallback
|
|
try:
|
|
await refresh_session_ttl(redis, session_id)
|
|
except Exception:
|
|
logger.debug("refresh_session_ttl failed (Redis may be down) — continuing")
|
|
|
|
if not session_data.get("is_active", True):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"detail": "User account deactivated", "code": "user_inactive"},
|
|
)
|
|
|
|
# Set RLS tenant context
|
|
tenant_id = uuid.UUID(session_data["tenant_id"])
|
|
await set_tenant_context(db, tenant_id)
|
|
|
|
# Set RLS user context for row-level security
|
|
user_id = uuid.UUID(session_data["user_id"])
|
|
from app.models.group import UserGroup
|
|
groups_q = await db.execute(
|
|
select(UserGroup.group_id)
|
|
.where(UserGroup.user_id == user_id)
|
|
.where(UserGroup.tenant_id == tenant_id)
|
|
)
|
|
group_ids = [row[0] for row in groups_q]
|
|
is_admin = session_data.get("is_system_admin", False)
|
|
await set_user_context(db, user_id, group_ids, is_admin)
|
|
|
|
# Check membership status (P1.7: suspended membership should not be usable)
|
|
from app.models.user import UserTenant
|
|
membership_q = await db.execute(
|
|
select(UserTenant.status)
|
|
.where(UserTenant.user_id == user_id)
|
|
.where(UserTenant.tenant_id == tenant_id)
|
|
)
|
|
membership_status = membership_q.scalar_one_or_none()
|
|
if membership_status is not None and membership_status != "active":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail={"detail": f"Mitgliedschaft ist {membership_status}, Zugriff verweigert", "code": "membership_suspended"},
|
|
)
|
|
|
|
# Load resolved permissions from cache (or DB on miss)
|
|
from app.core.permissions import get_cached_permissions
|
|
|
|
user_id = uuid.UUID(session_data["user_id"])
|
|
resolved = await get_cached_permissions(db, redis, user_id, tenant_id)
|
|
session_data["permissions"] = resolved.get("permissions", [])
|
|
session_data["denied_permissions"] = resolved.get("denied", [])
|
|
session_data["field_permissions"] = resolved.get("field_permissions", {})
|
|
session_data["is_system_admin"] = resolved.get("is_system_admin", False)
|
|
|
|
return session_data
|
|
|
|
|
|
async def get_current_user_bearer(
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> dict[str, Any]:
|
|
"""Get the current user from a Bearer API token.
|
|
|
|
Alternative to session-based auth for programmatic access (MCP, API clients).
|
|
Returns the same dict shape as get_current_user.
|
|
"""
|
|
auth_header = request.headers.get("Authorization", "")
|
|
if not auth_header.startswith("Bearer "):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"detail": "Bearer token required", "code": "not_authenticated"},
|
|
)
|
|
|
|
token = auth_header[7:] # Strip "Bearer "
|
|
from app.core.api_token import verify_api_token
|
|
|
|
user_data = await verify_api_token(db, token)
|
|
if user_data is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"detail": "Invalid or expired token", "code": "token_invalid"},
|
|
)
|
|
|
|
# Set RLS tenant context
|
|
tenant_id = uuid.UUID(user_data["tenant_id"])
|
|
await set_tenant_context(db, tenant_id)
|
|
|
|
# Set RLS user context
|
|
user_id = uuid.UUID(user_data["user_id"])
|
|
from app.models.group import UserGroup
|
|
groups_q = await db.execute(
|
|
select(UserGroup.group_id)
|
|
.where(UserGroup.user_id == user_id)
|
|
.where(UserGroup.tenant_id == tenant_id)
|
|
)
|
|
group_ids = [row[0] for row in groups_q]
|
|
is_admin = user_data.get("is_system_admin", False)
|
|
await set_user_context(db, user_id, group_ids, is_admin)
|
|
|
|
# Load resolved permissions
|
|
from app.core.permissions import get_cached_permissions
|
|
redis = get_redis()
|
|
resolved = await get_cached_permissions(db, redis, user_id, tenant_id)
|
|
user_data["permissions"] = resolved.get("permissions", [])
|
|
user_data["denied_permissions"] = resolved.get("denied", [])
|
|
user_data["field_permissions"] = resolved.get("field_permissions", {})
|
|
user_data["is_system_admin"] = resolved.get("is_system_admin", False)
|
|
|
|
return user_data
|
|
|
|
|
|
async def get_current_user_or_bearer(
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
redis: aioredis.Redis = Depends(get_redis_dep),
|
|
) -> dict[str, Any]:
|
|
"""Get current user from session cookie OR Bearer token.
|
|
|
|
Tries session auth first, falls back to Bearer token.
|
|
Used by MCP routes that accept both auth methods.
|
|
"""
|
|
auth_header = request.headers.get("Authorization", "")
|
|
if auth_header.startswith("Bearer "):
|
|
return await get_current_user_bearer(request, db)
|
|
return await get_current_user(request, db, redis)
|
|
|
|
|
|
async def require_admin(
|
|
current_user: dict[str, Any] = Depends(get_current_user),
|
|
) -> dict[str, Any]:
|
|
"""Require admin role (legacy + new permission system).
|
|
|
|
Legacy role string 'admin' is deprecated — log a warning when used.
|
|
New system uses is_system_admin or *:* permission.
|
|
"""
|
|
if current_user.get("is_system_admin"):
|
|
return current_user
|
|
|
|
# Legacy role string fallback — deprecated
|
|
if current_user.get("role") == "admin":
|
|
logger.warning(
|
|
"Legacy role string 'admin' used for user=%s — deprecated, "
|
|
"migrate to is_system_admin or *:* permission",
|
|
current_user.get("user_id"),
|
|
)
|
|
return current_user
|
|
|
|
# New permission system check
|
|
from app.core.permissions import check_permission
|
|
|
|
if check_permission(current_user, "*:*"):
|
|
return current_user
|
|
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail={"detail": "Admin access required", "code": "forbidden"},
|
|
)
|
|
|
|
|
|
async def require_write(
|
|
current_user: dict[str, Any] = Depends(get_current_user),
|
|
) -> dict[str, Any]:
|
|
"""Require write permission (admin, editor, or custom role with write perms).
|
|
|
|
Legacy role strings 'admin'/'editor' are deprecated — log a warning when used.
|
|
New system checks specific module:write permissions instead of broad wildcards.
|
|
"""
|
|
if current_user.get("is_system_admin"):
|
|
return current_user
|
|
|
|
# Legacy role string fallback — deprecated
|
|
role = current_user.get("role", "viewer")
|
|
if role in ("admin", "editor"):
|
|
logger.warning(
|
|
"Legacy role string '%s' used for user=%s in require_write — deprecated, "
|
|
"migrate to specific module:write permissions",
|
|
role, current_user.get("user_id"),
|
|
)
|
|
return current_user
|
|
|
|
# Check via permission system for specific write permissions
|
|
from app.core.permissions import check_permission
|
|
|
|
for perm in _WRITE_PERMISSIONS:
|
|
if check_permission(current_user, perm):
|
|
return current_user
|
|
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail={"detail": "Write access required", "code": "forbidden"},
|
|
)
|
|
|
|
|
|
def require_permission(permission: str):
|
|
"""FastAPI dependency factory: require a specific permission.
|
|
|
|
Usage:
|
|
@router.get("/contacts", dependencies=[Depends(require_permission("contacts:read"))])
|
|
"""
|
|
async def _check(
|
|
current_user: dict[str, Any] = Depends(get_current_user),
|
|
) -> dict[str, Any]:
|
|
if current_user.get("is_system_admin"):
|
|
return current_user
|
|
from app.core.permissions import check_permission
|
|
|
|
if check_permission(current_user, permission):
|
|
return current_user
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail={
|
|
"detail": f"Permission '{permission}' required",
|
|
"code": "forbidden",
|
|
},
|
|
)
|
|
|
|
return _check
|
|
|
|
|
|
def require_field_access(module: str, field: str, default: str = "read"):
|
|
"""FastAPI dependency factory: require field-level access.
|
|
|
|
Usage:
|
|
@router.get("/contacts/{id}", dependencies=[Depends(require_field_access("contacts", "annual_revenue"))])
|
|
"""
|
|
async def _check(
|
|
current_user: dict[str, Any] = Depends(get_current_user),
|
|
) -> dict[str, Any]:
|
|
if current_user.get("is_system_admin"):
|
|
return current_user
|
|
from app.core.permissions import check_field_access
|
|
|
|
access = check_field_access(current_user, module, field, default)
|
|
if access == "hidden":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail={
|
|
"detail": f"Field '{field}' is hidden",
|
|
"code": "field_hidden",
|
|
},
|
|
)
|
|
return current_user
|
|
|
|
return _check
|
|
|
|
|
|
async def get_tenant_id(
|
|
current_user: dict[str, Any] = Depends(get_current_user),
|
|
) -> uuid.UUID:
|
|
"""Extract tenant_id from current user session."""
|
|
return uuid.UUID(current_user["tenant_id"])
|
|
|
|
|
|
async def get_current_user_id(
|
|
current_user: dict[str, Any] = Depends(get_current_user),
|
|
) -> uuid.UUID:
|
|
"""Extract user_id from current user session."""
|
|
return uuid.UUID(current_user["user_id"])
|
|
|
|
|
|
def require_active_plugin(plugin_name: str):
|
|
"""FastAPI dependency factory: require that a plugin is active.
|
|
|
|
Checks both global activation (permission registry) and per-tenant
|
|
activation (tenant_plugin_activation table).
|
|
|
|
Uses the current_user dependency to get tenant_id — does NOT guess
|
|
the tenant from a new DB session via current_setting().
|
|
|
|
Uses Redis cache for per-tenant check to avoid DB query on every request.
|
|
Cache key: plugin-activation:{tenant_id}:{plugin_name}
|
|
TTL: 60 seconds. Invalidated on activate/deactivate.
|
|
|
|
Returns 403 if the plugin is not active.
|
|
Fails closed (503) on errors.
|
|
"""
|
|
async def _check(
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> None:
|
|
from app.core.permission_registry import get_permission_registry
|
|
try:
|
|
registry = get_permission_registry()
|
|
if not registry.is_plugin_active(plugin_name):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail={
|
|
"detail": f"Plugin '{plugin_name}' is not active",
|
|
"code": "plugin_inactive",
|
|
},
|
|
)
|
|
# Get tenant_id from existing db session (NOT a new session)
|
|
# The tenant context is set by middleware/get_current_user on this same session
|
|
from sqlalchemy import text as sa_text
|
|
|
|
result = await db.execute(
|
|
sa_text("SELECT current_setting('app.current_tenant_id', true)::uuid")
|
|
)
|
|
tenant_id = result.scalar()
|
|
|
|
if tenant_id is None:
|
|
# No tenant context — plugin is active by default (backward compatible)
|
|
return
|
|
|
|
# Per-tenant activation check with Redis cache
|
|
from app.core.redis import get_redis
|
|
from sqlalchemy import text
|
|
import json
|
|
|
|
redis = get_redis()
|
|
if redis is not None:
|
|
cache_key = f"plugin-activation:{tenant_id}:{plugin_name}"
|
|
cached = await redis.get(cache_key)
|
|
if cached is not None:
|
|
is_active = json.loads(cached)
|
|
if not is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail={
|
|
"detail": f"Plugin '{plugin_name}' is not active for this tenant",
|
|
"code": "plugin_inactive_tenant",
|
|
},
|
|
)
|
|
return # Cache hit — plugin is active for this tenant
|
|
|
|
# Cache miss — query DB using the existing db session (tenant context already set)
|
|
result = await db.execute(
|
|
text("""
|
|
SELECT is_active FROM tenant_plugin_activation
|
|
WHERE plugin_name = :name
|
|
AND tenant_id = :tid
|
|
"""),
|
|
{"name": plugin_name, "tid": tenant_id},
|
|
)
|
|
row = result.first()
|
|
if row is not None:
|
|
is_active = row[0]
|
|
if redis is not None:
|
|
await redis.setex(cache_key, 60, json.dumps(is_active))
|
|
if not is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail={
|
|
"detail": f"Plugin '{plugin_name}' is not active for this tenant",
|
|
"code": "plugin_inactive_tenant",
|
|
},
|
|
)
|
|
else:
|
|
# No entry = default active (backward compatible)
|
|
if redis is not None:
|
|
await redis.setex(cache_key, 60, json.dumps(True))
|
|
except HTTPException:
|
|
raise
|
|
except Exception as exc:
|
|
# Fail-closed: if registry check fails, deny access (P1.2 fix)
|
|
logger.error("Plugin activation check failed for '%s': %s", plugin_name, exc)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail={"detail": f"Plugin activation check failed", "code": "plugin_check_error"},
|
|
)
|
|
|
|
return _check
|