fish
fc69f51e23
单密码→多用户：auth.json 格式迁移，login 支持用户名密码， 新增用户管理 API 和前端页面，会话关联 username， auth middleware 将当前用户注入 request.state。 Co-Authored-By: Claude <noreply@anthropic.com>
292 lines
10 KiB
Python
292 lines
10 KiB
Python
"""访问认证 API。
|
|
|
|
端点:
|
|
GET /api/auth/status — 认证状态 + 当前用户信息
|
|
POST /api/auth/setup — 首次设置密码(仅限本机/内网, 防公网抢占)
|
|
POST /api/auth/login — 登录(用户名+密码 → 会话 token, 含限流)
|
|
POST /api/auth/logout — 注销当前会话
|
|
POST /api/auth/change-password — 改密码(需已登录)
|
|
GET /api/auth/users — 用户列表(需 admin)
|
|
POST /api/auth/users — 创建用户(需 admin)
|
|
DELETE /api/auth/users/{username} — 删除用户(需 admin)
|
|
POST /api/auth/users/{username}/reset-password — 重置密码(需 admin)
|
|
|
|
安全:
|
|
- setup 端点只接受本机/内网请求(request.client.host), 公网请求 403。
|
|
否则黑客可比用户更早扫到域名, 抢先设密码, 反客为主。
|
|
- login 限流: 同一来源 IP 连续失败 5 次, 锁 5 分钟(内存计数)。
|
|
- 会话 token 通过 HttpOnly cookie 下发, 前端无需手动管理。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from collections import defaultdict
|
|
from threading import Lock
|
|
|
|
from fastapi import APIRouter, HTTPException, Request, Response
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.services import auth
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
COOKIE_NAME = "tf_session"
|
|
_COOKIE_MAX_AGE = 30 * 24 * 3600 # 与 SESSION_TTL 一致
|
|
|
|
# 限流: { ip: (fail_count, lock_until_ts) }
|
|
_fail_counter: dict[str, tuple[int, float]] = defaultdict(lambda: (0, 0.0))
|
|
_fail_lock = Lock()
|
|
_MAX_FAILS = 5
|
|
_LOCK_SECONDS = 300
|
|
|
|
|
|
def _is_local_network(host: str | None) -> bool:
|
|
"""是否本机或内网请求。
|
|
|
|
反向代理(Nginx)场景下 request.client.host 是代理本身(127.0.0.1),
|
|
需信任 X-Forwarded-For 的最左(原始客户端)。本项目部署若经反代,
|
|
请在反代配置正确的 X-Forwarded-For(标准做法)。
|
|
"""
|
|
if not host:
|
|
return False
|
|
if host in ("127.0.0.1", "::1", "localhost"):
|
|
return True
|
|
# 内网网段: 10.x / 172.16-31.x / 192.168.x
|
|
if host.startswith("10.") or host.startswith("192.168."):
|
|
return True
|
|
if host.startswith("172."):
|
|
try:
|
|
second = int(host.split(".")[1])
|
|
if 16 <= second <= 31:
|
|
return True
|
|
except (IndexError, ValueError):
|
|
pass
|
|
return False
|
|
|
|
|
|
def _client_ip(request: Request) -> str:
|
|
"""取真实客户端 IP(信任反代 X-Forwarded-For)。"""
|
|
xff = request.headers.get("x-forwarded-for")
|
|
if xff:
|
|
return xff.split(",")[0].strip()
|
|
return request.client.host if request.client else "unknown"
|
|
|
|
|
|
def _check_login_rate_limit(ip: str) -> None:
|
|
"""登录失败限流检查, 触发则抛 429。"""
|
|
with _fail_lock:
|
|
count, until = _fail_counter.get(ip, (0, 0.0))
|
|
now = time.time()
|
|
if until > now:
|
|
wait = int(until - now)
|
|
raise HTTPException(
|
|
status_code=429,
|
|
detail=f"登录失败次数过多, 请 {wait} 秒后重试",
|
|
)
|
|
|
|
|
|
def _record_login_fail(ip: str) -> None:
|
|
"""记录一次登录失败, 达阈值则锁定。"""
|
|
with _fail_lock:
|
|
count, until = _fail_counter.get(ip, (0, 0.0))
|
|
count += 1
|
|
if count >= _MAX_FAILS:
|
|
until = time.time() + _LOCK_SECONDS
|
|
logger.warning("auth login locked for %s after %d fails", ip, count)
|
|
_fail_counter[ip] = (count, until)
|
|
|
|
|
|
def _clear_login_fails(ip: str) -> None:
|
|
"""登录成功后清除该 IP 的失败计数。"""
|
|
with _fail_lock:
|
|
_fail_counter.pop(ip, None)
|
|
|
|
|
|
def _get_current_user(request: Request) -> str:
|
|
"""从 cookie 获取当前登录的 username, 未登录抛 401。"""
|
|
token = request.cookies.get(COOKIE_NAME)
|
|
username = auth.get_session_user(token or "")
|
|
if not username:
|
|
raise HTTPException(status_code=401, detail="请先登录")
|
|
return username
|
|
|
|
|
|
def _require_admin(request: Request) -> str:
|
|
"""要求当前用户为 admin, 否则抛 403。"""
|
|
username = _get_current_user(request)
|
|
token = request.cookies.get(COOKIE_NAME, "")
|
|
role = auth.get_session_role(token)
|
|
if role != "admin":
|
|
raise HTTPException(status_code=403, detail="需要管理员权限")
|
|
return username
|
|
|
|
|
|
# ================================================================
|
|
# 端点
|
|
# ================================================================
|
|
|
|
class PasswordIn(BaseModel):
|
|
password: str = Field(min_length=6, max_length=128)
|
|
|
|
|
|
class LoginIn(BaseModel):
|
|
username: str = Field(default="admin", min_length=1, max_length=64, description="用户名, 默认 admin")
|
|
password: str = Field(min_length=1, max_length=128)
|
|
|
|
|
|
class ChangePasswordIn(BaseModel):
|
|
old_password: str = Field(min_length=1, max_length=128)
|
|
new_password: str = Field(min_length=6, max_length=128)
|
|
|
|
|
|
class CreateUserIn(BaseModel):
|
|
username: str = Field(min_length=1, max_length=64, pattern=r"^[a-zA-Z0-9_-]+$")
|
|
password: str = Field(min_length=6, max_length=128)
|
|
role: str = Field(default="viewer", pattern=r"^(admin|viewer)$")
|
|
|
|
|
|
class ResetPasswordIn(BaseModel):
|
|
new_password: str = Field(min_length=6, max_length=128)
|
|
|
|
|
|
@router.get("/status")
|
|
def auth_status(request: Request) -> dict:
|
|
"""认证状态: 是否已设密码 + 当前请求是否已登录 + 用户信息。"""
|
|
token = request.cookies.get(COOKIE_NAME) or ""
|
|
username = auth.get_session_user(token)
|
|
role = auth.get_session_role(token)
|
|
return {
|
|
"configured": auth.is_configured(),
|
|
"authenticated": bool(username),
|
|
"username": username or None,
|
|
"role": role or None,
|
|
}
|
|
|
|
|
|
@router.post("/setup")
|
|
def setup_password(req: PasswordIn, request: Request) -> dict:
|
|
"""首次设置访问密码。仅限本机/内网请求(防公网抢占)。
|
|
|
|
若已设置过密码, 返回 409(改密码走 /change-password)。
|
|
"""
|
|
# 关键: 限制只有服务器主人(本机/内网)能设密码
|
|
client_ip = _client_ip(request)
|
|
if not _is_local_network(client_ip):
|
|
logger.warning("setup rejected from non-local ip: %s", client_ip)
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="首次设置密码仅允许本机或内网访问,请通过 SSH/本地浏览器操作",
|
|
)
|
|
|
|
if auth.is_configured():
|
|
raise HTTPException(status_code=409, detail="密码已设置,如需修改请登录后使用改密码功能")
|
|
|
|
auth.set_password(req.password)
|
|
logger.info("access password set up from %s", client_ip)
|
|
return {"ok": True, "configured": True}
|
|
|
|
|
|
@router.post("/login")
|
|
def login(req: LoginIn, request: Request, response: Response) -> dict:
|
|
"""登录: 用户名+密码 → 会话 token(写 HttpOnly cookie)。含失败限流。"""
|
|
ip = _client_ip(request)
|
|
_check_login_rate_limit(ip)
|
|
|
|
if not auth.is_configured():
|
|
raise HTTPException(status_code=409, detail="尚未设置访问密码")
|
|
|
|
result = auth.login(req.username, req.password)
|
|
if not result:
|
|
_record_login_fail(ip)
|
|
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
|
|
|
token, username = result
|
|
_clear_login_fails(ip)
|
|
# HttpOnly: 防 XSS 窃取; SameSite=Lax: 防 CSRF; Path=/: 全站生效
|
|
response.set_cookie(
|
|
key=COOKIE_NAME,
|
|
value=token,
|
|
max_age=_COOKIE_MAX_AGE,
|
|
httponly=True,
|
|
samesite="lax",
|
|
path="/",
|
|
secure=False, # 自托管可能无 HTTPS, 不强制 secure(建议反代加 HTTPS)
|
|
)
|
|
return {"ok": True, "authenticated": True, "username": username}
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout(request: Request, response: Response) -> dict:
|
|
"""注销当前会话。"""
|
|
token = request.cookies.get(COOKIE_NAME)
|
|
if token:
|
|
auth.revoke_session(token)
|
|
response.delete_cookie(key=COOKIE_NAME, path="/")
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/change-password")
|
|
def change_password(req: ChangePasswordIn, request: Request) -> dict:
|
|
"""修改密码: 需验证旧密码。"""
|
|
token = request.cookies.get(COOKIE_NAME)
|
|
username = auth.get_session_user(token or "")
|
|
if not username:
|
|
raise HTTPException(status_code=401, detail="请先登录")
|
|
|
|
if not auth.change_password(username, req.old_password, req.new_password):
|
|
raise HTTPException(status_code=401, detail="旧密码错误")
|
|
|
|
return {"ok": True, "message": "密码已修改"}
|
|
|
|
|
|
# ================================================================
|
|
# 用户管理 (admin only)
|
|
# ================================================================
|
|
|
|
@router.get("/users")
|
|
def list_users(request: Request) -> dict:
|
|
"""用户列表(需 admin)。"""
|
|
_require_admin(request)
|
|
users = auth.list_users()
|
|
return {"users": users}
|
|
|
|
|
|
@router.post("/users")
|
|
def create_user(req: CreateUserIn, request: Request) -> dict:
|
|
"""创建用户(需 admin)。"""
|
|
_require_admin(request)
|
|
try:
|
|
auth.create_user(req.username, req.password, role=req.role)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e)) from e
|
|
return {"ok": True, "username": req.username}
|
|
|
|
|
|
@router.delete("/users/{username}")
|
|
def delete_user(username: str, request: Request) -> dict:
|
|
"""删除用户(需 admin, 不能删除 admin 自身)。"""
|
|
_require_admin(request)
|
|
try:
|
|
ok = auth.delete_user(username)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e)) from e
|
|
if not ok:
|
|
raise HTTPException(status_code=404, detail=f"用户 '{username}' 不存在")
|
|
return {"ok": True}
|
|
|
|
|
|
@router.post("/users/{username}/reset-password")
|
|
def reset_password(username: str, req: ResetPasswordIn, request: Request) -> dict:
|
|
"""重置用户密码(需 admin)。"""
|
|
_require_admin(request)
|
|
try:
|
|
ok = auth.admin_reset_password(username, req.new_password)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e)) from e
|
|
if not ok:
|
|
raise HTTPException(status_code=404, detail=f"用户 '{username}' 不存在")
|
|
return {"ok": True}
|