admin / Syanpse-Vanguard
public
Syanpse-Vanguard / synapse-vanguard-v3 / sv / auth / session.py
1618 B · main
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 | """JWT session issuance/validation and the `current_user` FastAPI dependency.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta, timezone from uuid import UUID import jwt from fastapi import Depends, HTTPException, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sv.auth.rbac import Role from sv.config import settings _bearer = HTTPBearer(auto_error=True) _ALGO = "HS256" @dataclass class User: id: str username: str role: Role tenant_id: UUID def issue_token(user_id: str, username: str, role: Role, tenant_id: UUID) -> str: now = datetime.now(timezone.utc) payload = { "sub": user_id, "usr": username, "role": role.value, "tid": str(tenant_id), "iat": now, "exp": now + timedelta(seconds=settings.jwt_ttl_seconds), } return jwt.encode(payload, settings.jwt_secret, algorithm=_ALGO) async def current_user( creds: HTTPAuthorizationCredentials = Depends(_bearer), ) -> User: try: payload = jwt.decode(creds.credentials, settings.jwt_secret, algorithms=[_ALGO]) except jwt.PyJWTError: raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid or expired token") try: return User( id=payload["sub"], username=payload["usr"], role=Role(payload["role"]), tenant_id=UUID(payload["tid"]), ) except (KeyError, ValueError): raise HTTPException(status.HTTP_401_UNAUTHORIZED, "malformed token claims") |