40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
"""Shared FastAPI dependencies: origin check and current-user resolution."""
|
|
from datetime import timezone
|
|
|
|
from fastapi import Depends, HTTPException, Request
|
|
from sqlalchemy.orm import Session as DbSession
|
|
|
|
from app import config, security
|
|
from app.db import get_db
|
|
from app.models import User, UserSession, utcnow
|
|
|
|
|
|
def require_same_origin(request: Request) -> None:
|
|
if not security.same_origin(request):
|
|
raise HTTPException(status_code=403, detail="Cross-origin request rejected")
|
|
|
|
|
|
def _as_utc(value):
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=timezone.utc)
|
|
return value.astimezone(timezone.utc)
|
|
|
|
|
|
def get_current_user(
|
|
request: Request, db: DbSession = Depends(get_db)
|
|
) -> User:
|
|
token = request.cookies.get(config.COOKIE_NAME)
|
|
if not token:
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
|
|
session = db.get(UserSession, security.hash_token(token))
|
|
if session is None:
|
|
raise HTTPException(status_code=401, detail="Not authenticated")
|
|
|
|
if _as_utc(session.expires_at) <= utcnow():
|
|
db.delete(session)
|
|
db.commit()
|
|
raise HTTPException(status_code=401, detail="Session expired")
|
|
|
|
return session.user
|