feat: initial postcard manager - FastAPI + Jinja2 + SQLite

app/auth.py

@@ -0,0 +1,86 @@
+from datetime import datetime, timezone
+from uuid import uuid4
+
+import bcrypt
+from fastapi import Cookie, Depends, HTTPException, Request
+from fastapi.responses import RedirectResponse
+from sqlalchemy.orm import Session
+
+from app.config import SECRET_KEY, SESSION_COOKIE_NAME
+from app.database import get_db
+from app.models import ApiKey, User
+
+# ---------- 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 ----------
+
+# In-memory session store: token -> user_id
+_sessions: dict[str, int] = {}
+
+
+def create_session(user_id: int) -> str:
+    token = uuid4().hex
+    _sessions[token] = user_id
+    return token
+
+
+def destroy_session(token: str) -> None:
+    _sessions.pop(token, None)
+
+
+def get_session_user_id(token: str | None) -> int | None:
+    if token is None:
+        return None
+    return _sessions.get(token)
+
+
+# ---------- Web Dependencies ----------
+
+def get_current_user_web(
+    request: Request,
+    db: Session = Depends(get_db),
+) -> User:
+    token = request.cookies.get(SESSION_COOKIE_NAME)
+    uid = get_session_user_id(token)
+    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: Session) -> 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)
+    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: Session = 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