Admin mode. Data driven content sections. Bugfixes.

backend/app/auth.py

@@ -0,0 +1,109 @@
+"""Auth helpers, JWT token creation/verification, and FastAPI dependencies."""
+
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+import httpx
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from jose import JWTError, jwt
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config import settings
+from app.database import get_session
+from app.user_models import User
+
+
+class AuthBearer(HTTPBearer):
+    """HTTP Bearer scheme that doesn't auto-fail on missing token."""
+    def __init__(self, auto_error: bool = False):
+        super().__init__(auto_error=auto_error)
+
+
+bearer = AuthBearer(auto_error=False)
+
+
+# ── Token helpers ──────────────────────────────────────────────
+
+def create_access_token(user_id: int, is_admin: bool) -> str:
+    """Create a JWT access token for the given user."""
+    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.JWT_EXPIRE_MINUTES)
+    payload = {
+        "sub": str(user_id),
+        "is_admin": is_admin,
+        "exp": expire,
+    }
+    return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")
+
+
+def decode_token(token: str) -> dict:
+    """Decode and verify a JWT token. Raises HTTPException 401 on failure."""
+    try:
+        payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
+        user_id = payload.get("sub")
+        if user_id is None:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
+        return payload
+    except JWTError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token")
+
+
+# ── FastAPI dependencies ──────────────────────────────────────
+
+async def get_current_user(
+    credentials: Optional[HTTPAuthorizationCredentials] = Depends(bearer),
+    session: AsyncSession = Depends(get_session),
+) -> User:
+    """Validate JWT and return the current User."""
+    if credentials is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
+
+    payload = decode_token(credentials.credentials)
+    user_id = int(payload["sub"])
+
+    result = await session.execute(select(User).where(User.id == user_id))
+    user = result.scalar_one_or_none()
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
+
+    return user
+
+
+async def get_current_admin_user(
+    current_user: User = Depends(get_current_user),
+) -> User:
+    """Raise 403 if the current user is not an admin."""
+    if not current_user.is_admin:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
+    return current_user
+
+
+# ── Google OIDC verification ─────────────────────────────────
+
+_GOOGLE_CERTS_URL = "https://www.googleapis.com/oauth2/v3/certs"
+_GOOGLE_TOKEN_INFO_URL = "https://oauth2.googleapis.com/tokeninfo"
+
+
+async def verify_google_id_token(id_token: str) -> dict:
+    """Verify a Google ID token and return (email, name, picture)."""
+    async with httpx.AsyncClient() as client:
+        # Use tokeninfo endpoint for simplicity (no signature parsing needed)
+        resp = await client.get(_GOOGLE_TOKEN_INFO_URL, params={"idtoken": id_token})
+
+    if resp.status_code != 200:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid Google ID token",
+        )
+
+    info = resp.json()
+    email = info.get("email")
+    if not email:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="No email in Google token")
+
+    return {
+        "email": email,
+        "name": info.get("name", email.split("@")[0]),
+        "picture": info.get("picture", ""),
+    }