安全补强: 6项P0/P1漏洞修复

P0-1: PHP配置注入防护 — install.py添加_escape_php_string()转义单引号和反斜杠
P0-2: WebSocket JWT校验 — _verify_ws_token()要求exp+sub字段
P1-1: 删除SHA256密码fallback — auth_service.py和auth.py直接import bcrypt
P1-3: LIKE通配符转义 — search.py添加_escape_like()并对所有ilike()加escape参数
P1-2: 安全响应头中间件 — main.py添加SecurityHeadersMiddleware注入4个安全头
P0-3: Refresh Token重用检测 — Admin模型添加token_version字段,token格式改为token:admin_id:version

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Your Name
2026-05-22 22:16:50 +08:00
parent cea5730804
commit 8530f0e0d5
8 changed files with 313 additions and 77 deletions
+46 -1
View File
@@ -8,11 +8,12 @@ from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession
import bcrypt
from server.api.dependencies import get_auth_service, get_db
from server.api.auth_jwt import get_current_admin
from server.application.services.auth_service import AuthService
from server.domain.models import Admin
from server.domain.models import Admin, AuditLog
router = APIRouter(prefix="/api/auth", tags=["auth"])
@@ -42,6 +43,11 @@ class TotpVerifyRequest(BaseModel):
totp_code: str = Field(..., min_length=6, max_length=6)
class ChangePasswordRequest(BaseModel):
current_password: str = Field(..., min_length=1, max_length=255)
new_password: str = Field(..., min_length=6, max_length=255)
# ── Public Routes (no JWT required) ──
@router.post("/login")
@@ -151,6 +157,45 @@ async def disable_totp(
return result
@router.put("/password")
async def change_password(
payload: ChangePasswordRequest,
admin: Admin = Depends(get_current_admin),
db: AsyncSession = Depends(get_db),
):
"""Change the current admin's password (requires current password verification)"""
from server.infrastructure.database.admin_repo import AdminRepositoryImpl
from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
admin_repo = AdminRepositoryImpl(db)
current_admin = await admin_repo.get_by_id(admin.id)
if not current_admin:
raise HTTPException(status_code=404, detail="Admin not found")
# Verify current password
if not bcrypt.checkpw(payload.current_password.encode(), current_admin.password_hash.encode()):
raise HTTPException(status_code=400, detail="当前密码错误")
# Hash new password
new_hash = bcrypt.hashpw(payload.new_password.encode(), bcrypt.gensalt()).decode()
current_admin.password_hash = new_hash
await admin_repo.update(current_admin)
# Audit log
audit_repo = AuditLogRepositoryImpl(db)
await audit_repo.create(AuditLog(
admin_username=admin.username,
action="change_password",
target_type="admin",
target_id=admin.id,
detail=f"Password changed for {admin.username}",
ip_address="",
))
return {"success": True, "message": "密码已修改"}
@router.get("/me")
async def get_me(
admin: Admin = Depends(get_current_admin),