88 lines
2.5 KiB
Python
88 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
import os
|
|
|
|
import jwt
|
|
from fastapi import Depends, HTTPException
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AuthUser:
|
|
username: str
|
|
role: str
|
|
|
|
|
|
security = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def _get_secret() -> str:
|
|
secret = os.getenv("JWT_SECRET", "").strip()
|
|
if not secret:
|
|
raise RuntimeError("JWT_SECRET must be set.")
|
|
return secret
|
|
|
|
|
|
def _get_algorithm() -> str:
|
|
return os.getenv("JWT_ALGORITHM", "HS256").strip() or "HS256"
|
|
|
|
|
|
def _get_expires_hours() -> int:
|
|
raw = os.getenv("JWT_EXPIRES_HOURS", "8").strip() or "8"
|
|
try:
|
|
expires_hours = int(raw)
|
|
except ValueError as exc:
|
|
raise RuntimeError("JWT_EXPIRES_HOURS must be an integer.") from exc
|
|
if expires_hours <= 0:
|
|
raise RuntimeError("JWT_EXPIRES_HOURS must be greater than zero.")
|
|
return expires_hours
|
|
|
|
|
|
def create_access_token(username: str, role: str) -> tuple[str, datetime]:
|
|
now = datetime.now(timezone.utc)
|
|
expires_at = now + timedelta(hours=_get_expires_hours())
|
|
payload = {
|
|
"sub": username,
|
|
"role": role,
|
|
"iat": int(now.timestamp()),
|
|
"exp": int(expires_at.timestamp()),
|
|
}
|
|
token = jwt.encode(payload, _get_secret(), algorithm=_get_algorithm())
|
|
return token, expires_at
|
|
|
|
|
|
def decode_access_token(token: str) -> AuthUser:
|
|
try:
|
|
payload = jwt.decode(token, _get_secret(), algorithms=[_get_algorithm()])
|
|
except jwt.ExpiredSignatureError as exc:
|
|
raise HTTPException(status_code=401, detail="Token has expired") from exc
|
|
except jwt.InvalidTokenError as exc:
|
|
raise HTTPException(status_code=401, detail="Invalid token") from exc
|
|
|
|
username = str(payload.get("sub") or "").strip()
|
|
role = str(payload.get("role") or "").strip().lower()
|
|
if not username or role not in {"admin", "viewer"}:
|
|
raise HTTPException(status_code=401, detail="Invalid token payload")
|
|
|
|
return AuthUser(username=username, role=role)
|
|
|
|
|
|
def validate_auth_environment() -> None:
|
|
_get_secret()
|
|
_get_algorithm()
|
|
_get_expires_hours()
|
|
|
|
|
|
def get_current_user(credentials: HTTPAuthorizationCredentials | None = Depends(security)) -> AuthUser:
|
|
if credentials is None or credentials.scheme.lower() != "bearer":
|
|
raise HTTPException(status_code=401, detail="Authorization token is required")
|
|
return decode_access_token(credentials.credentials)
|
|
|
|
|
|
def require_admin(user: AuthUser = Depends(get_current_user)) -> AuthUser:
|
|
if user.role != "admin":
|
|
raise HTTPException(status_code=403, detail="Admin access required")
|
|
return user
|