fix: harden authentication system (JWT, cookies, rate limiting, password policy)

- Replace hardcoded JWT secret with randomly generated key persisted to file
- Replace hardcoded default password with random password shown in logs
- Migrate token storage from localStorage to HttpOnly SameSite=strict cookie
- Add IP-based login rate limiter (5 attempts / 15 min, 429 on lockout)
- Add token_version for JWT revocation on password change
- Add password strength validation (min 6 chars, 3+ unique characters)
- Inject decoded user payload into request.state.user in auth middleware
- Add /api/auth/me and /api/auth/logout endpoints
- Narrow auth middleware exception handling (JWTError only, not all Exception)
- Update updated_at timestamp on password change
- Remove localStorage token management from frontend (axios, router, store)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

api/app/config.py

@@ -1,5 +1,9 @@
 # 硬编码配置
 import os
+import secrets
+import logging
+
+_logger = logging.getLogger("app.config")
 
 # api 目录的绝对路径（基于本文件位置计算，不依赖工作目录）
 _BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
@@ -28,6 +32,20 @@ DEFAULT_PAGE_SIZE = 20
 HOST = "0.0.0.0"
 PORT = 23994
 
-# JWT 认证配置
-JWT_SECRET = "elysia-todo-secret-key-change-in-production"
+
+# JWT 密钥（首次启动随机生成，持久化到文件）
+def _load_jwt_secret() -> str:
+    secret_file = os.path.join(_BASE_DIR, "data", ".jwt_secret")
+    if os.path.exists(secret_file):
+        with open(secret_file) as f:
+            return f.read().strip()
+    secret = secrets.token_hex(32)
+    os.makedirs(os.path.dirname(secret_file), exist_ok=True)
+    with open(secret_file, "w") as f:
+        f.write(secret)
+    _logger.warning("已生成新的 JWT 密钥")
+    return secret
+
+
+JWT_SECRET = _load_jwt_secret()
 ACCESS_TOKEN_EXPIRE_MINUTES = 1440  # 24小时

api/app/main.py

@@ -7,10 +7,12 @@ import time
 import json
 
 from app.config import CORS_ORIGINS, WEBUI_PATH, HOST, PORT
-from app.database import init_db
+from app.database import init_db, SessionLocal
+from app.models.user_settings import UserSettings
 from app.routers import api_router
 from app.utils.logger import logger
 from app.utils.auth import decode_access_token
+from jose import JWTError
 
 
 @asynccontextmanager
@@ -90,25 +92,35 @@ async def log_requests(request: Request, call_next):
     return response
 
 
-# 认证中间件（保护所有 /api/* 路由，除了 /api/auth/* 和 /health）
+# 认证中间件（保护 /api/*，仅放行 /health 和 /api/auth/login、/api/auth/logout）
 @app.middleware("http")
 async def auth_middleware(request: Request, call_next):
     path = request.url.path
 
-    # 不拦截：健康检查、静态文件、auth 路由
-    if path == "/health" or not path.startswith("/api/") or path.startswith("/api/auth/"):
+    # 不拦截：健康检查、静态文件、公开的 auth 端点
+    public_paths = {"/health", "/api/auth/login", "/api/auth/logout"}
+    if path in public_paths or not path.startswith("/api/"):
         return await call_next(request)
 
-    auth_header = request.headers.get("Authorization", "")
-    token = auth_header.replace("Bearer ", "")
+    token = request.cookies.get("access_token", "")
     if not token:
         return JSONResponse(status_code=401, content={"detail": "未登录"})
 
     try:
-        decode_access_token(token)
-    except Exception:
+        payload = decode_access_token(token)
+    except JWTError:
         return JSONResponse(status_code=401, content={"detail": "登录已过期，请重新登录"})
 
+    db = SessionLocal()
+    try:
+        settings = db.query(UserSettings).filter(UserSettings.id == 1).first()
+        if settings and payload.get("tv") != settings.token_version:
+            return JSONResponse(status_code=401, content={"detail": "密码已修改，请重新登录"})
+    finally:
+        db.close()
+
+    request.state.user = payload
+
     return await call_next(request)
 
 

api/app/models/user_settings.py

