feat(auth): 限制同时在线 IP 数 (默认 30, 第 31 拒绝)

背景: 防 token 泄漏被滥用 + 限共享账号人数。 - 新增 app/services/active_ip.py: Redis ZSET 'active_ips' 存 IP,score=last_seen_unix 登录/refresh 时 check_or_register_login_ip(): IP 已在 set → 刷新 score,放行(老用户重连) IP 不在 set + ZCARD < 30 → 加入,放行 IP 不在 set + ZCARD >= 30 → raise 429 每个已认证请求 _resolve_user() 调 touch_ip_dependency() 滑动 TTL,30 天没活动自动从 set 剔除 - get_client_ip() 取真实 IP,优先级 X-Forwarded-For > X-Real-IP > client.host trust_x_forwarded_for 默认 True(生产 Caddy/Nginx 后面) - config 加 3 个开关: site_max_active_ips: int = 30 site_active_ip_idle_days: int = 30 trust_x_forwarded_for: bool = True - admin.py 加 3 个端点: GET /admin/active-ips — 看当前活跃 IP 列表 + last_seen POST /admin/active-ips/kick — 强制踢出指定 IP(body={ip}) DELETE /admin/active-ips/{ip}— 简写踢出 - 注: refresh 也算 IP 占用(拿到 access token 就能用) 但已存在的 IP 直接放行,不会踢自己
2026-06-13 18:22:40 +08:00
parent 0df6e79e56
commit b500613d22
5 changed files with 259 additions and 4 deletions
--- a/backend/app/api/admin.py
+++ b/backend/app/api/admin.py
@@ -205,3 +205,58 @@ async def reset_quota(payload: QuotaReset) -> dict[str, Any]:
    key = f"translation:month:{now:%Y%m}"
    await r.set(key, payload.used_chars)
    return {"key": key, "value": payload.used_chars}
+
+
+# === 活跃 IP 看板(防 token 泄漏 / 排查异地登录)===
+class ActiveIpItem(BaseModel):
+    ip: str
+    last_seen_unix: int
+    last_seen_iso: str
+    idle_seconds: int
+
+
+class ActiveIpList(BaseModel):
+    items: list[ActiveIpItem]
+    limit: int
+    idle_days: int
+    count: int
+
+
+@router.get("/active-ips", response_model=ActiveIpList)
+async def list_active_ips_admin():
+    from app.config import settings
+    from app.services.active_ip import list_active_ips
+
+    items = await list_active_ips()
+    return ActiveIpList(
+        items=[ActiveIpItem(**i) for i in items],
+        limit=settings.site_max_active_ips,
+        idle_days=settings.site_active_ip_idle_days,
+        count=len(items),
+    )
+
+
+class KickIpRequest(BaseModel):
+    ip: str
+
+
+class KickIpResponse(BaseModel):
+    kicked: bool
+    ip: str
+
+
+@router.post("/active-ips/kick", response_model=KickIpResponse)
+async def kick_active_ip(payload: KickIpRequest):
+    """owner 强制剔除某个 IP。下次该 IP 登录时会再次被算作"新 IP"。"""
+    from app.services.active_ip import remove_active_ip
+
+    ok = await remove_active_ip(payload.ip)
+    return KickIpResponse(kicked=ok, ip=payload.ip)
+
+
+@router.delete("/active-ips/{ip}", response_model=KickIpResponse)
+async def delete_active_ip(ip: str):
+    from app.services.active_ip import remove_active_ip
+
+    ok = await remove_active_ip(ip)
+    return KickIpResponse(kicked=ok, ip=ip)
--- a/backend/app/api/auth.py
+++ b/backend/app/api/auth.py
@@ -3,7 +3,7 @@ from __future__ import annotations

 from datetime import datetime, timezone

