108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
"""Cookie signing helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
import time
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
from fastapi import HTTPException, Request, Response, WebSocket, status
|
|
from itsdangerous import BadSignature, URLSafeSerializer
|
|
|
|
from app.config import Settings
|
|
|
|
STUDENT_COOKIE = "qz_student"
|
|
ADMIN_COOKIE = "qz_admin"
|
|
STUDENT_MAX_AGE = 31_536_000
|
|
ADMIN_MAX_AGE = 86_400
|
|
|
|
|
|
def serializer(settings: Settings) -> URLSafeSerializer:
|
|
if not settings.secret_key:
|
|
raise HTTPException(status_code=500, detail="QUIZ_SECRET_KEY is not configured")
|
|
return URLSafeSerializer(settings.secret_key, salt="quiz-cookie-v1")
|
|
|
|
|
|
def sign_student(settings: Settings, sid: str, student_id: str, name: str, cookie_id: str | None = None) -> str:
|
|
payload = {
|
|
"sid": sid,
|
|
"student_id": student_id.strip(),
|
|
"name": name.strip(),
|
|
"cookie_id": cookie_id or str(uuid4()),
|
|
}
|
|
return serializer(settings).dumps(payload)
|
|
|
|
|
|
def sign_admin(settings: Settings) -> str:
|
|
return serializer(settings).dumps({"is_admin": True, "ts": int(time.time())})
|
|
|
|
|
|
def loads_cookie(settings: Settings, value: str | None) -> dict[str, Any] | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
payload = serializer(settings).loads(value)
|
|
except BadSignature:
|
|
return None
|
|
return payload if isinstance(payload, dict) else None
|
|
|
|
|
|
def get_student_identity(settings: Settings, request: Request, sid: str | None = None) -> dict[str, Any] | None:
|
|
payload = loads_cookie(settings, request.cookies.get(STUDENT_COOKIE))
|
|
if not payload:
|
|
return None
|
|
if sid is not None and payload.get("sid") != sid:
|
|
return None
|
|
required = {"sid", "student_id", "name", "cookie_id"}
|
|
return payload if required.issubset(payload) else None
|
|
|
|
|
|
def get_student_identity_ws(settings: Settings, websocket: WebSocket, sid: str) -> dict[str, Any] | None:
|
|
payload = loads_cookie(settings, websocket.cookies.get(STUDENT_COOKIE))
|
|
if not payload or payload.get("sid") != sid:
|
|
return None
|
|
required = {"sid", "student_id", "name", "cookie_id"}
|
|
return payload if required.issubset(payload) else None
|
|
|
|
|
|
def require_admin_request(settings: Settings, request: Request) -> None:
|
|
payload = loads_cookie(settings, request.cookies.get(ADMIN_COOKIE))
|
|
if not payload or payload.get("is_admin") is not True:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Admin login required")
|
|
|
|
|
|
def is_admin_ws(settings: Settings, websocket: WebSocket) -> bool:
|
|
payload = loads_cookie(settings, websocket.cookies.get(ADMIN_COOKIE))
|
|
return bool(payload and payload.get("is_admin") is True)
|
|
|
|
|
|
def verify_admin_password(settings: Settings, password: str) -> bool:
|
|
if not settings.admin_password:
|
|
return False
|
|
return secrets.compare_digest(password, settings.admin_password)
|
|
|
|
|
|
def set_student_cookie(settings: Settings, response: Response, value: str) -> None:
|
|
response.set_cookie(
|
|
STUDENT_COOKIE,
|
|
value,
|
|
max_age=STUDENT_MAX_AGE,
|
|
httponly=True,
|
|
samesite="lax",
|
|
secure=settings.secure_cookies,
|
|
path="/",
|
|
)
|
|
|
|
|
|
def set_admin_cookie(settings: Settings, response: Response, value: str) -> None:
|
|
response.set_cookie(
|
|
ADMIN_COOKIE,
|
|
value,
|
|
max_age=ADMIN_MAX_AGE,
|
|
httponly=True,
|
|
samesite="lax",
|
|
secure=settings.secure_cookies,
|
|
path="/",
|
|
)
|