@@ -27,6 +27,7 @@ class UserSettings(Base):
 
     # 认证
     password_hash = Column(String(255), default="")
+    token_version = Column(Integer, default=0)
 
     # 时间戳
     created_at = Column(DateTime, default=utcnow)

api/app/routers/auth.py

@@ -1,19 +1,42 @@
 from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi.responses import JSONResponse
 from sqlalchemy.orm import Session
 
 from app.database import get_db
 from app.models.user_settings import UserSettings
-from app.schemas.auth import LoginRequest, TokenResponse, ChangePasswordRequest
+from app.schemas.auth import LoginRequest, LoginResponse, ChangePasswordRequest, AuthStatusResponse
 from app.utils.auth import (
     hash_password, verify_password, create_access_token,
     get_current_user, set_default_password
 )
+from app.utils.datetime import utcnow
+from app.utils.rate_limiter import login_limiter
 
 router = APIRouter(prefix="/api/auth", tags=["认证"])
 
 
-@router.post("/login", response_model=TokenResponse)
-def login(data: LoginRequest, db: Session = Depends(get_db)):
+def _get_client_ip(request: Request) -> str:
+    forwarded = request.headers.get("X-Forwarded-For")
+    if forwarded:
+        return forwarded.split(",")[0].strip()
+    return request.client.host if request.client else "unknown"
+
+
+@router.post("/login", response_model=LoginResponse)
+def login(data: LoginRequest, request: Request, db: Session = Depends(get_db)):
+    ip = _get_client_ip(request)
+
+    allowed, retry_after = login_limiter.check(ip)
+    if not allowed:
+        return JSONResponse(
+            status_code=429,
+            content={
+                "detail": f"登录尝试过于频繁，请 {retry_after // 60 + 1} 分钟后再试",
+                "retry_after": retry_after,
+            },
+            headers={"Retry-After": str(retry_after)},
+        )
+
     settings = db.query(UserSettings).filter(UserSettings.id == 1).first()
     if not settings:
         settings = UserSettings(id=1)
@@ -24,10 +47,39 @@ def login(data: LoginRequest, db: Session = Depends(get_db)):
     set_default_password(db, settings)
 
     if not verify_password(data.password, settings.password_hash):
+        login_limiter.record_failure(ip)
         raise HTTPException(status_code=401, detail="密码错误")
 
-    token = create_access_token({"sub": str(settings.id)})
-    return TokenResponse(access_token=token)
+    login_limiter.reset(ip)
+
+    token = create_access_token({
+        "sub": str(settings.id),
+        "tv": settings.token_version,
+    })
+
+    response = JSONResponse(content={"message": "登录成功"})
+    response.set_cookie(
+        key="access_token",
+        value=token,
+        httponly=True,
+        samesite="strict",
+        max_age=86400,
+        path="/",
+    )
+    return response
+
+
+@router.post("/logout")
+def logout():
+    response = JSONResponse(content={"message": "已退出登录"})
+    response.delete_cookie("access_token", path="/")
+    return response
+
+
+@router.get("/me", response_model=AuthStatusResponse)
+def me(request: Request):
+    user = get_current_user(request)
+    return AuthStatusResponse(authenticated=True, user_id=user.get("sub", ""))
 
 
 @router.post("/change-password")
@@ -36,7 +88,7 @@ def change_password(
     request: Request,
     db: Session = Depends(get_db)
 ):
-    get_current_user(request)
+    user = get_current_user(request)
 
     settings = db.query(UserSettings).filter(UserSettings.id == 1).first()
     if not settings:
@@ -46,6 +98,8 @@ def change_password(
         raise HTTPException(status_code=400, detail="原密码错误")
 
     settings.password_hash = hash_password(data.new_password)
+    settings.token_version = (settings.token_version or 0) + 1
+    settings.updated_at = utcnow()
     db.commit()
 
     return {"message": "密码修改成功"}

api/app/schemas/auth.py

@@ -1,15 +1,28 @@
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 
 
 class LoginRequest(BaseModel):
     password: str = Field(..., min_length=1, max_length=100)
 
 
-class TokenResponse(BaseModel):
-    access_token: str
-    token_type: str = "bearer"
+class LoginResponse(BaseModel):
+    message: str = "登录成功"
 
 
 class ChangePasswordRequest(BaseModel):
     old_password: str = Field(..., min_length=1, max_length=100)
-    new_password: str = Field(..., min_length=1, max_length=100)
+    new_password: str = Field(..., min_length=6, max_length=100)
+
+    @field_validator("new_password")
+    @classmethod
+    def validate_password_strength(cls, v: str) -> str:
+        if len(v) < 6:
+            raise ValueError("密码长度至少6位")
+        if len(set(v)) < 3:
+            raise ValueError("密码不能过于简单（需包含至少3种不同字符）")
+        return v
+
+
+class AuthStatusResponse(BaseModel):
+    authenticated: bool
+    user_id: str

api/app/utils/auth.py

@@ -1,13 +1,15 @@
 from datetime import datetime, timedelta, timezone
 from typing import Optional
+import secrets
+
 from jose import JWTError, jwt
 from passlib.context import CryptContext
 from fastapi import Request, HTTPException
 from sqlalchemy.orm import Session
 
 from app.config import JWT_SECRET, ACCESS_TOKEN_EXPIRE_MINUTES
-from app.database import get_db
 from app.models.user_settings import UserSettings
+from app.utils.logger import logger
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
@@ -34,8 +36,7 @@ def decode_access_token(token: str) -> dict:
 
 
 def get_current_user(request: Request) -> dict:
-    auth_header = request.headers.get("Authorization", "")
-    token = auth_header.replace("Bearer ", "")
+    token = request.cookies.get("access_token", "")
     if not token:
         raise HTTPException(status_code=401, detail="未登录")
     try:
@@ -47,5 +48,10 @@ def get_current_user(request: Request) -> dict:
 
 def set_default_password(db: Session, settings: UserSettings):
     if not settings.password_hash:
-        settings.password_hash = hash_password("elysia")
+        password = secrets.token_urlsafe(8)[:8]
+        settings.password_hash = hash_password(password)
         db.commit()
+        logger.warning("=" * 50)
+        logger.warning(f"  初始密码: {password}")
+        logger.warning("  请登录后立即修改！")
+        logger.warning("=" * 50)

api/app/utils/rate_limiter.py

@@ -0,0 +1,59 @@
+"""
+登录频率限制器（内存实现，单用户场景适用）
+"""
+import time
+
+
+class LoginRateLimiter:
+    """IP 级别的登录频率限制"""
+
+    MAX_ATTEMPTS = 5
+    WINDOW_SECONDS = 900       # 统计窗口：15 分钟
+    LOCKOUT_SECONDS = 900      # 锁定时间：15 分钟
+
+    def __init__(self):
+        self._attempts: dict[str, list[float]] = {}
+        self._lockout: dict[str, float] = {}
+
+    def _cleanup(self):
+        """清除过期的记录"""
+        now = time.time()
+        self._attempts = {
+            ip: [t for t in times if now - t < self.WINDOW_SECONDS]
+            for ip, times in self._attempts.items()
+        }
+        self._attempts = {ip: times for ip, times in self._attempts.items() if times}
+
+        self._lockout = {
+            ip: until for ip, until in self._lockout.items()
+            if now < until
+        }
+
+    def check(self, ip: str) -> tuple[bool, int | None]:
+        """返回 (是否允许, 剩余等待秒数)"""
+        self._cleanup()
+
+        now = time.time()
+
+        if ip in self._lockout:
+            remaining = int(self._lockout[ip] - now)
+            if remaining > 0:
+                return False, remaining
+            del self._lockout[ip]
+
+        return True, None
+
+    def record_failure(self, ip: str):
+        """记录一次失败"""
+        now = time.time()
+        self._attempts.setdefault(ip, []).append(now)
+        if len(self._attempts[ip]) >= self.MAX_ATTEMPTS:
+            self._lockout[ip] = now + self.LOCKOUT_SECONDS
+
+    def reset(self, ip: str):
+        """登录成功后重置计数"""
+        self._attempts.pop(ip, None)
+        self._lockout.pop(ip, None)
+
+
+login_limiter = LoginRateLimiter()