340 lines
12 KiB
Python
340 lines
12 KiB
Python
from datetime import datetime, timedelta
|
|
import secrets
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, Cookie, HTTPException, Request, Response, status
|
|
|
|
from app.config import auth_settings
|
|
from app.database import Database
|
|
from app.schemas import AuthUser, CaptchaResponse, LoginRequest, LoginResponse
|
|
from app.security import create_token, hash_secret, verify_password
|
|
|
|
|
|
router = APIRouter(prefix="/auth", tags=["auth"])
|
|
|
|
CAPTCHA_TTL_SECONDS = 300
|
|
SESSION_COOKIE_NAME = "admin_session"
|
|
SESSION_TTL_SECONDS = 7 * 24 * 60 * 60
|
|
CAPTCHA_FAILURE_THRESHOLD = 5
|
|
LOCK_FAILURE_THRESHOLD = 10
|
|
FAILURE_WINDOW_MINUTES = 15
|
|
LOCK_MINUTES = 15
|
|
|
|
|
|
def utc_now() -> datetime:
|
|
return datetime.utcnow().replace(microsecond=0)
|
|
|
|
|
|
def to_db_time(value: datetime) -> str:
|
|
return value.isoformat()
|
|
|
|
|
|
def from_db_time(value: str | None) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(value)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def get_client_ip(request: Request) -> str:
|
|
forwarded_for = request.headers.get("x-forwarded-for")
|
|
if forwarded_for:
|
|
return forwarded_for.split(",")[0].strip()
|
|
return request.client.host if request.client else "unknown"
|
|
|
|
|
|
def get_user_agent(request: Request) -> str:
|
|
return request.headers.get("user-agent", "")[:300]
|
|
|
|
|
|
def is_local_request(request: Request) -> bool:
|
|
host = request.url.hostname or ""
|
|
return host in {"127.0.0.1", "localhost", "::1"}
|
|
|
|
|
|
def ensure_https(request: Request) -> None:
|
|
if not auth_settings.require_https:
|
|
return
|
|
forwarded_proto = request.headers.get("x-forwarded-proto", "")
|
|
is_https = request.url.scheme == "https" or forwarded_proto.lower() == "https"
|
|
if is_https:
|
|
return
|
|
if auth_settings.allow_insecure_local and is_local_request(request):
|
|
return
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="登录必须通过 HTTPS 连接，请使用安全地址后重试",
|
|
)
|
|
|
|
|
|
def create_auth_user(row) -> AuthUser:
|
|
return AuthUser(
|
|
id=row["id"],
|
|
username=row["username"],
|
|
name=row["name"],
|
|
is_admin=bool(row["is_admin"]),
|
|
)
|
|
|
|
|
|
def count_recent_failures(db: Database, username: str, ip_address: str) -> int:
|
|
since = to_db_time(utc_now() - timedelta(minutes=FAILURE_WINDOW_MINUTES))
|
|
row = db.execute(
|
|
"""
|
|
SELECT COUNT(*)
|
|
FROM login_attempts
|
|
WHERE success = 0
|
|
AND created_at >= ?
|
|
AND (lower(username) = ? OR ip_address = ?)
|
|
""",
|
|
(since, username, ip_address),
|
|
).fetchone()
|
|
return int(row[0])
|
|
|
|
|
|
def record_login_attempt(
|
|
db: Database,
|
|
username: str,
|
|
ip_address: str,
|
|
user_agent: str,
|
|
success: bool,
|
|
failure_reason: str | None = None,
|
|
) -> None:
|
|
db.execute(
|
|
"""
|
|
INSERT INTO login_attempts (username, ip_address, user_agent, success, failure_reason)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(username, ip_address, user_agent, int(success), failure_reason),
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def detect_new_login_context(db: Database, user_id: int, ip_address: str, user_agent: str) -> tuple[bool, bool]:
|
|
old_ip = db.execute(
|
|
"""
|
|
SELECT 1 FROM login_events
|
|
WHERE user_id = ? AND ip_address = ?
|
|
LIMIT 1
|
|
""",
|
|
(user_id, ip_address),
|
|
).fetchone()
|
|
old_device = db.execute(
|
|
"""
|
|
SELECT 1 FROM login_events
|
|
WHERE user_id = ? AND user_agent = ?
|
|
LIMIT 1
|
|
""",
|
|
(user_id, user_agent),
|
|
).fetchone()
|
|
had_login = db.execute("SELECT 1 FROM login_events WHERE user_id = ? LIMIT 1", (user_id,)).fetchone()
|
|
return had_login is not None and old_ip is None, had_login is not None and old_device is None
|
|
|
|
|
|
def record_login_event(db: Database, user_id: int, ip_address: str, user_agent: str, is_new_ip: bool, is_new_device: bool) -> None:
|
|
db.execute(
|
|
"""
|
|
INSERT INTO login_events (user_id, ip_address, user_agent, is_new_ip, is_new_device)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""",
|
|
(user_id, ip_address, user_agent, int(is_new_ip), int(is_new_device)),
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def set_session_cookie(response: Response, token: str) -> None:
|
|
response.set_cookie(
|
|
key=SESSION_COOKIE_NAME,
|
|
value=token,
|
|
max_age=SESSION_TTL_SECONDS,
|
|
httponly=True,
|
|
secure=auth_settings.cookie_secure,
|
|
samesite="lax",
|
|
path="/",
|
|
)
|
|
|
|
|
|
def clear_session_cookie(response: Response) -> None:
|
|
response.delete_cookie(key=SESSION_COOKIE_NAME, path="/")
|
|
|
|
|
|
@router.get("/captcha", response_model=CaptchaResponse)
|
|
def get_captcha(db: Database) -> CaptchaResponse:
|
|
left = secrets.randbelow(8) + 2
|
|
right = secrets.randbelow(8) + 2
|
|
answer = str(left + right)
|
|
captcha_id = uuid4().hex
|
|
expires_at = utc_now() + timedelta(seconds=CAPTCHA_TTL_SECONDS)
|
|
|
|
db.execute(
|
|
"""
|
|
INSERT INTO login_captchas (id, answer_hash, expires_at)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
(captcha_id, hash_secret(answer.lower()), to_db_time(expires_at)),
|
|
)
|
|
db.commit()
|
|
|
|
return CaptchaResponse(
|
|
captcha_id=captcha_id,
|
|
challenge=f"{left} + {right} = ?",
|
|
expires_in=CAPTCHA_TTL_SECONDS,
|
|
)
|
|
|
|
|
|
def verify_captcha(db: Database, captcha_id: str | None, captcha_code: str | None) -> None:
|
|
if not captcha_id or not captcha_code:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请输入验证码后再登录")
|
|
|
|
row = db.execute(
|
|
"""
|
|
SELECT id, answer_hash, expires_at, used_at
|
|
FROM login_captchas
|
|
WHERE id = ?
|
|
""",
|
|
(captcha_id,),
|
|
).fetchone()
|
|
if row is None or row["used_at"]:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="验证码已失效，请刷新后重试")
|
|
|
|
expires_at = from_db_time(row["expires_at"])
|
|
if expires_at is None or utc_now() > expires_at:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="验证码已过期，请刷新后重试")
|
|
|
|
db.execute("UPDATE login_captchas SET used_at = ? WHERE id = ?", (to_db_time(utc_now()), captcha_id))
|
|
db.commit()
|
|
|
|
if hash_secret(captcha_code.strip().lower()) != row["answer_hash"]:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="验证码错误，请重新输入")
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
def login(payload: LoginRequest, request: Request, response: Response, db: Database) -> LoginResponse:
|
|
ensure_https(request)
|
|
|
|
username = payload.username.strip().lower()
|
|
ip_address = get_client_ip(request)
|
|
user_agent = get_user_agent(request)
|
|
recent_failures = count_recent_failures(db, username, ip_address)
|
|
|
|
try:
|
|
verify_captcha(db, payload.captcha_id, payload.captcha_code)
|
|
except HTTPException as exc:
|
|
record_login_attempt(db, username, ip_address, user_agent, False, "captcha_error")
|
|
raise exc
|
|
|
|
row = db.execute(
|
|
"""
|
|
SELECT id, username, name, is_admin, password_hash, failed_login_count, locked_until
|
|
FROM users
|
|
WHERE lower(username) = ?
|
|
""",
|
|
(username,),
|
|
).fetchone()
|
|
if row is None:
|
|
record_login_attempt(db, username, ip_address, user_agent, False, "unknown_user")
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="账号不存在，请检查用户名")
|
|
|
|
locked_until = from_db_time(row["locked_until"])
|
|
if locked_until is not None and utc_now() < locked_until:
|
|
record_login_attempt(db, username, ip_address, user_agent, False, "locked")
|
|
raise HTTPException(status_code=status.HTTP_423_LOCKED, detail="账号已临时锁定，请 15 分钟后再试")
|
|
|
|
password_hash = row["password_hash"]
|
|
if not password_hash:
|
|
record_login_attempt(db, username, ip_address, user_agent, False, "password_missing")
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="账号尚未设置密码，请联系管理员")
|
|
|
|
if not verify_password(payload.password, password_hash):
|
|
failure_count = int(row["failed_login_count"] or 0) + 1
|
|
total_failures = max(failure_count, recent_failures + 1)
|
|
locked_value = None
|
|
message = "密码错误，请重新输入"
|
|
if total_failures >= LOCK_FAILURE_THRESHOLD:
|
|
locked_value = to_db_time(utc_now() + timedelta(minutes=LOCK_MINUTES))
|
|
message = "密码连续错误次数过多，账号已临时锁定 15 分钟"
|
|
elif total_failures >= CAPTCHA_FAILURE_THRESHOLD:
|
|
message = "密码错误，请输入验证码后重试"
|
|
db.execute(
|
|
"""
|
|
UPDATE users
|
|
SET failed_login_count = ?, locked_until = ?
|
|
WHERE id = ?
|
|
""",
|
|
(failure_count, locked_value, row["id"]),
|
|
)
|
|
db.commit()
|
|
record_login_attempt(db, username, ip_address, user_agent, False, "wrong_password")
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=message)
|
|
|
|
token = create_token()
|
|
expires_at = utc_now() + timedelta(seconds=SESSION_TTL_SECONDS)
|
|
db.execute(
|
|
"""
|
|
INSERT INTO auth_sessions (token_hash, user_id, expires_at)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
(hash_secret(token), row["id"], to_db_time(expires_at)),
|
|
)
|
|
db.execute(
|
|
"""
|
|
UPDATE users
|
|
SET failed_login_count = 0, locked_until = NULL
|
|
WHERE id = ?
|
|
""",
|
|
(row["id"],),
|
|
)
|
|
db.commit()
|
|
set_session_cookie(response, token)
|
|
record_login_attempt(db, username, ip_address, user_agent, True)
|
|
|
|
is_new_ip, is_new_device = detect_new_login_context(db, row["id"], ip_address, user_agent)
|
|
record_login_event(db, row["id"], ip_address, user_agent, is_new_ip, is_new_device)
|
|
requires_second_verification = is_new_ip or is_new_device
|
|
security_notice = None
|
|
if requires_second_verification:
|
|
security_notice = "检测到新 IP 或新设备登录，后续应接入短信/邮箱二次验证与通知"
|
|
|
|
return LoginResponse(
|
|
expires_in=SESSION_TTL_SECONDS,
|
|
user=create_auth_user(row),
|
|
requires_second_verification=requires_second_verification,
|
|
security_notice=security_notice,
|
|
)
|
|
|
|
|
|
@router.get("/me", response_model=AuthUser)
|
|
def get_current_user(db: Database, admin_session: str | None = Cookie(default=None)) -> AuthUser:
|
|
if not admin_session:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="请先登录")
|
|
|
|
row = db.execute(
|
|
"""
|
|
SELECT users.id, users.username, users.name, users.is_admin, auth_sessions.expires_at, auth_sessions.revoked_at
|
|
FROM auth_sessions
|
|
JOIN users ON users.id = auth_sessions.user_id
|
|
WHERE auth_sessions.token_hash = ?
|
|
""",
|
|
(hash_secret(admin_session),),
|
|
).fetchone()
|
|
if row is None or row["revoked_at"]:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="登录状态已失效，请重新登录")
|
|
|
|
expires_at = from_db_time(row["expires_at"])
|
|
if expires_at is None or utc_now() > expires_at:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="登录已过期，请重新登录")
|
|
|
|
return create_auth_user(row)
|
|
|
|
|
|
@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
|
|
def logout(response: Response, db: Database, admin_session: str | None = Cookie(default=None)) -> None:
|
|
if admin_session:
|
|
db.execute(
|
|
"UPDATE auth_sessions SET revoked_at = ? WHERE token_hash = ?",
|
|
(to_db_time(utc_now()), hash_secret(admin_session)),
|
|
)
|
|
db.commit()
|
|
clear_session_cookie(response)
|