wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
832 lines
28 KiB
Python
832 lines
28 KiB
Python
"""Auth router — login, logout, status, registration, profile, and user-management endpoints."""
|
|
|
|
from contextvars import Token as _CtxToken
|
|
import logging
|
|
import re
|
|
|
|
from fastapi import (
|
|
APIRouter,
|
|
Cookie,
|
|
Depends,
|
|
File,
|
|
Header,
|
|
HTTPException,
|
|
Response,
|
|
UploadFile,
|
|
WebSocket,
|
|
status,
|
|
)
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel, field_validator
|
|
|
|
from deeptutor.services.config import load_auth_settings
|
|
|
|
# SameSite=None lets the cookie work when the browser accesses the frontend via
|
|
# 127.0.0.1 and the backend via localhost (different origins on the same machine).
|
|
# Browsers require Secure=True for SameSite=None, but that needs HTTPS — so in
|
|
# local dev we fall back to SameSite=Lax and tell users to use localhost:// URLs.
|
|
_SECURE = bool(load_auth_settings()["cookie_secure"])
|
|
_SAMESITE = "none" if _SECURE else "lax"
|
|
|
|
from deeptutor.multi_user.context import set_current_user, user_from_token_payload
|
|
from deeptutor.multi_user.paths import local_admin_user
|
|
from deeptutor.services.auth import (
|
|
AUTH_ENABLED,
|
|
POCKETBASE_ENABLED,
|
|
TOKEN_EXPIRE_HOURS,
|
|
TokenPayload,
|
|
add_user,
|
|
authenticate,
|
|
authenticate_pb,
|
|
create_token,
|
|
decode_token,
|
|
delete_user,
|
|
get_user_info,
|
|
is_first_user,
|
|
list_users,
|
|
register_pb,
|
|
set_avatar,
|
|
set_role,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter()
|
|
|
|
_COOKIE_NAME = "dt_token"
|
|
_COOKIE_MAX_AGE = TOKEN_EXPIRE_HOURS * 3600
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
"""Payload for the POST /login endpoint."""
|
|
|
|
username: str
|
|
password: str
|
|
|
|
|
|
class RegisterRequest(BaseModel):
|
|
"""Payload for the POST /register endpoint."""
|
|
|
|
username: str
|
|
password: str
|
|
|
|
@field_validator("username")
|
|
@classmethod
|
|
def username_valid(cls, v: str) -> str:
|
|
import re
|
|
|
|
v = v.strip()
|
|
if not v:
|
|
raise ValueError("Email cannot be empty")
|
|
# Accept standard email addresses (used by PocketBase mode) or plain
|
|
# usernames (used by the built-in SQLite/JSON auth mode).
|
|
email_re = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
|
|
plain_re = re.compile(r"^[A-Za-z0-9_\-.]{3,64}$")
|
|
if not email_re.match(v) and not plain_re.match(v):
|
|
raise ValueError("Enter a valid email address")
|
|
return v
|
|
|
|
@field_validator("password")
|
|
@classmethod
|
|
def password_valid(cls, v: str) -> str:
|
|
if len(v) < 8:
|
|
raise ValueError("Password must be at least 8 characters")
|
|
return v
|
|
|
|
|
|
class SetRoleRequest(BaseModel):
|
|
"""Payload for the PUT /users/{username}/role endpoint."""
|
|
|
|
role: str
|
|
|
|
@field_validator("role")
|
|
@classmethod
|
|
def role_valid(cls, v: str) -> str:
|
|
if v not in ("admin", "user"):
|
|
raise ValueError("Role must be 'admin' or 'user'")
|
|
return v
|
|
|
|
|
|
class AuthStatusResponse(BaseModel):
|
|
"""Response body for the GET /status endpoint."""
|
|
|
|
enabled: bool
|
|
authenticated: bool
|
|
user_id: str | None = None
|
|
username: str | None = None
|
|
role: str | None = None
|
|
is_admin: bool = False
|
|
avatar: str = ""
|
|
|
|
|
|
class UserInfo(BaseModel):
|
|
"""Single user record returned by the GET /users and /profile endpoints."""
|
|
|
|
id: str = ""
|
|
username: str
|
|
role: str
|
|
created_at: str
|
|
disabled: bool = False
|
|
avatar: str = ""
|
|
|
|
|
|
# Markers settable through PUT /profile. Image markers ("img:<version>") are
|
|
# managed exclusively by the upload endpoint so users cannot point their
|
|
# avatar at a file that was never validated.
|
|
_ICON_MARKER_RE = re.compile(r"^icon:[a-z0-9-]{1,32}:[a-z0-9-]{1,32}$")
|
|
|
|
# User ids are generated as "u_<uuid hex>" (plus the "local-admin" /
|
|
# "env-admin" sentinels); reject anything else before it reaches the
|
|
# filesystem layer.
|
|
_USER_ID_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
|
|
|
|
|
|
class UpdateProfileRequest(BaseModel):
|
|
"""Payload for the PUT /profile endpoint."""
|
|
|
|
avatar: str
|
|
|
|
@field_validator("avatar")
|
|
@classmethod
|
|
def avatar_valid(cls, v: str) -> str:
|
|
v = v.strip()
|
|
if v and not _ICON_MARKER_RE.match(v):
|
|
raise ValueError("Avatar must be empty or 'icon:<name>:<color>'")
|
|
return v
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared helper — extract token from cookie or Bearer header
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _bearer_token_from_header(authorization: str | None) -> str | None:
|
|
"""Parse ``Authorization: Bearer <token>`` without using ``HTTPBearer``.
|
|
|
|
``HTTPBearer`` is a class-based dependency whose ``__call__`` is annotated
|
|
``request: Request``. FastAPI doesn't inject a Request into WebSocket
|
|
dependency resolution, which makes ``HTTPBearer`` raise ``TypeError`` the
|
|
moment a router with this dep mounts a WS endpoint. Doing the parse by
|
|
hand keeps ``require_auth`` HTTP/WS-symmetric.
|
|
"""
|
|
if not authorization:
|
|
return None
|
|
parts = authorization.split(None, 1)
|
|
if len(parts) == 2 and parts[0].lower() == "bearer":
|
|
token = parts[1].strip()
|
|
return token or None
|
|
return None
|
|
|
|
|
|
def _extract_token(authorization: str | None, dt_token: str | None) -> str | None:
|
|
return _bearer_token_from_header(authorization) or dt_token
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dependencies — reusable auth guards for other routers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _install_current_user(payload: TokenPayload | None) -> _CtxToken:
|
|
"""Install the request-local current-user ContextVar from an auth result.
|
|
|
|
Single point of truth for ``payload → CurrentUser`` so HTTP and WebSocket
|
|
entry points produce identical user objects. ``payload is None`` means
|
|
"no JWT was required" (AUTH_ENABLED=false) and resolves to the local
|
|
admin user; a non-None payload resolves through ``user_from_token_payload``.
|
|
|
|
Returns the ContextVar reset token. HTTP callers ignore it (the request
|
|
ends with the task, so the var is GC'd with the task context). WebSocket
|
|
callers keep it and call ``reset_current_user`` in their ``finally`` block,
|
|
because a WS connection outlives the dependency-resolution task.
|
|
|
|
⚠ Invariant: every authenticated entry point MUST call this before the
|
|
handler runs. Skipping it leaves ``get_current_path_service()`` falling
|
|
back to the admin workspace — the silent-routing root cause of #481.
|
|
"""
|
|
user = local_admin_user() if payload is None else user_from_token_payload(payload)
|
|
return set_current_user(user)
|
|
|
|
|
|
async def require_auth(
|
|
authorization: str | None = Header(default=None, alias="Authorization"),
|
|
dt_token: str | None = Cookie(default=None),
|
|
) -> TokenPayload | None:
|
|
"""
|
|
FastAPI dependency that enforces authentication when AUTH_ENABLED=true.
|
|
|
|
Accepts the JWT from either:
|
|
- Authorization: Bearer <token> header
|
|
- dt_token cookie
|
|
|
|
``Header`` and ``Cookie`` are kept here in place of ``HTTPBearer`` so the
|
|
function stays usable from WebSocket call sites that don't go through
|
|
FastAPI's standard HTTP request lifecycle.
|
|
|
|
Returns the authenticated TokenPayload, or None if auth is disabled.
|
|
Raises HTTP 401 if auth is enabled but the token is missing or invalid.
|
|
|
|
Declared ``async def`` so the ``set_current_user`` call runs in the same
|
|
asyncio context as the endpoint. A sync dependency is dispatched via
|
|
``anyio.to_thread.run_sync``, which executes the function in a worker
|
|
thread under a *copy* of the request context; any ``ContextVar.set``
|
|
inside that thread is discarded when the thread returns, leaving the
|
|
endpoint to read the unset default. That regression was the root cause
|
|
of #481.
|
|
"""
|
|
if not AUTH_ENABLED:
|
|
_install_current_user(None)
|
|
return None
|
|
|
|
token = _extract_token(authorization, dt_token)
|
|
if not token:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Not authenticated",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
payload = decode_token(token)
|
|
if not payload:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired token",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
_install_current_user(payload)
|
|
return payload
|
|
|
|
|
|
class _WsAuthFailed:
|
|
"""Sentinel: ws_require_auth failed and closed the WebSocket."""
|
|
|
|
|
|
ws_auth_failed: _WsAuthFailed = _WsAuthFailed()
|
|
|
|
|
|
async def ws_require_auth(ws: WebSocket) -> _CtxToken | _WsAuthFailed:
|
|
"""Authenticate a WebSocket connection and set the user ContextVar.
|
|
|
|
Must be called **before** ``ws.accept()`` so the server can reject
|
|
unauthenticated upgrades cleanly.
|
|
|
|
Returns a ContextVar reset token on success, or ``ws_auth_failed``
|
|
on failure (the WebSocket is already closed — the caller should
|
|
``return`` immediately).
|
|
|
|
Usage::
|
|
|
|
user_token = await ws_require_auth(ws)
|
|
if user_token is ws_auth_failed:
|
|
return
|
|
await ws.accept()
|
|
try:
|
|
...
|
|
finally:
|
|
reset_current_user(user_token)
|
|
"""
|
|
if not AUTH_ENABLED:
|
|
return _install_current_user(None)
|
|
|
|
token = ws.query_params.get("token") or ws.cookies.get("dt_token")
|
|
payload = decode_token(token) if token else None
|
|
if not payload:
|
|
await ws.close(code=4001)
|
|
return ws_auth_failed
|
|
|
|
return _install_current_user(payload)
|
|
|
|
|
|
async def require_admin(
|
|
payload: TokenPayload | None = Depends(require_auth),
|
|
) -> TokenPayload:
|
|
"""
|
|
FastAPI dependency that requires the caller to be an admin.
|
|
|
|
Raises HTTP 403 if the authenticated user is not an admin.
|
|
When AUTH_ENABLED=false, all requests are treated as admin.
|
|
|
|
``async def`` mirrors ``require_auth`` so the dependency chain stays on
|
|
the event loop and the user ContextVar set by ``require_auth`` is visible
|
|
to the endpoint.
|
|
"""
|
|
if not AUTH_ENABLED:
|
|
return _local_admin_token_payload()
|
|
|
|
if payload is None or payload.role != "admin":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin access required",
|
|
)
|
|
return payload
|
|
|
|
|
|
def _local_admin_token_payload() -> TokenPayload:
|
|
"""Synthetic admin payload used when AUTH_ENABLED=false.
|
|
|
|
Mirrors the local admin identity (LOCAL_ADMIN_USERNAME / LOCAL_ADMIN_ID)
|
|
so audit logs and self-reference checks behave the same as in multi-user
|
|
mode. Values are kept aligned with ``local_admin_user()`` in
|
|
``deeptutor/multi_user/paths.py``.
|
|
"""
|
|
from deeptutor.multi_user.models import LOCAL_ADMIN_ID, LOCAL_ADMIN_USERNAME
|
|
|
|
return TokenPayload(
|
|
username=LOCAL_ADMIN_USERNAME,
|
|
role="admin",
|
|
user_id=LOCAL_ADMIN_ID,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public endpoints (no auth required)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/status", response_model=AuthStatusResponse)
|
|
async def auth_status(
|
|
authorization: str | None = Header(default=None, alias="Authorization"),
|
|
dt_token: str | None = Cookie(default=None),
|
|
) -> AuthStatusResponse:
|
|
"""Return whether auth is enabled and whether the current request is authenticated."""
|
|
if not AUTH_ENABLED:
|
|
return AuthStatusResponse(
|
|
enabled=False,
|
|
authenticated=True,
|
|
user_id="local-admin",
|
|
username="local",
|
|
role="admin",
|
|
is_admin=True,
|
|
)
|
|
|
|
token = _extract_token(authorization, dt_token)
|
|
payload = decode_token(token) if token else None
|
|
avatar = ""
|
|
if payload is not None:
|
|
info = get_user_info(payload.username)
|
|
if info:
|
|
avatar = str(info.get("avatar") or "")
|
|
return AuthStatusResponse(
|
|
enabled=True,
|
|
authenticated=payload is not None,
|
|
user_id=payload.user_id if payload else None,
|
|
username=payload.username if payload else None,
|
|
role=payload.role if payload else None,
|
|
is_admin=payload.role == "admin" if payload else False,
|
|
avatar=avatar,
|
|
)
|
|
|
|
|
|
@router.post("/login")
|
|
async def login(body: LoginRequest, response: Response) -> dict:
|
|
"""Validate credentials and set a JWT cookie."""
|
|
if not AUTH_ENABLED:
|
|
return {"ok": True, "message": "Auth is disabled — no login required."}
|
|
|
|
if POCKETBASE_ENABLED:
|
|
# PocketBase mode: email = username field for backwards-compat with the
|
|
# existing LoginRequest schema; users can pass their email as "username".
|
|
pb_result = authenticate_pb(body.username, body.password)
|
|
if not pb_result:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Incorrect email or password",
|
|
)
|
|
payload, pb_token = pb_result
|
|
response.set_cookie(
|
|
key=_COOKIE_NAME,
|
|
value=pb_token,
|
|
httponly=True,
|
|
samesite=_SAMESITE,
|
|
max_age=_COOKIE_MAX_AGE,
|
|
secure=_SECURE,
|
|
)
|
|
logger.info(f"User '{payload.username}' logged in via PocketBase (role={payload.role!r})")
|
|
return {
|
|
"ok": True,
|
|
"user_id": payload.user_id,
|
|
"username": payload.username,
|
|
"role": payload.role,
|
|
"is_admin": payload.role == "admin",
|
|
}
|
|
|
|
# Standard JWT + bcrypt mode
|
|
result = authenticate(body.username, body.password)
|
|
if not result:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Incorrect username or password",
|
|
)
|
|
|
|
token = create_token(result.username, result.role, result.user_id)
|
|
response.set_cookie(
|
|
key=_COOKIE_NAME,
|
|
value=token,
|
|
httponly=True,
|
|
samesite=_SAMESITE,
|
|
max_age=_COOKIE_MAX_AGE,
|
|
secure=_SECURE,
|
|
)
|
|
|
|
logger.info(f"User '{result.username}' logged in (role={result.role!r})")
|
|
return {
|
|
"ok": True,
|
|
"user_id": result.user_id,
|
|
"username": result.username,
|
|
"role": result.role,
|
|
"is_admin": result.role == "admin",
|
|
}
|
|
|
|
|
|
@router.post("/logout")
|
|
async def logout(response: Response) -> dict:
|
|
"""Clear the JWT cookie."""
|
|
response.delete_cookie(key=_COOKIE_NAME, samesite=_SAMESITE)
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/register", status_code=status.HTTP_201_CREATED)
|
|
async def register(body: RegisterRequest) -> dict:
|
|
"""
|
|
Bootstrap-only registration.
|
|
|
|
Public endpoint that creates the *first* admin account when the user store
|
|
is empty. Once an admin exists, this endpoint is closed; further accounts
|
|
must be created by an admin via ``POST /api/v1/auth/users``.
|
|
|
|
Only available when AUTH_ENABLED=true.
|
|
"""
|
|
if not AUTH_ENABLED:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Auth is disabled — registration is not available.",
|
|
)
|
|
|
|
if POCKETBASE_ENABLED:
|
|
# PocketBase deployments are documented as single-user. Keep registration
|
|
# closed and require admins to provision users in the PocketBase admin UI.
|
|
if not is_first_user():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Self-registration is closed. Ask an administrator to create your account.",
|
|
)
|
|
result = register_pb(username=body.username, email=body.username, password=body.password)
|
|
if not result:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Registration failed — username or email may already be taken.",
|
|
)
|
|
logger.info(f"First user registered via PocketBase: '{body.username}'")
|
|
return {
|
|
"ok": True,
|
|
"user_id": result.get("id", ""),
|
|
"username": body.username,
|
|
"role": "user",
|
|
"is_first_user": True,
|
|
"is_admin": False,
|
|
}
|
|
|
|
# Standard mode — only allowed before the first admin exists.
|
|
if not is_first_user():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Self-registration is closed. Ask an administrator to create your account.",
|
|
)
|
|
|
|
existing = {u["username"] for u in list_users()}
|
|
if body.username in existing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Username already taken",
|
|
)
|
|
|
|
add_user(body.username, body.password)
|
|
user_id = ""
|
|
role = "user"
|
|
for item in list_users():
|
|
if item.get("username") == body.username:
|
|
user_id = str(item.get("id") or "")
|
|
role = str(item.get("role") or "user")
|
|
break
|
|
logger.info(f"First user (admin) registered: '{body.username}'")
|
|
return {
|
|
"ok": True,
|
|
"user_id": user_id,
|
|
"username": body.username,
|
|
"role": role,
|
|
"is_first_user": True,
|
|
"is_admin": role == "admin",
|
|
}
|
|
|
|
|
|
@router.get("/is_first_user")
|
|
async def check_is_first_user() -> dict:
|
|
"""Return whether the user store is empty (used by the register UI)."""
|
|
return {"is_first_user": is_first_user() if AUTH_ENABLED else False}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Profile endpoints (any authenticated user, self-service)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_AVATAR_MAX_BYTES = 1 * 1024 * 1024
|
|
_AVATAR_MEDIA_TYPES = {"png": "image/png", "jpg": "image/jpeg", "webp": "image/webp"}
|
|
|
|
|
|
def _sniff_image(data: bytes) -> str | None:
|
|
"""Detect a supported raster image format from its magic bytes.
|
|
|
|
The uploaded filename and Content-Type are attacker-controlled, so the
|
|
stored extension (and the media type served back) is derived from the
|
|
bytes alone. SVG is deliberately unsupported — serving user-supplied SVG
|
|
is a stored-XSS vector.
|
|
"""
|
|
if data[:8] == b"\x89PNG\r\n\x1a\n":
|
|
return "png"
|
|
if data[:3] == b"\xff\xd8\xff":
|
|
return "jpg"
|
|
if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
|
|
return "webp"
|
|
return None
|
|
|
|
|
|
def _require_profile_identity(payload: TokenPayload | None) -> TokenPayload:
|
|
"""Shared guard for the self-service profile endpoints."""
|
|
if not AUTH_ENABLED or payload is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Auth is disabled — profiles are not available.",
|
|
)
|
|
return payload
|
|
|
|
|
|
@router.get("/profile", response_model=UserInfo)
|
|
async def get_profile(
|
|
payload: TokenPayload | None = Depends(require_auth),
|
|
) -> UserInfo:
|
|
"""Return the current user's own account info."""
|
|
current = _require_profile_identity(payload)
|
|
info = get_user_info(current.username)
|
|
if info is None:
|
|
# PocketBase-backed identities have no local record; fall back to the
|
|
# token claims so the profile page still renders.
|
|
return UserInfo(
|
|
id=current.user_id,
|
|
username=current.username,
|
|
role=current.role,
|
|
created_at="",
|
|
)
|
|
return UserInfo(**info)
|
|
|
|
|
|
@router.put("/profile")
|
|
async def update_profile(
|
|
body: UpdateProfileRequest,
|
|
payload: TokenPayload | None = Depends(require_auth),
|
|
) -> dict:
|
|
"""Update the current user's own avatar marker (icon choice or reset).
|
|
|
|
Only the validated ``icon:<name>:<color>`` form (or empty string) is
|
|
accepted here; ``img:`` markers are owned by the upload endpoint.
|
|
"""
|
|
current = _require_profile_identity(payload)
|
|
if not set_avatar(current.username, body.avatar):
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
# The marker no longer references an uploaded image, so drop the file.
|
|
from deeptutor.multi_user.identity import delete_avatar_file
|
|
|
|
if current.user_id and _USER_ID_RE.match(current.user_id):
|
|
delete_avatar_file(current.user_id)
|
|
return {"ok": True, "avatar": body.avatar}
|
|
|
|
|
|
@router.put("/profile/avatar")
|
|
async def upload_avatar(
|
|
file: UploadFile = File(...),
|
|
payload: TokenPayload | None = Depends(require_auth),
|
|
) -> dict:
|
|
"""Upload an avatar image for the current user.
|
|
|
|
The client is expected to crop/resize before uploading; the server only
|
|
enforces a size cap and validates the format by magic bytes. Not available
|
|
in PocketBase mode (those identities have no local user record).
|
|
"""
|
|
current = _require_profile_identity(payload)
|
|
if POCKETBASE_ENABLED:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Avatar upload is not available in PocketBase mode.",
|
|
)
|
|
if not current.user_id or not _USER_ID_RE.match(current.user_id):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Cannot store an avatar for this account.",
|
|
)
|
|
info = get_user_info(current.username)
|
|
if info is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
|
|
data = await file.read(_AVATAR_MAX_BYTES + 1)
|
|
if len(data) > _AVATAR_MAX_BYTES:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
|
detail="Avatar image is too large (max 1 MB).",
|
|
)
|
|
ext = _sniff_image(data)
|
|
if ext is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
|
|
detail="Avatar must be a PNG, JPEG or WebP image.",
|
|
)
|
|
|
|
from deeptutor.multi_user.identity import save_avatar_file
|
|
|
|
# Bump the version embedded in the marker so clients cache-bust the URL.
|
|
previous = str(info.get("avatar") or "")
|
|
version = 1
|
|
if previous.startswith("img:"):
|
|
try:
|
|
version = int(previous.split(":", 1)[1]) + 1
|
|
except ValueError:
|
|
version = 1
|
|
marker = f"img:{version}"
|
|
|
|
save_avatar_file(current.user_id, data, ext)
|
|
if not set_avatar(current.username, marker):
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
logger.info(f"User '{current.username}' uploaded a new avatar ({ext}, {len(data)} bytes)")
|
|
return {"ok": True, "avatar": marker}
|
|
|
|
|
|
@router.delete("/profile/avatar")
|
|
async def remove_avatar(
|
|
payload: TokenPayload | None = Depends(require_auth),
|
|
) -> dict:
|
|
"""Remove the current user's uploaded avatar image and reset the marker."""
|
|
current = _require_profile_identity(payload)
|
|
from deeptutor.multi_user.identity import delete_avatar_file
|
|
|
|
if current.user_id and _USER_ID_RE.match(current.user_id):
|
|
delete_avatar_file(current.user_id)
|
|
set_avatar(current.username, "")
|
|
return {"ok": True, "avatar": ""}
|
|
|
|
|
|
@router.get("/avatar/{user_id}")
|
|
async def get_avatar_image(
|
|
user_id: str,
|
|
_: TokenPayload | None = Depends(require_auth),
|
|
) -> FileResponse:
|
|
"""Serve a stored avatar image. Any authenticated user may view avatars
|
|
(they appear in the admin table and next to the viewer's own profile)."""
|
|
if not _USER_ID_RE.match(user_id):
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Avatar not found")
|
|
|
|
from deeptutor.multi_user.identity import get_avatar_file
|
|
|
|
target = get_avatar_file(user_id)
|
|
if target is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Avatar not found")
|
|
|
|
media_type = _AVATAR_MEDIA_TYPES.get(target.suffix.lstrip("."), "application/octet-stream")
|
|
headers = {
|
|
# Private user content; the marker version in the URL handles busting.
|
|
"Cache-Control": "private, max-age=86400",
|
|
"X-Content-Type-Options": "nosniff",
|
|
"Content-Disposition": "inline",
|
|
}
|
|
return FileResponse(path=str(target), media_type=media_type, headers=headers)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Admin-only endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/users", response_model=list[UserInfo])
|
|
async def get_users(_: TokenPayload = Depends(require_admin)) -> list[UserInfo]:
|
|
"""List all registered users. Requires admin role."""
|
|
return [UserInfo(**u) for u in list_users()]
|
|
|
|
|
|
@router.post("/users", status_code=status.HTTP_201_CREATED)
|
|
async def admin_create_user(
|
|
body: RegisterRequest,
|
|
current: TokenPayload = Depends(require_admin),
|
|
) -> dict:
|
|
"""Admin-only: create a new user account.
|
|
|
|
Replaces the public ``/register`` flow once the first admin exists. The
|
|
new account is always created with role=``user``; admins can promote
|
|
later via ``PUT /users/{username}/role``.
|
|
"""
|
|
if not AUTH_ENABLED:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Auth is disabled — user creation is not available.",
|
|
)
|
|
|
|
if POCKETBASE_ENABLED:
|
|
result = register_pb(username=body.username, email=body.username, password=body.password)
|
|
if not result:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Failed to create user — username may already be taken.",
|
|
)
|
|
logger.info(
|
|
f"Admin '{current.username if current else 'local'}' created PocketBase user "
|
|
f"'{body.username}'"
|
|
)
|
|
return {
|
|
"ok": True,
|
|
"user_id": result.get("id", ""),
|
|
"username": body.username,
|
|
"role": "user",
|
|
"is_admin": False,
|
|
}
|
|
|
|
existing = {u["username"] for u in list_users()}
|
|
if body.username in existing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Username already taken",
|
|
)
|
|
|
|
add_user(body.username, body.password)
|
|
user_id = ""
|
|
role = "user"
|
|
for item in list_users():
|
|
if item.get("username") == body.username:
|
|
user_id = str(item.get("id") or "")
|
|
role = str(item.get("role") or "user")
|
|
break
|
|
logger.info(
|
|
f"Admin '{current.username if current else 'local'}' created user '{body.username}' "
|
|
f"(role={role!r})"
|
|
)
|
|
return {
|
|
"ok": True,
|
|
"user_id": user_id,
|
|
"username": body.username,
|
|
"role": role,
|
|
"is_admin": role == "admin",
|
|
}
|
|
|
|
|
|
@router.delete("/users/{username}", status_code=status.HTTP_200_OK)
|
|
async def remove_user(
|
|
username: str,
|
|
current: TokenPayload = Depends(require_admin),
|
|
) -> dict:
|
|
"""Delete a user. Admins cannot delete their own account."""
|
|
if current and username == current.username:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="You cannot delete your own account",
|
|
)
|
|
|
|
# Capture the id before the record disappears so the avatar file can go too.
|
|
info = get_user_info(username)
|
|
|
|
removed = delete_user(username)
|
|
if not removed:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
|
|
user_id = str(info.get("id") or "") if info else ""
|
|
if user_id and _USER_ID_RE.match(user_id):
|
|
from deeptutor.multi_user.identity import delete_avatar_file
|
|
|
|
delete_avatar_file(user_id)
|
|
|
|
logger.info(f"Admin '{current.username if current else 'local'}' deleted user '{username}'")
|
|
return {"ok": True}
|
|
|
|
|
|
@router.put("/users/{username}/role", status_code=status.HTTP_200_OK)
|
|
async def update_user_role(
|
|
username: str,
|
|
body: SetRoleRequest,
|
|
current: TokenPayload = Depends(require_admin),
|
|
) -> dict:
|
|
"""Change a user's role. Admins cannot change their own role."""
|
|
if current and username == current.username:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="You cannot change your own role",
|
|
)
|
|
|
|
updated = set_role(username, body.role)
|
|
if not updated:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
|
|
logger.info(
|
|
f"Admin '{current.username if current else 'local'}' set '{username}' role to {body.role!r}"
|
|
)
|
|
return {"ok": True, "username": username, "role": body.role}
|