103 lines
3.3 KiB
Python
103 lines
3.3 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
from uuid import uuid4
|
|
|
|
import bcrypt
|
|
from fastapi import Cookie, Depends, HTTPException, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy.orm import Session as DbSession
|
|
|
|
from app.config import SECRET_KEY, SESSION_COOKIE_NAME
|
|
from app.database import get_db
|
|
from app.models import ApiKey, User, UserSession
|
|
|
|
|
|
# ---------- Password ----------
|
|
|
|
def hash_password(password: str) -> str:
|
|
return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
|
|
|
|
|
|
def verify_password(password: str, password_hash: str) -> bool:
|
|
return bcrypt.checkpw(password.encode(), password_hash.encode())
|
|
|
|
|
|
# ---------- Session ----------
|
|
|
|
def create_session(user_id: int, db: DbSession, remember_me: bool = False) -> str:
|
|
token = uuid4().hex
|
|
expires_at = (datetime.now(timezone.utc) + timedelta(days=365)) if remember_me else None
|
|
db.add(UserSession(token=token, user_id=user_id, expires_at=expires_at))
|
|
db.commit()
|
|
return token
|
|
|
|
|
|
def destroy_session(token: str, db: DbSession) -> None:
|
|
db.query(UserSession).filter(UserSession.token == token).delete()
|
|
db.commit()
|
|
|
|
|
|
def get_session_user_id(token: str | None, db: DbSession) -> int | None:
|
|
if token is None:
|
|
return None
|
|
s = db.query(UserSession).filter(UserSession.token == token).first()
|
|
if s is None:
|
|
return None
|
|
if s.expires_at and s.expires_at.replace(tzinfo=timezone.utc) < datetime.now(timezone.utc):
|
|
db.delete(s)
|
|
db.commit()
|
|
return None
|
|
return s.user_id
|
|
|
|
|
|
def cleanup_expired_sessions(db: DbSession) -> None:
|
|
now = datetime.now(timezone.utc)
|
|
db.query(UserSession).filter(
|
|
UserSession.expires_at.isnot(None),
|
|
UserSession.expires_at < now,
|
|
).delete()
|
|
db.commit()
|
|
|
|
|
|
# ---------- Web Dependencies ----------
|
|
|
|
def get_current_user_web(
|
|
request: Request,
|
|
db: DbSession = Depends(get_db),
|
|
) -> User:
|
|
token = request.cookies.get(SESSION_COOKIE_NAME)
|
|
uid = get_session_user_id(token, db)
|
|
if uid is None:
|
|
raise HTTPException(status_code=303, headers={"Location": "/login"})
|
|
user = db.query(User).filter(User.id == uid).first()
|
|
if user is None:
|
|
raise HTTPException(status_code=303, headers={"Location": "/login"})
|
|
return user
|
|
|
|
|
|
def redirect_if_not_logged_in(request: Request, db: DbSession) -> User | None:
|
|
"""Return user if logged in, else None — caller decides redirect."""
|
|
token = request.cookies.get(SESSION_COOKIE_NAME)
|
|
uid = get_session_user_id(token, db)
|
|
if uid is None:
|
|
return None
|
|
return db.query(User).filter(User.id == uid).first()
|
|
|
|
|
|
# ---------- API Dependencies ----------
|
|
|
|
def get_current_user_api(
|
|
request: Request,
|
|
db: DbSession = Depends(get_db),
|
|
) -> User:
|
|
auth_header = request.headers.get("Authorization", "")
|
|
if not auth_header.startswith("Bearer "):
|
|
raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")
|
|
token = auth_header[7:]
|
|
api_key = db.query(ApiKey).filter(ApiKey.key == token).first()
|
|
if api_key is None:
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
user = db.query(User).filter(User.id == api_key.user_id).first()
|
|
if user is None:
|
|
raise HTTPException(status_code=401, detail="User not found")
|
|
return user
|