112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
from typing import Optional
|
|
from uuid import UUID
|
|
from fastapi import Request, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.database import get_db
|
|
from app.modules.auth.models.user_model import User
|
|
from app.middleware.auth import get_current_user
|
|
from app.modules.tenant.models.tenant_model import Tenant
|
|
|
|
|
|
def get_tenant_from_header(
|
|
request: Request, db: Session = Depends(get_db)
|
|
) -> Optional[Tenant]:
|
|
"""
|
|
Extracts X-Tenant-ID from the request headers and validates it.
|
|
Used for public routes like registration where the user is not authenticated yet.
|
|
"""
|
|
tenant_id_str = request.headers.get("X-Tenant-ID")
|
|
if not tenant_id_str:
|
|
default_tenant = db.query(Tenant).filter(Tenant.slug == "default").first()
|
|
return default_tenant
|
|
|
|
try:
|
|
tenant_id = UUID(tenant_id_str)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid X-Tenant-ID format"
|
|
)
|
|
|
|
tenant = db.query(Tenant).filter(Tenant.id == tenant_id).first()
|
|
if not tenant:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found"
|
|
)
|
|
|
|
if not tenant.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Tenant is currently inactive"
|
|
)
|
|
|
|
return tenant
|
|
|
|
|
|
def get_tenant_id(user: User = Depends(get_current_user)) -> Optional[UUID]:
|
|
"""
|
|
Extracts tenant_id from the authenticated user context.
|
|
Returns None for superadmins.
|
|
"""
|
|
return user.tenant_id
|
|
|
|
|
|
def is_superadmin(user: User = Depends(get_current_user)) -> bool:
|
|
"""
|
|
Checks whether the user is a superadmin.
|
|
|
|
Reads the explicit `is_superadmin` flag, and nothing else.
|
|
|
|
It used to also accept `tenant_id IS NULL`. That made the *absence* of
|
|
tenant context a grant of authority: any bug that dropped the tenant did not
|
|
deny, it escalated. The fallback was kept for exactly one release so that
|
|
reverting the B1.0 migration could not lock the operators out mid-deploy;
|
|
that release has now shipped and this is its removal.
|
|
|
|
**Deploy order matters.** This change assumes `users.is_superadmin` exists
|
|
and has been backfilled — that is migration `b1_0_explicit_superadmin`,
|
|
which sets the flag for every active user with no tenant. Ship this without
|
|
that migration and every superadmin loses access. `scripts/b1_preflight.py`
|
|
reports anyone still relying on the old signal before you find out the hard
|
|
way.
|
|
|
|
A user who loses tenant context now has no tenant and no privilege, which is
|
|
the whole point.
|
|
"""
|
|
return bool(getattr(user, "is_superadmin", False))
|
|
|
|
|
|
def require_superadmin(user: User = Depends(get_current_user)) -> User:
|
|
"""
|
|
Dependency that enforces superadmin access.
|
|
Returns the user if superadmin, raises HTTP 403 otherwise.
|
|
"""
|
|
if not is_superadmin(user):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Superadmin access required in the Document Vault",
|
|
)
|
|
return user
|
|
|
|
|
|
def require_tenant(request: Request, user: User = Depends(get_current_user)) -> UUID:
|
|
"""
|
|
Dependency that enforces tenant context.
|
|
Superadmins can provide X-Tenant-ID to simulate tenant context.
|
|
"""
|
|
if user.tenant_id is None:
|
|
tenant_id_str = request.headers.get("X-Tenant-ID")
|
|
if not tenant_id_str:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Superadmins must provide X-Tenant-ID header to perform this operation",
|
|
)
|
|
try:
|
|
return UUID(tenant_id_str)
|
|
except ValueError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Invalid X-Tenant-ID format",
|
|
)
|
|
|
|
return user.tenant_id
|