Spaces:
Runtime error
Runtime error
File size: 3,795 Bytes
00a4363 013ff92 00a4363 | 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 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | """JWT helpers. Revocation = bump User.token_version; stale tokens get the same 401 as expired ones."""
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.config import settings
from app.db import get_db
from app.models import User, Workspace, WorkspaceMember
_bearer = HTTPBearer(auto_error=False)
_UNAUTHORIZED = HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid or expired token")
def _encode(user_id: uuid.UUID, token_version: int, kind: str, ttl: timedelta) -> str:
now = datetime.now(UTC)
payload = {"sub": str(user_id), "tv": token_version, "typ": kind, "iat": now, "exp": now + ttl}
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
def create_access_token(user_id: uuid.UUID, token_version: int) -> str:
return _encode(user_id, token_version, "access", timedelta(hours=settings.jwt_expiration_hours))
def create_refresh_token(user_id: uuid.UUID, token_version: int) -> str:
return _encode(user_id, token_version, "refresh", timedelta(days=settings.jwt_refresh_expiration_days))
def create_login_ticket(user_id: uuid.UUID, token_version: int, created: bool) -> str:
"""Short lived handoff from the OAuth callback redirect to the SPA, swapped for real tokens."""
now = datetime.now(UTC)
payload = {"sub": str(user_id), "tv": token_version, "typ": "ticket", "new": created, "iat": now,
"exp": now + timedelta(seconds=60)}
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
def encode_signed(payload: dict, ttl: timedelta) -> str:
return jwt.encode({**payload, "exp": datetime.now(UTC) + ttl}, settings.jwt_secret,
algorithm=settings.jwt_algorithm)
def decode_token_full(token: str, expected_type: str = "access") -> dict:
try:
payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
except jwt.PyJWTError as exc:
raise _UNAUTHORIZED from exc
if payload.get("typ") != expected_type:
raise _UNAUTHORIZED
return payload
def token_version_is_current(user: User, payload: dict) -> bool:
return payload.get("tv") == user.token_version
def load_user_from_token(db: Session, token: str, expected_type: str = "access") -> User:
payload = decode_token_full(token, expected_type)
try:
user_id = uuid.UUID(payload["sub"])
except (KeyError, ValueError) as exc:
raise _UNAUTHORIZED from exc
user = db.get(User, user_id)
if not user or not user.is_active or user.deleted_at or not token_version_is_current(user, payload):
raise _UNAUTHORIZED
return user
def get_current_user(
creds: HTTPAuthorizationCredentials | None = Depends(_bearer), db: Session = Depends(get_db)
) -> User:
if not creds:
raise _UNAUTHORIZED
return load_user_from_token(db, creds.credentials)
@dataclass
class Ctx:
user: User
workspace: Workspace
db: Session
def get_ctx(user: User = Depends(get_current_user), db: Session = Depends(get_db)) -> Ctx:
"""Current user plus their active workspace (first membership for now; team switching in Phase 4)."""
workspace = db.scalar(
select(Workspace)
.join(WorkspaceMember, WorkspaceMember.workspace_id == Workspace.id)
.where(WorkspaceMember.user_id == user.id)
.order_by(WorkspaceMember.created_at)
.limit(1)
)
if not workspace:
raise HTTPException(status.HTTP_409_CONFLICT, "No workspace for user")
return Ctx(user=user, workspace=workspace, db=db)
|