-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Request, status
 from jwt.exceptions import InvalidTokenError
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -18,6 +18,7 @@ from app.core.security import (
 from app.database import get_session
 from app.models.user import User
 from app.schemas.auth import LoginRequest, RefreshRequest, TokenPair
+from app.services.active_ip import check_or_register_login_ip, get_client_ip

 router = APIRouter(prefix="/auth", tags=["auth"])

@@ -33,7 +34,16 @@ def _pair_for(user: User) -> TokenPair:


@router.post("/login", response_model=TokenPair)
-async def login(body: LoginRequest, session: AsyncSession = Depends(get_session)):
+async def login(
+    body: LoginRequest,
+    request: Request,
+    session: AsyncSession = Depends(get_session),
+):
+    # === 先校验 IP 上限(在查 DB 前,避免密码错误也消耗 DB)===
+    # 注:这一步是公开的(未认证),不暴露"这个 IP 是不是 owner"的信息
+    client_ip = get_client_ip(request)
+    await check_or_register_login_ip(client_ip)
+
    result = await session.execute(select(User).where(User.username == body.username))
    user = result.scalars().first()
    if not user or not user.is_active or not verify_password(body.password, user.password_hash):
@@ -44,7 +54,16 @@ async def login(body: LoginRequest, session: AsyncSession = Depends(get_session)


@router.post("/refresh", response_model=TokenPair)
-async def refresh(body: RefreshRequest, session: AsyncSession = Depends(get_session)):
+async def refresh(
+    body: RefreshRequest,
+    request: Request,
+    session: AsyncSession = Depends(get_session),
+):
+    # refresh 也要算 IP 占用(因为 refresh 拿到 access token 就能用)
+    # 但如果 IP 已经在 set 里,等于合法老用户,直接放过
+    client_ip = get_client_ip(request)
+    await check_or_register_login_ip(client_ip)
+
    try:
        payload = decode_token(body.refresh_token)
        if payload.get("type") != "refresh":
--- a/backend/app/config.py
+++ b/backend/app/config.py
@@ -120,6 +120,15 @@ class Settings(BaseSettings):
    fetch_fail_pause_threshold: int = 3
    fetch_max_retries: int = 2

+    # ===== 站点并发登录 IP 限制 =====
+    # 限制同时在线的客户端 IP 数(防滥用 + 防 token 泄漏被滥用)
+    # Redis ZSET 滑动窗口:每次已认证请求刷新 score,30 天没活动自动剔除
+    # 第 (limit+1) 个新 IP 登录时直接 429
+    site_max_active_ips: int = 30
+    site_active_ip_idle_days: int = 30
+    # 是否信任反向代理的 X-Forwarded-For 头(生产用 Caddy/Nginx 必开;直连调测关)
+    trust_x_forwarded_for: bool = True
+
    # ===== Caddy / 域名 =====
    domain: str = ""
    acme_email: str = ""
--- a/backend/app/core/deps.py
+++ b/backend/app/core/deps.py
@@ -3,7 +3,7 @@ from __future__ import annotations

 from datetime import datetime, timezone

-from fastapi import Depends, HTTPException, status
+from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from jwt.exceptions import InvalidTokenError
 from sqlalchemy import select
@@ -13,11 +13,13 @@ from app.core.security import decode_token, hash_api_token
 from app.database import get_session
 from app.models.api_token import ApiToken
 from app.models.user import User, UserRole
+from app.services.active_ip import touch_ip_dependency

 _bearer = HTTPBearer(auto_error=False)


 async def _resolve_user(
+    request: Request,
    creds: HTTPAuthorizationCredentials | None = Depends(_bearer),
    session: AsyncSession = Depends(get_session),
 ) -> User:
@@ -40,6 +42,8 @@ async def _resolve_user(
        if user and user.is_active:
            api_row.last_used_at = datetime.now(timezone.utc)
            await session.commit()
+            # 顺带刷新活跃 IP(防 token 泄漏后被滥用,绑定客户端 IP)
+            await touch_ip_dependency(request)
            return user

    # 2) 试 JWT
@@ -57,6 +61,9 @@ async def _resolve_user(
    user = result.scalars().first()
    if user is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "User not found or inactive")
+
+    # 已认证 → 刷新当前 IP 活跃时间
+    await touch_ip_dependency(request)
    return user


--- a/backend/app/services/active_ip.py
+++ b/backend/app/services/active_ip.py
@@ -0,0 +1,165 @@
+"""站点活跃 IP 限制(防 token 泄漏被滥用)
+
+- Redis SortedSet `active_ips` 存当前活跃客户端 IP
+  - member = IP 字符串
+  - score  = last_seen unix 秒
+- 每次已认证请求: 刷新该 IP 的 score(滑动 TTL)
+- 登录时: 检查 ZCARD
+  - 当前 IP 已在 set → 直接放行(自己人不踢)
+  - 当前 IP 不在 set 且 ZCARD >= limit → 拒绝登录
+  - 当前 IP 不在 set 且 ZCARD <  limit → 加入 set,放行
+- 30 天没活动的 IP 自动从 set 剔除
+- 整个 key 30 天无活动自动过期(EXPIRE)
+
+注意:
+- 信任 X-Forwarded-For 必须显式开启(trust_x_forwarded_for=True),
+  否则攻击者伪造头绕过 IP 限制
+- 不限流请求本身(只限流"新 IP 登录"),活跃 IP 持续刷新即可
+- API token(api_tokens 表)也走这个限流(同一个 IP 拿到 token 后被滥用也防)
+"""
+from __future__ import annotations
+
+import logging
+import time
+from typing import Annotated
+
+from fastapi import Depends, HTTPException, Request, status
+
+from app.config import settings
+from app.redis_client import get_redis
+
+logger = logging.getLogger("news.auth.active_ip")
+
+REDIS_KEY = "active_ips"
+
+
+def get_client_ip(request: Request) -> str:
+    """提取客户端真实 IP。
+
+    优先级:
+    1) X-Forwarded-For 第一个(被代理链信任时)
+    2) X-Real-IP(Nginx 风格)
+    3) request.client.host(直连)
+
+    仅在配置 trust_x_forwarded_for=True 时信任 1+2,否则只用 3。
+    """
+    if settings.trust_x_forwarded_for:
+        xff = request.headers.get("x-forwarded-for")
+        if xff:
+            # 取第一个(原始客户端),可能形如 "1.2.3.4, 10.0.0.1"
+            ip = xff.split(",")[0].strip()
+            if ip:
+                return ip
+        xri = request.headers.get("x-real-ip")
+        if xri:
+            return xri.strip()
+    if request.client and request.client.host:
+        return request.client.host
+    return "unknown"
+
+
+async def _prune_and_count(now: int) -> int:
+    """清理过期 IP 并返回当前活跃数。"""
+    r = get_redis()
+    idle_cutoff = now - settings.site_active_ip_idle_days * 86400
+    # 删 score < idle_cutoff
+    await r.zremrangebyscore(REDIS_KEY, 0, idle_cutoff - 1)
+    return await r.zcard(REDIS_KEY)
+
+
+async def check_or_register_login_ip(ip: str) -> None:
+    """登录/refresh 时调用。
+
+    - 如果 IP 已在 set 里:刷新 score,放行
+    - 如果 IP 不在 set 里,且活跃数 < limit:加入 set,放行
+    - 如果 IP 不在 set 里,且活跃数 >= limit:raise 429
+
+    注意:不限制 owner/非 owner(整个站点一个池,所有用户共享)
+    """
+    r = get_redis()
+    now = int(time.time())
+    limit = settings.site_max_active_ips
+
+    # 看 IP 是否已在 set
+    existing = await r.zscore(REDIS_KEY, ip)
+    if existing is not None:
+        # 已在 set,刷新 score 即可(滑动 TTL)
+        await r.zadd(REDIS_KEY, {ip: now})
+        await r.expire(REDIS_KEY, settings.site_active_ip_idle_days * 86400)
+        return
+
+    # 不在 set — 先清理过期再计数
+    count = await _prune_and_count(now)
+    if count >= limit:
+        # 第 (limit+1) 个新 IP,拒绝
+        logger.warning(
+            "active IP limit reached: %d >= %d, reject login from %s",
+            count, limit, ip,
+        )
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail=(
+                f"站点同时在线 IP 数已达上限({limit} 个)。"
+                f"如有疑问请等待空闲 IP 退出后重试,或联系 owner 调整 site_max_active_ips。"
+            ),
+            headers={"Retry-After": "3600"},
+        )
+
+    # 通过:加入 set
+    await r.zadd(REDIS_KEY, {ip: now})
+    await r.expire(REDIS_KEY, settings.site_active_ip_idle_days * 86400)
+    logger.info("active IP registered: %s (total now %d)", ip, count + 1)
+
+
+async def touch_active_ip(ip: str) -> None:
+    """每次已认证请求时调用,刷新该 IP 的 score(滑动 TTL)。
+
+    已废弃 IP 不主动删,等下次 login 时 prune。
+    """
+    r = get_redis()
+    now = int(time.time())
+    # NX:仅当 member 不存在时设置 — 但我们要 UPDATE score,不能用 NX
+    # 正常 ZADD 即可:member 存在就更新 score
+    await r.zadd(REDIS_KEY, {ip: now})
+    # 同时顺手 prune,避免 set 无限膨胀
+    await _prune_and_count(now)
+    await r.expire(REDIS_KEY, settings.site_active_ip_idle_days * 86400)
+
+
+async def list_active_ips() -> list[dict]:
+    """给 owner 看:当前活跃 IP + last_seen。
+    按 last_seen 倒序。
+    """
+    r = get_redis()
+    now = int(time.time())
+    await _prune_and_count(now)
+    rows = await r.zrange(REDIS_KEY, 0, -1, withscores=True, desc=True)
+    return [
+        {
+            "ip": ip,
+            "last_seen_unix": int(score),
+            "last_seen_iso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(score)),
+            "idle_seconds": int(now - score),
+        }
+        for ip, score in rows
+    ]
+
+
+async def remove_active_ip(ip: str) -> bool:
+    """owner 强制踢人。"""
+    r = get_redis()
+    removed = await r.zrem(REDIS_KEY, ip)
+    return bool(removed)
+
+
+# === FastAPI 依赖 ===
+async def touch_ip_dependency(request: Request) -> None:
+    """每个已认证请求的依赖:刷新当前 IP 的 last_seen。
+
+    不会失败(吞掉异常),避免 Redis 抖动影响正常请求。
+    """
+    try:
+        ip = get_client_ip(request)
+        await touch_active_ip(ip)
+    except Exception as e:  # noqa: BLE001
+        logger.debug("touch_active_ip failed (ignored): %s", e)