dekun
126 lines
3.9 KiB
Python
126 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import hmac
|
|
import time
|
|
from collections import defaultdict
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from pydantic import BaseModel, Field
|
|
|
|
from ..config import Settings, get_settings
|
|
from ..credentials import get_credentials, update_credentials, upsert_env_file
|
|
from .auth import LoginRequest, LoginResponse, issue_token, require_user
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
_login_hits: dict[str, list[float]] = defaultdict(list)
|
|
|
|
|
|
class ChangeCredentialsRequest(BaseModel):
|
|
current_password: str = Field(min_length=1)
|
|
new_username: str = Field(min_length=1, max_length=64)
|
|
new_password: str = Field(min_length=8, max_length=128)
|
|
|
|
|
|
def _client_ip(request: Request) -> str:
|
|
xff = request.headers.get("x-forwarded-for") or ""
|
|
if xff.strip():
|
|
return xff.split(",")[0].strip()
|
|
if request.client:
|
|
return request.client.host or "unknown"
|
|
return "unknown"
|
|
|
|
|
|
def _rate_limit_login(ip: str, settings: Settings) -> None:
|
|
now = time.time()
|
|
window = float(settings.login_window_sec)
|
|
max_n = int(settings.login_max_attempts)
|
|
hits = [t for t in _login_hits[ip] if now - t < window]
|
|
_login_hits[ip] = hits
|
|
if len(hits) >= max_n:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
|
detail=f"登录过于频繁，请 {int(window)} 秒后再试",
|
|
)
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
async def login(
|
|
body: LoginRequest,
|
|
request: Request,
|
|
settings: Annotated[Settings, Depends(get_settings)],
|
|
) -> LoginResponse:
|
|
ip = _client_ip(request)
|
|
_rate_limit_login(ip, settings)
|
|
user, pwd = get_credentials()
|
|
user_ok = hmac.compare_digest(
|
|
body.username.encode("utf-8"), user.encode("utf-8")
|
|
)
|
|
pwd_ok = hmac.compare_digest(
|
|
body.password.encode("utf-8"), pwd.encode("utf-8")
|
|
)
|
|
if not (user_ok and pwd_ok):
|
|
_login_hits[ip].append(time.time())
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误"
|
|
)
|
|
_login_hits.pop(ip, None)
|
|
token, ttl = issue_token(body.username, settings)
|
|
return LoginResponse(
|
|
token=token,
|
|
username=body.username,
|
|
expires_in=ttl,
|
|
env_name=settings.env_name,
|
|
mode=settings.mode,
|
|
)
|
|
|
|
|
|
@router.get("/me")
|
|
async def me(
|
|
username: Annotated[str, Depends(require_user)],
|
|
settings: Annotated[Settings, Depends(get_settings)],
|
|
) -> dict:
|
|
return {
|
|
"username": username,
|
|
"env_name": settings.env_name,
|
|
"mode": settings.mode,
|
|
"sim": settings.is_sim,
|
|
}
|
|
|
|
|
|
@router.post("/change-credentials", response_model=LoginResponse)
|
|
async def change_credentials(
|
|
body: ChangeCredentialsRequest,
|
|
username: Annotated[str, Depends(require_user)],
|
|
settings: Annotated[Settings, Depends(get_settings)],
|
|
) -> LoginResponse:
|
|
_cur_user, cur_pwd = get_credentials()
|
|
if not hmac.compare_digest(
|
|
body.current_password.encode("utf-8"), cur_pwd.encode("utf-8")
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST, detail="当前密码不正确"
|
|
)
|
|
try:
|
|
update_credentials(
|
|
new_username=body.new_username, new_password=body.new_password
|
|
)
|
|
# 作废旧 token
|
|
new_ver = int(settings.auth_token_version) + 1
|
|
upsert_env_file("AUTH_TOKEN_VERSION", str(new_ver))
|
|
get_settings.cache_clear()
|
|
except ValueError as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)
|
|
) from e
|
|
settings2 = get_settings()
|
|
token, ttl = issue_token(body.new_username.strip(), settings2)
|
|
return LoginResponse(
|
|
token=token,
|
|
username=body.new_username.strip(),
|
|
expires_in=ttl,
|
|
env_name=settings2.env_name,
|
|
mode=settings2.mode,
|
|
)
|