"""Shared API dependencies: dashboard user auth (JWT) and satellite auth (API key).""" from fastapi import Depends, Header, Request from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db from app.core.exceptions import ForbiddenError, UnauthorizedError from app.core.security import decode_token from app.models.satellite import Satellite, hash_api_key from app.models.user import User bearer_scheme = HTTPBearer(auto_error=False) async def get_current_user( credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme), db: AsyncSession = Depends(get_db), ) -> User: if credentials is None: raise UnauthorizedError("Authorization header fehlt") payload = decode_token(credentials.credentials) username = payload.get("sub") if not username: raise UnauthorizedError("Token enthält keinen Benutzer") result = await db.execute(select(User).where(User.username == username)) user = result.scalar_one_or_none() if not user or not user.is_active: raise UnauthorizedError("Benutzer unbekannt oder deaktiviert") return user async def require_admin(user: User = Depends(get_current_user)) -> User: if not user.is_admin: raise ForbiddenError("Administratorrechte erforderlich") return user async def get_current_satellite( x_api_key: str | None = Header(default=None), db: AsyncSession = Depends(get_db), ) -> Satellite: """Authenticate a satellite by its API key (X-Api-Key header).""" if not x_api_key: raise UnauthorizedError("X-Api-Key header fehlt") result = await db.execute( select(Satellite).where(Satellite.api_key_hash == hash_api_key(x_api_key)) ) satellite = result.scalar_one_or_none() if not satellite or not satellite.is_active: raise UnauthorizedError("Satellite unbekannt oder deaktiviert") return satellite def client_ip(request: Request) -> str | None: return request.client.host if request.client else None