Bernt
aee0f09db8
- Datafabrik: Dockerfile fix, agentorkestrering fungerar - Vision: Identify-modell, FAISS, OCR alla testade - API: Alla 7 integrationstester passerade - Upplösare: Entitetsupplösning verifierad
116 lines
3.4 KiB
Python
116 lines
3.4 KiB
Python
"""
|
|
LandveX Admin Backend — API Dependencies
|
|
FastAPI dependencies for auth, database, and request context
|
|
"""
|
|
from typing import Optional, Dict, Any
|
|
from uuid import UUID
|
|
|
|
from fastapi import Depends, Request, HTTPException, status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.session import get_db
|
|
from app.core.security import get_current_user, require_admin
|
|
from app.services.tenant_service import TenantService
|
|
from app.services.audit_service import AuditService
|
|
from app.core.exceptions import TenantNotFoundException, UnauthorizedTenantAccessException
|
|
|
|
|
|
security = HTTPBearer(auto_error=False)
|
|
|
|
|
|
async def get_tenant_service(
|
|
db: AsyncSession = Depends(get_db)
|
|
) -> TenantService:
|
|
"""Dependency to get tenant service."""
|
|
return TenantService(db)
|
|
|
|
|
|
async def get_audit_service(
|
|
db: AsyncSession = Depends(get_db)
|
|
) -> AuditService:
|
|
"""Dependency to get audit service."""
|
|
return AuditService(db)
|
|
|
|
|
|
async def get_request_context(
|
|
request: Request,
|
|
current_user: Optional[Dict[str, Any]] = Depends(get_current_user)
|
|
) -> Dict[str, Any]:
|
|
"""Build request context for audit logging."""
|
|
|
|
context = {
|
|
"request_id": request.headers.get("X-Request-ID", ""),
|
|
"request_method": request.method,
|
|
"request_path": str(request.url.path),
|
|
"actor_ip": request.client.host if request.client else None,
|
|
"actor_user_agent": request.headers.get("user-agent", ""),
|
|
}
|
|
|
|
if current_user:
|
|
context["actor_id"] = current_user.get("id")
|
|
|
|
return context
|
|
|
|
|
|
async def get_current_tenant_id(
|
|
request: Request
|
|
) -> Optional[str]:
|
|
"""Extract tenant ID from request (header or path)."""
|
|
# Try header first
|
|
tenant_id = request.headers.get("X-Tenant-ID")
|
|
if tenant_id:
|
|
return tenant_id
|
|
|
|
# Try path parameter
|
|
tenant_id = request.path_params.get("tenant_id")
|
|
if tenant_id:
|
|
return tenant_id
|
|
|
|
return None
|
|
|
|
|
|
async def require_tenant_admin(
|
|
tenant_id: UUID,
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
tenant_service: TenantService = Depends(get_tenant_service)
|
|
) -> Dict[str, Any]:
|
|
"""Require admin access to a specific tenant."""
|
|
|
|
user_role = current_user.get("role", "")
|
|
user_tenant = current_user.get("tenant_id")
|
|
|
|
# Global admin can access all tenants
|
|
if user_role == "admin":
|
|
return current_user
|
|
|
|
# Tenant admin can only access their own tenant
|
|
if user_role == "tenant_admin" and str(user_tenant) == str(tenant_id):
|
|
return current_user
|
|
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Admin access required for tenant {tenant_id}"
|
|
)
|
|
|
|
|
|
async def optional_auth(
|
|
credentials: HTTPAuthorizationCredentials = Depends(security)
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Optional authentication - returns user if token valid, None otherwise."""
|
|
if credentials is None:
|
|
return None
|
|
|
|
from app.core.security import decode_token
|
|
payload = decode_token(credentials.credentials)
|
|
if payload is None:
|
|
return None
|
|
|
|
return {
|
|
"id": payload.get("sub"),
|
|
"email": payload.get("email"),
|
|
"role": payload.get("role", "user"),
|
|
"tenant_id": payload.get("tenant_id"),
|
|
"permissions": payload.get("permissions", []),
|
|
}
|