yml2213
6fc453017c
- JWT 记录 iat，exp=min(now+24h, iat+7d) - 已认证请求自动续签 httpOnly Cookie - 新增 ACCESS_TOKEN_IDLE_HOURS / ACCESS_TOKEN_MAX_HOURS
250 lines
7.3 KiB
Python
250 lines
7.3 KiB
Python
"""认证路由：安全登录（短信 + DCAuth v2）与旧账号密码兜底。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, Request, Response
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..database import get_db
|
|
from ..deps import get_current_user
|
|
from ..errors import ApiError
|
|
from ..models import AuditLog, DeviceChallenge, User
|
|
from ..permissions import ROLE_LABELS, get_user_permissions
|
|
from ..schemas import (
|
|
LoginCompleteRequest,
|
|
LoginPrepareRequest,
|
|
LoginPrepareResponse,
|
|
LoginRequest,
|
|
SmsCodeRequest,
|
|
SmsCodeResponse,
|
|
TokenResponse,
|
|
)
|
|
from ..security import issue_access_token, verify_password
|
|
from ..services.dc_auth_service import (
|
|
APP_ID,
|
|
CHALLENGE_PURPOSE,
|
|
consume_device_ticket,
|
|
create_login_challenge,
|
|
ensure_device_binding,
|
|
get_challenge,
|
|
mask_device_id,
|
|
normalize_api_base,
|
|
)
|
|
from ..services.login_sms_service import (
|
|
mask_phone,
|
|
send_login_sms_code,
|
|
verify_login_sms_code,
|
|
)
|
|
from ..session import clear_session_cookie, set_session_cookie
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["认证"])
|
|
|
|
|
|
def _find_user(db: Session, username_value: str) -> User | None:
|
|
"""用户名只 trim，不做大小写折叠、不限制字符集。
|
|
|
|
MySQL 默认排序规则大小写不敏感，因此查出来后还要做一次 Python 侧精确比对，
|
|
避免 ``Admin`` 命中 ``admin``。
|
|
"""
|
|
username = str(username_value or "").strip()
|
|
if not username:
|
|
return None
|
|
user = db.query(User).filter(User.username == username).first()
|
|
if user is not None and user.username != username:
|
|
return None
|
|
return user
|
|
|
|
|
|
def _authenticate(db: Session, username: str, password: str) -> User:
|
|
user = _find_user(db, username)
|
|
if user is None or not verify_password(password, user.password_hash):
|
|
raise ApiError(401, "用户名或密码错误", "invalid_credentials")
|
|
if not user.is_active or user.deleted_at is not None:
|
|
raise ApiError(403, "账号已禁用，请联系管理员", "user_disabled")
|
|
return user
|
|
|
|
|
|
def _request_origin(request: Request) -> str:
|
|
origin = str(request.headers.get("origin") or "").strip()
|
|
if origin:
|
|
return origin
|
|
return str(request.base_url).rstrip("/")
|
|
|
|
|
|
def _issue_session(
|
|
db: Session,
|
|
response: Response,
|
|
user: User,
|
|
method: str,
|
|
device_id: str = "",
|
|
) -> TokenResponse:
|
|
issued = issue_access_token({"sub": str(user.id), "role": user.role})
|
|
set_session_cookie(response, issued.token, issued.expires_at)
|
|
|
|
detail = f"method={method}"
|
|
if device_id:
|
|
detail += f"; device={mask_device_id(device_id)}"
|
|
db.add(
|
|
AuditLog(
|
|
user_id=user.id,
|
|
username=user.username,
|
|
action="login",
|
|
target="auth",
|
|
detail=detail,
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
return TokenResponse(
|
|
access_token=issued.token,
|
|
role=user.role,
|
|
username=user.username,
|
|
permissions=get_user_permissions(user),
|
|
)
|
|
|
|
|
|
def _require_phone(user: User) -> str:
|
|
phone = str(user.phone or "").strip()
|
|
if not phone:
|
|
raise ApiError(
|
|
409,
|
|
"该账号尚未绑定手机号，请联系管理员",
|
|
"phone_unbound",
|
|
)
|
|
return phone
|
|
|
|
|
|
@router.post("/sms-code", response_model=SmsCodeResponse)
|
|
def send_sms_code(
|
|
req: SmsCodeRequest,
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = _authenticate(db, req.username, req.password)
|
|
phone = _require_phone(user)
|
|
request_ip = request.client.host if request.client else ""
|
|
result = send_login_sms_code(db, user, phone, request_ip=request_ip)
|
|
return SmsCodeResponse(
|
|
sent=True,
|
|
phone_masked=result["phoneMasked"],
|
|
expires_in_seconds=result["expiresInSeconds"],
|
|
provider=result["provider"],
|
|
)
|
|
|
|
|
|
@router.post("/prepare", response_model=LoginPrepareResponse)
|
|
def prepare_login(
|
|
req: LoginPrepareRequest,
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = _authenticate(db, req.username, req.password)
|
|
phone = _require_phone(user)
|
|
|
|
# 先校验并消费短信码，再创建 challenge（与参考实现一致）。
|
|
verify_login_sms_code(db, user.id, req.sms_code)
|
|
|
|
api_base = normalize_api_base(req.api_base or _request_origin(request))
|
|
challenge = create_login_challenge(db, user.id, api_base)
|
|
return LoginPrepareResponse(
|
|
challenge_id=challenge["challengeId"],
|
|
challenge=challenge["challenge"],
|
|
expires_at=challenge["expiresAt"],
|
|
poll_token=challenge["pollToken"],
|
|
launch_url=challenge["launchUrl"],
|
|
phone_masked=mask_phone(phone),
|
|
app=challenge["app"],
|
|
)
|
|
|
|
|
|
@router.post("/complete", response_model=TokenResponse)
|
|
def complete_login(
|
|
req: LoginCompleteRequest,
|
|
response: Response,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
challenge = get_challenge(db, req.challenge_id)
|
|
if (
|
|
challenge is None
|
|
or challenge.user_id is None
|
|
or challenge.app != APP_ID
|
|
or challenge.purpose != CHALLENGE_PURPOSE
|
|
):
|
|
raise ApiError(
|
|
404,
|
|
"设备认证会话不存在，请重新登录",
|
|
"challenge_not_found",
|
|
)
|
|
|
|
user = (
|
|
db.query(User)
|
|
.filter(User.id == challenge.user_id, User.deleted_at.is_(None))
|
|
.first()
|
|
)
|
|
if user is None or not user.is_active:
|
|
raise ApiError(401, "账号已不可用，请重新登录", "user_disabled")
|
|
|
|
consumed: DeviceChallenge = consume_device_ticket(
|
|
db,
|
|
req.challenge_id,
|
|
req.poll_token,
|
|
req.device_ticket,
|
|
)
|
|
if consumed.user_id != user.id:
|
|
raise ApiError(401, "设备认证票据无效", "ticket_invalid")
|
|
|
|
ensure_device_binding(db, user, consumed)
|
|
return _issue_session(db, response, user, "sms_device", consumed.device_id)
|
|
|
|
|
|
@router.post("/login", response_model=TokenResponse)
|
|
def login(
|
|
req: LoginRequest,
|
|
response: Response,
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""旧账号兜底：仅当用户未绑定手机号时允许密码单因素登录。"""
|
|
user = _authenticate(db, req.username, req.password)
|
|
if str(user.phone or "").strip():
|
|
raise ApiError(
|
|
403,
|
|
"该账号已启用安全登录，请使用用户名 + 密码 + 短信验证码 + 本机验证",
|
|
"secure_login_required",
|
|
)
|
|
|
|
return _issue_session(db, response, user, "password_legacy")
|
|
|
|
|
|
@router.get("/me")
|
|
def me(current_user: User = Depends(get_current_user)):
|
|
return {
|
|
"id": current_user.id,
|
|
"username": current_user.username,
|
|
"role": current_user.role,
|
|
"role_label": ROLE_LABELS.get(current_user.role, current_user.role),
|
|
"is_active": current_user.is_active,
|
|
"remark": current_user.remark,
|
|
"permissions": get_user_permissions(current_user),
|
|
"phone_masked": mask_phone(current_user.phone),
|
|
"phone_bound": bool(str(current_user.phone or "").strip()),
|
|
}
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout(
|
|
response: Response,
|
|
current_user: User = Depends(get_current_user),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
clear_session_cookie(response)
|
|
db.add(
|
|
AuditLog(
|
|
user_id=current_user.id,
|
|
username=current_user.username,
|
|
action="logout",
|
|
target="auth",
|
|
)
|
|
)
|
|
db.commit()
|
|
return {"message": "已登出"}
|