浅念
6880ff1b97
后端: - POST /api/auth/login:admin 账密(Settings 默认 admin/admin123, ADMIN_USER/ADMIN_PASSWORD 可覆盖)换取共享 api_token,免 token 中间件 - 防爆破:同 IP 15min 窗口内 5 次失败锁 15min(429),成功登录清零 - 错误信息统一「用户名或密码错误」(防用户名枚举) - 测试 7 个新增(TDD),全量 500 绿 前端: - LoginView + 路由守卫(未登录跳 /login,回跳 redirect) - axios 请求拦截器自动带 Bearer;401 清登录态回登录页 - 媒体直链统一 mediaUrl() 带 ?token=(img/audio 标签发不了 header), 替换 4 处裸拼;downloadBlob 手动带头 - isLoggedIn 按 key 存在判定 — 后端鉴权关闭时空 token 不锁死登录页 - MainLayout 侧栏底部退出登录 compose:两份编排透传 ADMIN_USER/ADMIN_PASSWORD(默认 admin/admin123) Co-Authored-By: Claude Code <noreply@anthropic.com>
107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
"""Simple shared-token auth for the yunke data-mirror API (Task 2.9).
|
|
|
|
Trust model (per design doc §8 + plan Task 2.9):
|
|
|
|
* **No token configured** (``Settings.api_token`` empty, the default) -> auth
|
|
is DISABLED. This is a local/intranet tool; when no token is set we trust the
|
|
network boundary and every request passes through. This keeps local dev and
|
|
single-box deployments zero-config.
|
|
* **Token configured** -> every request must present the same token via either
|
|
the ``Authorization: Bearer <token>`` header or the ``?token=<token>`` query
|
|
parameter. Missing / mismatched tokens get ``401 Unauthorized``.
|
|
|
|
The dependency is applied globally on the FastAPI app (see
|
|
:mod:`app.api.main`). ``GET /api/health`` is exempt so liveness probes can
|
|
reach it without knowing the token — that exemption is encoded here in
|
|
:data:`_PUBLIC_PATHS` so the rule lives with the auth logic itself.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
import time
|
|
|
|
from fastapi import HTTPException, Request, status
|
|
|
|
from app.config import get_settings
|
|
|
|
# Request paths that bypass auth even when a token is configured. Keep this
|
|
# minimal so the protected surface is easy to reason about — the liveness
|
|
# probe and the login endpoint itself (the caller has no token yet).
|
|
_PUBLIC_PATHS = frozenset({"/api/health", "/api/auth/login"})
|
|
|
|
# ---- Brute-force guard for POST /api/auth/login ------------------------- #
|
|
# In-memory per-IP failure tracker: MAX failures within WINDOW seconds locks
|
|
# the IP out for LOCKOUT seconds. Process-local by design (single-backend
|
|
# deployment; a restart simply clears the slate).
|
|
_LOGIN_MAX_FAILURES = 5
|
|
_LOGIN_FAILURE_WINDOW_S = 15 * 60
|
|
_LOGIN_LOCKOUT_S = 15 * 60
|
|
# ip -> list of monotonic timestamps of recent failed logins
|
|
_login_failures: dict[str, list[float]] = {}
|
|
|
|
|
|
def _client_ip(request: Request) -> str:
|
|
"""Best-effort client IP for the lockout key (test transport has none)."""
|
|
return request.client.host if request.client else "unknown"
|
|
|
|
|
|
def _login_locked_out(ip: str) -> bool:
|
|
"""True when ``ip`` burned through MAX failures inside the window."""
|
|
now = time.monotonic()
|
|
stamps = [t for t in _login_failures.get(ip, []) if now - t < _LOGIN_FAILURE_WINDOW_S]
|
|
_login_failures[ip] = stamps
|
|
return len(stamps) >= _LOGIN_MAX_FAILURES
|
|
|
|
|
|
def _record_login_failure(ip: str) -> None:
|
|
_login_failures.setdefault(ip, []).append(time.monotonic())
|
|
|
|
|
|
def _reset_login_failures(ip: str) -> None:
|
|
_login_failures.pop(ip, None)
|
|
|
|
|
|
async def require_auth(request: Request) -> None:
|
|
"""No-op when ``api_token`` is unset; otherwise enforce a shared bearer token.
|
|
|
|
Settings are read on every call (``get_settings`` is ``lru_cache``-ed, so
|
|
this is cheap) so rotating the token — env change plus
|
|
``get_settings.cache_clear()`` — takes effect without a process restart.
|
|
The comparison uses :func:`secrets.compare_digest` on UTF-8 bytes to avoid
|
|
a timing side channel and to tolerate non-ASCII token values.
|
|
"""
|
|
settings = get_settings()
|
|
expected = settings.api_token
|
|
if not expected:
|
|
# No token configured -> trust the network boundary.
|
|
return
|
|
|
|
if request.url.path in _PUBLIC_PATHS:
|
|
return
|
|
|
|
presented = _extract_token(request)
|
|
if presented is None or not secrets.compare_digest(
|
|
presented.encode("utf-8"), expected.encode("utf-8")
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="invalid or missing API token",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
|
|
def _extract_token(request: Request) -> str | None:
|
|
"""Pull the presented token from the ``Authorization`` header or ``?token=``.
|
|
|
|
Header wins when present and well-formed (``Bearer <value>``); otherwise we
|
|
fall back to the ``token`` query parameter, which is convenient for
|
|
browser-visible endpoints like media streaming.
|
|
"""
|
|
auth_header = request.headers.get("Authorization")
|
|
if auth_header:
|
|
scheme, _, value = auth_header.partition(" ")
|
|
if scheme.lower() == "bearer" and value:
|
|
return value
|
|
return request.query_params.get("token")
|