1252 lines
54 KiB
Python
1252 lines
54 KiB
Python
"""Authentication endpoints."""
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import os
|
||
import time
|
||
from ipaddress import ip_address, ip_network
|
||
from typing import Literal
|
||
|
||
import httpx
|
||
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
|
||
from fastapi.security import OAuth2PasswordRequestForm
|
||
from pydantic import BaseModel, ConfigDict, EmailStr, Field, TypeAdapter, ValidationError, field_validator, model_validator
|
||
|
||
from app.gateway.admin_query_window import recent_admin_query_cutoff
|
||
from app.gateway.auth import (
|
||
UserResponse,
|
||
create_access_token,
|
||
)
|
||
from app.gateway.auth.config import get_auth_config
|
||
from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse
|
||
from app.gateway.auth.models import User
|
||
from app.gateway.auth.password import has_account_password, hash_password_async, is_password_hash, verify_account_password
|
||
from app.gateway.auth.repositories.base import CredentialConflictError
|
||
from app.gateway.csrf_middleware import is_secure_request
|
||
from app.gateway.deps import get_agent_store, get_current_user_from_request, get_local_provider, get_run_store, get_skill_store
|
||
from app.gateway.perf_logging import finish_trace, trace_request, trace_stage
|
||
from deerflow.config import get_app_config
|
||
from deerflow.config.system_settings import AuthenticationSettings, load_system_settings
|
||
from deerflow.persistence.agents.base import AgentStore
|
||
from deerflow.persistence.skills.base import SkillStore
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
|
||
_USER_NOT_LOOKED_UP = object()
|
||
|
||
|
||
# ── Request/Response Models ──────────────────────────────────────────────
|
||
|
||
|
||
class LoginResponse(BaseModel):
|
||
"""Response model for login — token only lives in HttpOnly cookie."""
|
||
|
||
expires_in: int # seconds
|
||
needs_setup: bool = False
|
||
|
||
|
||
# Top common-password blocklist. Drawn from the public SecLists "10k worst
|
||
# passwords" set, lowercased + length>=8 only (shorter ones already fail
|
||
# the min_length check). Kept tight on purpose: this is the **lower bound**
|
||
# defense, not a full HIBP / passlib check, and runs in-process per request.
|
||
_COMMON_PASSWORDS: frozenset[str] = frozenset(
|
||
{
|
||
"password",
|
||
"password1",
|
||
"password12",
|
||
"password123",
|
||
"password1234",
|
||
"12345678",
|
||
"123456789",
|
||
"1234567890",
|
||
"qwerty12",
|
||
"qwertyui",
|
||
"qwerty123",
|
||
"abc12345",
|
||
"abcd1234",
|
||
"iloveyou",
|
||
"letmein1",
|
||
"welcome1",
|
||
"welcome123",
|
||
"admin123",
|
||
"administrator",
|
||
"passw0rd",
|
||
"p@ssw0rd",
|
||
"monkey12",
|
||
"trustno1",
|
||
"sunshine",
|
||
"princess",
|
||
"football",
|
||
"baseball",
|
||
"superman",
|
||
"batman123",
|
||
"starwars",
|
||
"dragon123",
|
||
"master123",
|
||
"shadow12",
|
||
"michael1",
|
||
"jennifer",
|
||
"computer",
|
||
}
|
||
)
|
||
|
||
|
||
def _password_is_common(password: str) -> bool:
|
||
"""Case-insensitive blocklist check.
|
||
|
||
Lowercases the input so trivial mutations like ``Password`` /
|
||
``PASSWORD`` are also rejected. Does not normalize digit substitutions
|
||
(``p@ssw0rd`` is included as a literal entry instead) — keeping the
|
||
rule cheap and predictable.
|
||
"""
|
||
return password.lower() in _COMMON_PASSWORDS
|
||
|
||
|
||
def _validate_strong_password(value: str) -> str:
|
||
"""Pydantic field-validator body shared by Register + ChangePassword.
|
||
|
||
Constraint = function, not type-level mixin. The two request models
|
||
have no "is-a" relationship; they only share the password-strength
|
||
rule. Lifting it into a free function lets each model bind it via
|
||
``@field_validator(field_name)`` without inheritance gymnastics.
|
||
"""
|
||
if _password_is_common(value):
|
||
raise ValueError("密码过于常见，请设置更安全的密码")
|
||
if value != value.strip():
|
||
raise ValueError("密码首尾不能包含空格")
|
||
return value
|
||
|
||
|
||
class RegisterRequest(BaseModel):
|
||
"""Request model for user registration."""
|
||
|
||
username: str | None = Field(default=None, min_length=1, max_length=128)
|
||
email: EmailStr | None = None
|
||
password: str = Field(..., min_length=8, max_length=255)
|
||
model_config = ConfigDict(extra="forbid")
|
||
|
||
_strong_password = field_validator("password")(classmethod(lambda cls, v: _validate_strong_password(v)))
|
||
|
||
@model_validator(mode="after")
|
||
def validate_identity(self):
|
||
if bool(self.username) == bool(self.email):
|
||
raise ValueError("请填写用户名或邮箱（只需一项）")
|
||
if self.username is not None:
|
||
value = self.username.strip()
|
||
if "@" not in value and (not value or value[0] in "._-" or value[-1] in "._-" or any(not (ch.isalnum() or ch in "._-") for ch in value)):
|
||
raise ValueError("用户名仅支持字母、数字、点、下划线和短横线，首尾须为字母或数字")
|
||
self.username = value
|
||
if "@" in value:
|
||
TypeAdapter(EmailStr).validate_python(value)
|
||
return self
|
||
|
||
|
||
class UsernameLoginRequest(BaseModel):
|
||
"""Request model for passwordless username login.
|
||
|
||
Behavior:
|
||
- Existing user: direct login without password
|
||
- Missing user: auto-register and login without password
|
||
"""
|
||
|
||
username: str = Field(..., min_length=1, max_length=128)
|
||
|
||
|
||
class UsernameLoginResponse(BaseModel):
|
||
"""Response model for passwordless username login.
|
||
|
||
Returns the access token directly in the body so non-browser clients
|
||
can use it as a Bearer credential, while a session cookie is still set
|
||
for browser-based callers.
|
||
"""
|
||
|
||
access_token: str
|
||
token_type: str = "bearer"
|
||
expires_in: int
|
||
user_id: str
|
||
email: str
|
||
system_role: Literal["admin", "user"] = "user"
|
||
needs_setup: bool = False
|
||
created: bool = Field(default=False, description="True when the user was auto-registered by this call")
|
||
username: str | None = Field(
|
||
default=None,
|
||
description=(
|
||
"The login username this call resolved (token login: the value read "
|
||
"from getTokenInfo; username login: the supplied username). Used by "
|
||
"the frontend to append ?username= on task-workspace jump links; the "
|
||
"stored email local-part is lossy (lowercased/normalized)."
|
||
),
|
||
)
|
||
|
||
|
||
class ChangePasswordRequest(BaseModel):
|
||
"""Request model for password change (also handles setup flow)."""
|
||
|
||
current_password: str = Field(default="", max_length=255)
|
||
new_password: str = Field(..., min_length=8, max_length=255)
|
||
new_email: EmailStr | None = None
|
||
|
||
_strong_password = field_validator("new_password")(classmethod(lambda cls, v: _validate_strong_password(v)))
|
||
|
||
|
||
class MessageResponse(BaseModel):
|
||
"""Generic message response."""
|
||
|
||
message: str
|
||
|
||
|
||
class PasswordChangeResponse(MessageResponse):
|
||
access_token: str
|
||
token_type: str = "bearer"
|
||
expires_in: int
|
||
|
||
|
||
# ── Helpers ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _normalize_username_local_part(username: str) -> str | None:
|
||
"""Return the canonical local-part for a username, or None if user typed an email.
|
||
|
||
Used both to build the synthetic email and to look up legacy accounts that
|
||
were registered with a different default domain. Returning ``None`` for
|
||
raw-email input tells callers to skip the domain-agnostic fallback —
|
||
if you typed ``alice@old.com``, you meant that specific row.
|
||
"""
|
||
raw = username.strip()
|
||
if not raw:
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名不能为空")
|
||
if "@" in raw:
|
||
return None
|
||
normalized = "".join(ch if (ch.isalnum() or ch in "._-") else "-" for ch in raw.lower()).strip("._-")
|
||
if not normalized:
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名不合法")
|
||
return normalized
|
||
|
||
|
||
def _username_to_email(username: str) -> str:
|
||
"""Map a frontend username to a valid email used by user storage.
|
||
|
||
- If username already looks like an email (`contains @`), use it directly.
|
||
- Otherwise generate `<normalized-username>@<domain>`.
|
||
"""
|
||
local = _normalize_username_local_part(username)
|
||
if local is None:
|
||
return username.strip().lower()
|
||
|
||
# Default mail domain for passwordless / username-only registrations.
|
||
# Kept overridable via env so individual deployments can swap it out, but the
|
||
# built-in default targets the cm.com tenant.
|
||
domain = os.getenv("DEER_FLOW_USERNAME_LOGIN_DOMAIN", "cm.com").strip().lower() or "cm.com"
|
||
return f"{local}@{domain}"
|
||
|
||
|
||
# 后台岗位同步任务的强引用集合：防止 create_task 的任务被 GC 中途回收。
|
||
_position_sync_tasks: set = set()
|
||
_POSITION_SYNC_CACHE_SECONDS = 60.0
|
||
_MAX_POSITION_SYNC_CACHE_USERS = 10_000
|
||
_position_sync_completed_at: dict[str, float] = {}
|
||
_position_sync_inflight: dict[str, asyncio.Task[None]] = {}
|
||
|
||
|
||
def _sync_user_position(request: Request, user_id: str) -> None:
|
||
"""Best-effort: schedule the user's 岗位 (position) sync off the login path.
|
||
|
||
Ensures a user who has never been synced — e.g. a brand-new account that
|
||
inherits the ``is_default`` position — gets the default agents / tasks /
|
||
prompts loaded into their "我的" tables. The sync is idempotent but touches
|
||
a dozen tables per user, so it runs as a fire-and-forget task instead of
|
||
blocking the login response; the materialized rows land a moment later and
|
||
failures are logged, never surfaced.
|
||
"""
|
||
normalized_user_id = str(user_id)
|
||
now = time.monotonic()
|
||
completed_at = _position_sync_completed_at.get(normalized_user_id)
|
||
if completed_at is not None and now - completed_at < _POSITION_SYNC_CACHE_SECONDS:
|
||
return
|
||
if normalized_user_id in _position_sync_inflight:
|
||
return
|
||
|
||
sync = getattr(request.app.state, "position_sync", None)
|
||
if sync is None:
|
||
return
|
||
|
||
async def _run() -> None:
|
||
try:
|
||
await sync.sync_user(normalized_user_id)
|
||
_position_sync_completed_at[normalized_user_id] = time.monotonic()
|
||
if len(_position_sync_completed_at) > _MAX_POSITION_SYNC_CACHE_USERS:
|
||
oldest = sorted(_position_sync_completed_at, key=_position_sync_completed_at.get)
|
||
for stale_user_id in oldest[: len(oldest) // 2]:
|
||
_position_sync_completed_at.pop(stale_user_id, None)
|
||
except Exception:
|
||
logger.debug("position sync on login failed for user %s", user_id, exc_info=True)
|
||
finally:
|
||
_position_sync_inflight.pop(normalized_user_id, None)
|
||
|
||
try:
|
||
task = asyncio.create_task(_run())
|
||
except RuntimeError:
|
||
# 防御：请求处理器内必有运行中的事件循环，这里理论上走不到。
|
||
logger.debug("position sync scheduling failed for user %s", user_id, exc_info=True)
|
||
return
|
||
_position_sync_tasks.add(task)
|
||
_position_sync_inflight[normalized_user_id] = task
|
||
task.add_done_callback(_position_sync_tasks.discard)
|
||
|
||
|
||
async def _get_or_create_user_by_username(
|
||
username: str,
|
||
*,
|
||
existing_user: User | None | object = _USER_NOT_LOOKED_UP,
|
||
):
|
||
"""Resolve a username to a local user, auto-registering when absent.
|
||
|
||
Shared by ``/login/username`` and ``/login/token`` — both turn an
|
||
externally-supplied username into a local account and sign it in. Returns
|
||
``(user, created)`` where ``created`` is True only when this call inserted
|
||
the row. Roles are persisted authorization data, never inferred from names.
|
||
"""
|
||
provider = get_local_provider()
|
||
email = _username_to_email(username)
|
||
local_part = _normalize_username_local_part(username)
|
||
|
||
created = False
|
||
# A bare username is globally unique across synthetic email domains. Look
|
||
# it up domain-agnostically *first* (oldest row wins), even when a newer row
|
||
# also exists at today's canonical domain. Choosing the exact canonical
|
||
# row first made the same username alternate backend user_ids after a
|
||
# domain migration, which in turn made its existing threads inaccessible.
|
||
if existing_user is not _USER_NOT_LOOKED_UP:
|
||
# /login/username already performed this lookup to validate the
|
||
# account-specific password. Reuse both hits and misses instead of
|
||
# issuing the same MySQL query twice during a login burst.
|
||
user = existing_user
|
||
elif local_part is not None:
|
||
user = await provider.find_user_by_email_local_part(local_part)
|
||
if user is not None and user.email != email:
|
||
logger.info("Matched username %r to stable legacy account %s", username, user.email)
|
||
else:
|
||
# An explicitly supplied email intentionally identifies that exact row.
|
||
user = await provider.get_user_by_email(email)
|
||
if user is None:
|
||
try:
|
||
# 黑名单模式:从未登录过的新用户(此处即首次自动注册)默认放行(approved=True)。
|
||
# 管理员可在白名单管理页对个别用户「取消放行」来封禁——仅当登录白名单开关打开时生效。
|
||
user = await provider.create_user(email=email, password=None, system_role="user", needs_setup=False, approved=True)
|
||
created = True
|
||
logger.info("Auto-registered passwordless user: %s", email)
|
||
except ValueError:
|
||
# Concurrent first login may create the same user in parallel.
|
||
user = await provider.get_user_by_email(email)
|
||
if user is None and local_part is not None:
|
||
user = await provider.find_user_by_email_local_part(local_part)
|
||
if user is None:
|
||
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="该用户名已存在，请使用原账号登录")
|
||
return user, created
|
||
|
||
|
||
async def _lookup_user_by_username(username: str) -> User | None:
|
||
"""Resolve a username to an **existing** local user, without creating one.
|
||
|
||
Mirrors the lookup half of ``_get_or_create_user_by_username`` (canonical
|
||
``<username>@<domain>`` email + legacy local-part fallback) so the
|
||
per-account 登录口令 gate can read the target account's ``login_password``
|
||
before deciding whether to admit the login. Returns ``None`` for a username
|
||
that has never registered (the personal-password gate then simply does not
|
||
apply, and ``_get_or_create_user_by_username`` later auto-registers it).
|
||
"""
|
||
provider = get_local_provider()
|
||
email = _username_to_email(username)
|
||
local_part = _normalize_username_local_part(username)
|
||
if local_part is not None:
|
||
return await provider.find_user_by_email_local_part(local_part)
|
||
return await provider.get_user_by_email(email)
|
||
|
||
|
||
def enforce_user_approved(user) -> None:
|
||
"""登录白名单拦截(**黑名单模式**):开关打开且用户被「取消放行」时拒绝登录。
|
||
|
||
新账号首次登录即默认放行(``approved=True``,见 ``_get_or_create_user_by_username``
|
||
与 ``create_user``),所以从未在本系统登录过的用户可以直接使用;白名单开关只用来
|
||
封禁管理员显式「取消放行」(``approved=False``) 的个别用户。
|
||
|
||
规则(三条硬约束防锁死):
|
||
- 开关关(``login_whitelist.enabled`` 默认 False)→ 直接放行,人人可登录。
|
||
- admin → 永远放行(避免管理员把自己关在门外)。
|
||
- 其余:``approved`` 为 False(被管理员取消放行)→ 403 ``PENDING_APPROVAL``;
|
||
前端据此提示「等待管理员开通」,且不写任何登录态。
|
||
|
||
放在各登录端点拿到 user 之后、签发 JWT 之前调用。
|
||
"""
|
||
try:
|
||
enabled = load_system_settings().login_whitelist.enabled
|
||
except Exception:
|
||
# 读配置失败时绝不误伤(白名单是附加闸门,故障时回到「不拦截」)。
|
||
logger.warning("读取登录白名单配置失败,本次登录不拦截", exc_info=True)
|
||
return
|
||
if not enabled:
|
||
return
|
||
if getattr(user, "system_role", None) == "admin":
|
||
return
|
||
if not getattr(user, "approved", False):
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail={"code": "PENDING_APPROVAL", "message": "您的账号已被限制登录,请联系管理员开通"},
|
||
)
|
||
|
||
|
||
async def _fetch_username_from_token(token: str) -> str:
|
||
"""Resolve an upstream access token to a username via the configured endpoint.
|
||
|
||
Calls ``auth_login.token_login.token_info_url`` with the caller-supplied
|
||
token placed directly in the ``Authorization: Bearer <token>`` header — the
|
||
upstream reads the bearer token from there, not from a cookie. Expects
|
||
``{"state": "200", "data": {"username": ...}}`` and returns ``data.username``.
|
||
Any upstream failure, non-200 ``state``, or missing username raises HTTP 401.
|
||
"""
|
||
config = get_app_config().auth_login.token_login
|
||
|
||
# Offline test shortcut: a token listed in ``mock_tokens`` resolves to its
|
||
# mapped username without ever calling the (possibly unreachable) upstream
|
||
# ``token_info_url``. Lets token login be exercised end-to-end where the real
|
||
# getTokenInfo host can't be reached (external networks). Empty in prod.
|
||
mocked = config.mock_tokens.get(token)
|
||
if mocked and str(mocked).strip():
|
||
return str(mocked).strip()
|
||
|
||
if not config.token_info_url:
|
||
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="token 登录未配置(token_info_url 为空)")
|
||
|
||
# TLS verification for the (HTTPS) upstream. A custom CA bundle wins; else
|
||
# the boolean toggle. UAT/internal hosts often need verify_ssl=false because
|
||
# they serve self-signed or private-CA certs — without this the call dies
|
||
# with an SSLError and every token login silently 401s.
|
||
verify: bool | str = config.ca_cert_path if config.ca_cert_path else config.verify_ssl
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=config.timeout_seconds, verify=verify) as client:
|
||
upstream = await client.post(
|
||
config.token_info_url,
|
||
headers={
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
},
|
||
)
|
||
except httpx.HTTPError as exc:
|
||
# Surface the concrete failure (SSL handshake, DNS, timeout, ...) in logs
|
||
# so HTTPS/cert misconfiguration is diagnosable; client still sees 401.
|
||
logger.warning(
|
||
"Token login: upstream call to %s failed (verify=%r): %s: %s",
|
||
config.token_info_url,
|
||
verify,
|
||
type(exc).__name__,
|
||
exc,
|
||
)
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="token 校验失败").model_dump())
|
||
|
||
if upstream.status_code != 200:
|
||
logger.warning("Token login: upstream returned HTTP %s", upstream.status_code)
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="token 无效").model_dump())
|
||
|
||
try:
|
||
payload = upstream.json()
|
||
except ValueError:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="token 接口返回格式异常").model_dump())
|
||
|
||
# Upstream signals success with the string state "200" (see API docs).
|
||
if str(payload.get("state")) != "200":
|
||
logger.warning("Token login: upstream state=%r msg=%r", payload.get("state"), payload.get("msg"))
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="token 无效或已过期").model_dump())
|
||
|
||
username = (payload.get("data") or {}).get("username")
|
||
if not username or not str(username).strip():
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="未能从 token 解析出用户名").model_dump())
|
||
|
||
return str(username).strip()
|
||
|
||
|
||
def _set_session_cookie(response: Response, token: str, request: Request) -> None:
|
||
"""Set the access_token HttpOnly cookie on the response."""
|
||
config = get_auth_config()
|
||
is_https = is_secure_request(request)
|
||
response.set_cookie(
|
||
key="access_token",
|
||
value=token,
|
||
httponly=True,
|
||
secure=is_https,
|
||
samesite="lax",
|
||
max_age=config.token_expiry_days * 24 * 3600 if is_https else None,
|
||
)
|
||
|
||
|
||
# ── Rate Limiting ────────────────────────────────────────────────────────
|
||
# In-process dict — not shared across workers.
|
||
#
|
||
# **Limitation**: with multi-worker deployments (e.g., gunicorn -w N), each
|
||
# worker maintains its own lockout table, so an attacker effectively gets
|
||
# N × _MAX_LOGIN_ATTEMPTS guesses before being locked out everywhere. For
|
||
# production multi-worker setups, replace this with a shared store (Redis,
|
||
# database-backed counter) to enforce a true per-IP limit.
|
||
|
||
_MAX_LOGIN_ATTEMPTS = 100
|
||
_LOCKOUT_SECONDS = 300 # 5 minutes
|
||
|
||
# ip → (fail_count, lock_until_timestamp)
|
||
_login_attempts: dict[str, tuple[int, float]] = {}
|
||
|
||
|
||
def _trusted_proxies() -> list:
|
||
"""Parse ``AUTH_TRUSTED_PROXIES`` env var into a list of ip_network objects.
|
||
|
||
Comma-separated CIDR or single-IP entries. Empty / unset = no proxy is
|
||
trusted (direct mode). Invalid entries are skipped with a logger warning.
|
||
Read live so env-var overrides take effect immediately and tests can
|
||
``monkeypatch.setenv`` without poking a module-level cache.
|
||
"""
|
||
raw = os.getenv("AUTH_TRUSTED_PROXIES", "").strip()
|
||
if not raw:
|
||
return []
|
||
nets = []
|
||
for entry in raw.split(","):
|
||
entry = entry.strip()
|
||
if not entry:
|
||
continue
|
||
try:
|
||
nets.append(ip_network(entry, strict=False))
|
||
except ValueError:
|
||
logger.warning("AUTH_TRUSTED_PROXIES: ignoring invalid entry %r", entry)
|
||
return nets
|
||
|
||
|
||
def _get_client_ip(request: Request) -> str:
|
||
"""Extract the real client IP for rate limiting.
|
||
|
||
Trust model:
|
||
|
||
- The TCP peer (``request.client.host``) is always the baseline. It is
|
||
whatever the kernel reports as the connecting socket — unforgeable
|
||
by the client itself.
|
||
- ``X-Real-IP`` is **only** honored if the TCP peer is in the
|
||
``AUTH_TRUSTED_PROXIES`` allowlist (set via env var, comma-separated
|
||
CIDR or single IPs). When set, the gateway is assumed to be behind a
|
||
reverse proxy (nginx, Cloudflare, ALB, …) that overwrites
|
||
``X-Real-IP`` with the original client address.
|
||
- With no ``AUTH_TRUSTED_PROXIES`` set, ``X-Real-IP`` is silently
|
||
ignored — closing the bypass where any client could rotate the
|
||
header to dodge per-IP rate limits in dev / direct-gateway mode.
|
||
|
||
``X-Forwarded-For`` is intentionally NOT used because it is naturally
|
||
client-controlled at the *first* hop and the trust chain is harder to
|
||
audit per-request.
|
||
"""
|
||
peer_host = request.client.host if request.client else None
|
||
|
||
trusted = _trusted_proxies()
|
||
if trusted and peer_host:
|
||
try:
|
||
peer_ip = ip_address(peer_host)
|
||
if any(peer_ip in net for net in trusted):
|
||
real_ip = request.headers.get("x-real-ip", "").strip()
|
||
if real_ip:
|
||
return real_ip
|
||
except ValueError:
|
||
# peer_host wasn't a parseable IP (e.g. "unknown") — fall through
|
||
pass
|
||
|
||
return peer_host or "unknown"
|
||
|
||
|
||
def _check_rate_limit(ip: str) -> None:
|
||
"""Raise 429 if the IP is currently locked out."""
|
||
record = _login_attempts.get(ip)
|
||
if record is None:
|
||
return
|
||
fail_count, lock_until = record
|
||
if fail_count >= _MAX_LOGIN_ATTEMPTS:
|
||
if time.time() < lock_until:
|
||
raise HTTPException(
|
||
status_code=429,
|
||
detail="登录尝试次数过多,请稍后再试",
|
||
)
|
||
del _login_attempts[ip]
|
||
|
||
|
||
_MAX_TRACKED_IPS = 10000
|
||
|
||
|
||
def _record_login_failure(ip: str) -> None:
|
||
"""Record a failed login attempt for the given IP."""
|
||
# Evict expired lockouts when dict grows too large
|
||
if len(_login_attempts) >= _MAX_TRACKED_IPS:
|
||
now = time.time()
|
||
expired = [k for k, (c, t) in _login_attempts.items() if c >= _MAX_LOGIN_ATTEMPTS and now >= t]
|
||
for k in expired:
|
||
del _login_attempts[k]
|
||
# If still too large, evict cheapest-to-lose half: below-threshold
|
||
# IPs (lock_until=0.0) sort first, then earliest-expiring lockouts.
|
||
if len(_login_attempts) >= _MAX_TRACKED_IPS:
|
||
by_time = sorted(_login_attempts.items(), key=lambda kv: kv[1][1])
|
||
for k, _ in by_time[: len(by_time) // 2]:
|
||
del _login_attempts[k]
|
||
|
||
record = _login_attempts.get(ip)
|
||
if record is None:
|
||
_login_attempts[ip] = (1, 0.0)
|
||
else:
|
||
new_count = record[0] + 1
|
||
lock_until = time.time() + _LOCKOUT_SECONDS if new_count >= _MAX_LOGIN_ATTEMPTS else 0.0
|
||
_login_attempts[ip] = (new_count, lock_until)
|
||
|
||
|
||
def _record_login_success(ip: str) -> None:
|
||
"""Clear failure counter for the given IP on successful login."""
|
||
_login_attempts.pop(ip, None)
|
||
|
||
|
||
# ── Endpoints ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _require_login_method(method: str) -> None:
|
||
settings = load_system_settings().authentication
|
||
if not getattr(settings, method):
|
||
labels = {"password_login_enabled": "账号密码登录", "registration_enabled": "自助注册", "sso_enabled": "外部系统免登"}
|
||
raise HTTPException(status_code=403, detail=f"管理员尚未开启{labels[method]}")
|
||
|
||
|
||
def _login_response(user: User, token: str, *, username: str | None = None, created: bool = False) -> UsernameLoginResponse:
|
||
return UsernameLoginResponse(
|
||
access_token=token,
|
||
expires_in=get_auth_config().token_expiry_days * 24 * 3600,
|
||
user_id=str(user.id),
|
||
email=user.email,
|
||
system_role=user.system_role,
|
||
needs_setup=user.needs_setup,
|
||
created=created,
|
||
username=username or user.email.split("@", 1)[0],
|
||
)
|
||
|
||
|
||
async def _prepare_login_session(user: User) -> User:
|
||
"""Apply the administrator's one-online-session policy after authentication."""
|
||
if not load_system_settings().authentication.single_session_enabled:
|
||
return user
|
||
try:
|
||
return await get_local_provider().start_login_session(user)
|
||
except CredentialConflictError as exc:
|
||
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
||
|
||
|
||
class AuthenticationOptions(AuthenticationSettings):
|
||
needs_setup: bool = False
|
||
|
||
|
||
@router.get("/options", response_model=AuthenticationOptions)
|
||
async def authentication_options(request: Request, response: Response) -> AuthenticationOptions:
|
||
"""Anonymous, non-secret capability discovery for the login page."""
|
||
response.headers["Cache-Control"] = "no-store"
|
||
trace = trace_request(request, "auth_options")
|
||
try:
|
||
stage_started = time.perf_counter()
|
||
authentication = load_system_settings().authentication.model_dump()
|
||
trace_stage(trace, "load_system_settings", stage_started)
|
||
|
||
stage_started = time.perf_counter()
|
||
needs_setup = await get_local_provider().count_admin_users() == 0
|
||
trace_stage(trace, "count_admin_users", stage_started)
|
||
return AuthenticationOptions(**authentication, needs_setup=needs_setup)
|
||
finally:
|
||
finish_trace(trace)
|
||
|
||
|
||
@router.post("/login/local", response_model=UsernameLoginResponse)
|
||
async def login_local(
|
||
request: Request,
|
||
response: Response,
|
||
form_data: OAuth2PasswordRequestForm = Depends(),
|
||
):
|
||
"""Username or email login with the account's unified password/passphrase."""
|
||
_require_login_method("password_login_enabled")
|
||
client_ip = _get_client_ip(request)
|
||
_check_rate_limit(client_ip)
|
||
|
||
user = await _lookup_user_by_username(form_data.username)
|
||
if user is not None and not await verify_account_password(form_data.password, user):
|
||
user = None
|
||
|
||
if user is None:
|
||
_record_login_failure(client_ip)
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="账号或密码错误").model_dump(),
|
||
)
|
||
|
||
_record_login_success(client_ip)
|
||
enforce_user_approved(user)
|
||
user = await _prepare_login_session(user)
|
||
token = create_access_token(str(user.id), token_version=user.token_version, auth_method="password")
|
||
_set_session_cookie(response, token, request)
|
||
_sync_user_position(request, user.id)
|
||
|
||
return _login_response(user, token, username=form_data.username.strip())
|
||
|
||
|
||
def _username_login_request(username: object) -> UsernameLoginRequest:
|
||
if not isinstance(username, str):
|
||
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="username is required")
|
||
try:
|
||
return UsernameLoginRequest(username=username)
|
||
except ValidationError as exc:
|
||
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="username is required") from exc
|
||
|
||
|
||
async def _read_username_login(request: Request) -> UsernameLoginRequest:
|
||
"""Accept the historical JSON body and a form body.
|
||
|
||
``application/x-www-form-urlencoded`` is a CORS simple content type. Old
|
||
Firefox aborts a credentialed JSON login during preflight and reports
|
||
NS_BINDING_ABORTED, so the browser client sends a form body instead.
|
||
"""
|
||
media_type = request.headers.get("content-type", "").split(";", 1)[0].strip().lower()
|
||
if media_type == "application/x-www-form-urlencoded":
|
||
form = await request.form()
|
||
return _username_login_request(form.get("username"))
|
||
try:
|
||
payload = await request.json()
|
||
except json.JSONDecodeError as exc:
|
||
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="username is required") from exc
|
||
username = payload.get("username") if isinstance(payload, dict) else None
|
||
return _username_login_request(username)
|
||
|
||
|
||
@router.post("/login/username", response_model=UsernameLoginResponse)
|
||
async def login_by_username(
|
||
request: Request,
|
||
response: Response,
|
||
password: str | None = Query(default=None, description="Account password or shared passphrase; required when password login is enabled."),
|
||
):
|
||
"""Compatibility username login, gated by the current credential policy.
|
||
|
||
Personal password/passphrase wins. With password login enabled, accounts
|
||
without personal credentials must supply the existing shared passphrase
|
||
(default 123ewq). Identity-only login is retained only while password login
|
||
is disabled and the historical shared-password gate is also disabled.
|
||
"""
|
||
body = await _read_username_login(request)
|
||
_require_login_method("sso_enabled")
|
||
username_login_config = get_app_config().auth_login.username_login
|
||
password_login_enabled = load_system_settings().authentication.password_login_enabled
|
||
|
||
# Per-account password takes precedence: look the account up (no create) and,
|
||
# if it carries a personal 登录口令, require an exact match — ignoring the
|
||
# shared gate entirely for that account.
|
||
existing = await _lookup_user_by_username(body.username)
|
||
credentials_verified = False
|
||
if has_account_password(existing) or password_login_enabled or username_login_config.require_password:
|
||
client_ip = _get_client_ip(request)
|
||
_check_rate_limit(client_ip)
|
||
if not await verify_account_password(password or "", existing):
|
||
_record_login_failure(client_ip)
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="口令错误").model_dump(),
|
||
)
|
||
credentials_verified = True
|
||
_record_login_success(client_ip)
|
||
|
||
user, created = await _get_or_create_user_by_username(body.username, existing_user=existing)
|
||
|
||
# A registration/reset may commit after the initial lookup/password gate.
|
||
# Never mint the new account/version's session using an earlier decision.
|
||
if has_account_password(user) and (existing is None or existing.id != user.id or existing.token_version != user.token_version):
|
||
client_ip = _get_client_ip(request)
|
||
_check_rate_limit(client_ip)
|
||
if not await verify_account_password(password or "", user):
|
||
_record_login_failure(client_ip)
|
||
raise HTTPException(status_code=401, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="口令错误").model_dump())
|
||
credentials_verified = True
|
||
|
||
enforce_user_approved(user)
|
||
user = await _prepare_login_session(user)
|
||
token = create_access_token(str(user.id), token_version=user.token_version, auth_method="passphrase" if credentials_verified else None)
|
||
_set_session_cookie(response, token, request)
|
||
_sync_user_position(request, user.id)
|
||
|
||
return UsernameLoginResponse(
|
||
access_token=token,
|
||
token_type="bearer",
|
||
expires_in=get_auth_config().token_expiry_days * 24 * 3600,
|
||
user_id=str(user.id),
|
||
email=user.email,
|
||
system_role=user.system_role,
|
||
needs_setup=user.needs_setup,
|
||
created=created,
|
||
username=body.username,
|
||
)
|
||
|
||
|
||
@router.post("/login/token", response_model=UsernameLoginResponse)
|
||
async def login_by_token(
|
||
request: Request,
|
||
response: Response,
|
||
token: str = Query(..., min_length=1, description="Upstream access token to exchange for a username."),
|
||
):
|
||
"""Token-exchange login.
|
||
|
||
Verifies an upstream access token by calling the configured
|
||
``token_info_url`` (see ``auth_login.token_login``), reads the resolved
|
||
username, then signs the matching local account in — auto-registering it
|
||
when the username is not yet in the database. Gated by
|
||
``auth_login.token_login.enabled``.
|
||
|
||
Returns the JWT access token in the response body and also sets it as a
|
||
session cookie for browser callers.
|
||
"""
|
||
_require_login_method("sso_enabled")
|
||
config = get_app_config().auth_login.token_login
|
||
if not config.enabled:
|
||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="token 登录未开启")
|
||
|
||
client_ip = _get_client_ip(request)
|
||
_check_rate_limit(client_ip)
|
||
|
||
try:
|
||
username = await _fetch_username_from_token(token)
|
||
except HTTPException as exc:
|
||
if exc.status_code == status.HTTP_401_UNAUTHORIZED:
|
||
_record_login_failure(client_ip)
|
||
raise
|
||
|
||
_record_login_success(client_ip)
|
||
|
||
user, created = await _get_or_create_user_by_username(username)
|
||
|
||
enforce_user_approved(user)
|
||
user = await _prepare_login_session(user)
|
||
access_token = create_access_token(str(user.id), token_version=user.token_version, auth_method="sso")
|
||
_set_session_cookie(response, access_token, request)
|
||
_sync_user_position(request, user.id)
|
||
|
||
return UsernameLoginResponse(
|
||
access_token=access_token,
|
||
token_type="bearer",
|
||
expires_in=get_auth_config().token_expiry_days * 24 * 3600,
|
||
user_id=str(user.id),
|
||
email=user.email,
|
||
system_role=user.system_role,
|
||
needs_setup=user.needs_setup,
|
||
created=created,
|
||
# The raw username resolved from getTokenInfo (pre email-normalization),
|
||
# so the frontend can append the exact ?username= on jump links.
|
||
username=username,
|
||
)
|
||
|
||
|
||
@router.post("/register", response_model=UsernameLoginResponse, status_code=status.HTTP_201_CREATED)
|
||
async def register(request: Request, response: Response, body: RegisterRequest):
|
||
"""Register a new user account (always 'user' role).
|
||
|
||
Admin is auto-created on first boot. This endpoint creates regular users.
|
||
Auto-login by setting the session cookie.
|
||
"""
|
||
_require_login_method("password_login_enabled")
|
||
_require_login_method("registration_enabled")
|
||
client_ip = _get_client_ip(request)
|
||
_check_rate_limit(client_ip)
|
||
provider = get_local_provider()
|
||
username = body.username or str(body.email)
|
||
email = _username_to_email(username)
|
||
# Match every domain, even for explicit emails: a self-registration must
|
||
# never shadow the stable account selected by bare-username login.
|
||
duplicate = await provider.find_user_by_email_local_part(email.split("@", 1)[0])
|
||
try:
|
||
if duplicate is not None:
|
||
raise ValueError("Duplicate username")
|
||
user = await provider.create_user(email=email, password=body.password, system_role="user", approved=True)
|
||
except ValueError:
|
||
_record_login_failure(client_ip)
|
||
raise HTTPException(
|
||
status_code=status.HTTP_409_CONFLICT,
|
||
detail=AuthErrorResponse(code=AuthErrorCode.EMAIL_ALREADY_EXISTS, message="该用户名或邮箱已注册").model_dump(),
|
||
)
|
||
|
||
enforce_user_approved(user)
|
||
user = await _prepare_login_session(user)
|
||
token = create_access_token(str(user.id), token_version=user.token_version, auth_method="password")
|
||
_set_session_cookie(response, token, request)
|
||
_sync_user_position(request, user.id)
|
||
|
||
return _login_response(user, token, username=username, created=True)
|
||
|
||
|
||
@router.post("/logout", response_model=MessageResponse)
|
||
async def logout(request: Request, response: Response):
|
||
"""Logout current user by clearing the cookie."""
|
||
response.delete_cookie(key="access_token", secure=is_secure_request(request), samesite="lax")
|
||
return MessageResponse(message="Successfully logged out")
|
||
|
||
|
||
@router.post("/change-password", response_model=PasswordChangeResponse)
|
||
async def change_password(request: Request, response: Response, body: ChangePasswordRequest):
|
||
"""Change password for the currently authenticated user.
|
||
|
||
Also handles the first-boot setup flow:
|
||
- If new_email is provided, updates email (checks uniqueness)
|
||
- If user.needs_setup is True and new_email is given, clears needs_setup
|
||
- Always increments token_version to invalidate old sessions
|
||
- Re-issues session cookie with new token_version
|
||
"""
|
||
user = await get_current_user_from_request(request)
|
||
client_ip = _get_client_ip(request)
|
||
_check_rate_limit(client_ip)
|
||
if not await verify_account_password(body.current_password, user):
|
||
_record_login_failure(client_ip)
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="当前密码错误").model_dump())
|
||
|
||
provider = get_local_provider()
|
||
|
||
# Update email if provided
|
||
if body.new_email is not None:
|
||
existing = await provider.find_user_by_email_local_part(body.new_email.split("@", 1)[0])
|
||
if existing and str(existing.id) != str(user.id):
|
||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=AuthErrorResponse(code=AuthErrorCode.EMAIL_ALREADY_EXISTS, message="Email already in use").model_dump())
|
||
try:
|
||
user = await provider.update_credentials(user, await hash_password_async(body.new_password), email=body.new_email)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=409, detail="账号信息已变更，请刷新后重试") from exc
|
||
_record_login_success(client_ip)
|
||
|
||
# Re-issue cookie with new token_version
|
||
token = create_access_token(str(user.id), token_version=user.token_version, auth_method="password")
|
||
_set_session_cookie(response, token, request)
|
||
|
||
return PasswordChangeResponse(message="密码已更新", access_token=token, expires_in=get_auth_config().token_expiry_days * 24 * 3600)
|
||
|
||
|
||
@router.get("/me", response_model=UserResponse)
|
||
async def get_me(request: Request):
|
||
"""Get current authenticated user info."""
|
||
user = await get_current_user_from_request(request)
|
||
return UserResponse(
|
||
id=str(user.id),
|
||
email=user.email,
|
||
system_role=user.system_role,
|
||
needs_setup=user.needs_setup,
|
||
has_password=has_account_password(user),
|
||
requires_current_password=True,
|
||
uses_default_password=not has_account_password(user),
|
||
)
|
||
|
||
|
||
_SETUP_STATUS_COOLDOWN: dict[str, float] = {}
|
||
_SETUP_STATUS_COOLDOWN_SECONDS = 60
|
||
_MAX_TRACKED_SETUP_STATUS_IPS = 10000
|
||
|
||
|
||
@router.get("/setup-status")
|
||
async def setup_status(request: Request):
|
||
"""Check if an admin account exists. Returns needs_setup=True when no admin exists."""
|
||
client_ip = _get_client_ip(request)
|
||
now = time.time()
|
||
last_check = _SETUP_STATUS_COOLDOWN.get(client_ip, 0)
|
||
elapsed = now - last_check
|
||
if elapsed < _SETUP_STATUS_COOLDOWN_SECONDS:
|
||
retry_after = max(1, int(_SETUP_STATUS_COOLDOWN_SECONDS - elapsed))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
|
||
detail="Setup status check is rate limited",
|
||
headers={"Retry-After": str(retry_after)},
|
||
)
|
||
# Evict stale entries when dict grows too large to bound memory usage.
|
||
if len(_SETUP_STATUS_COOLDOWN) >= _MAX_TRACKED_SETUP_STATUS_IPS:
|
||
cutoff = now - _SETUP_STATUS_COOLDOWN_SECONDS
|
||
stale = [k for k, t in _SETUP_STATUS_COOLDOWN.items() if t < cutoff]
|
||
for k in stale:
|
||
del _SETUP_STATUS_COOLDOWN[k]
|
||
# If still too large after evicting expired entries, remove oldest half.
|
||
if len(_SETUP_STATUS_COOLDOWN) >= _MAX_TRACKED_SETUP_STATUS_IPS:
|
||
by_time = sorted(_SETUP_STATUS_COOLDOWN.items(), key=lambda kv: kv[1])
|
||
for k, _ in by_time[: len(by_time) // 2]:
|
||
del _SETUP_STATUS_COOLDOWN[k]
|
||
_SETUP_STATUS_COOLDOWN[client_ip] = now
|
||
admin_count = await get_local_provider().count_admin_users()
|
||
return {"needs_setup": admin_count == 0}
|
||
|
||
|
||
class InitializeAdminRequest(BaseModel):
|
||
"""Request model for first-boot admin account creation."""
|
||
|
||
email: EmailStr
|
||
password: str = Field(..., min_length=8, max_length=255)
|
||
|
||
_strong_password = field_validator("password")(classmethod(lambda cls, v: _validate_strong_password(v)))
|
||
|
||
|
||
@router.post("/initialize", response_model=UsernameLoginResponse, status_code=status.HTTP_201_CREATED)
|
||
async def initialize_admin(request: Request, response: Response, body: InitializeAdminRequest):
|
||
"""Create the first admin account on initial system setup.
|
||
|
||
Only callable when no admin exists. Returns 409 Conflict if an admin
|
||
already exists.
|
||
|
||
On success, the admin account is created with ``needs_setup=False`` and
|
||
the session cookie is set.
|
||
"""
|
||
admin_count = await get_local_provider().count_admin_users()
|
||
if admin_count > 0:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_409_CONFLICT,
|
||
detail=AuthErrorResponse(code=AuthErrorCode.SYSTEM_ALREADY_INITIALIZED, message="System already initialized").model_dump(),
|
||
)
|
||
|
||
try:
|
||
user = await get_local_provider().create_initial_admin(email=str(body.email), password=body.password)
|
||
except ValueError:
|
||
# DB unique-constraint race: another concurrent request beat us.
|
||
raise HTTPException(
|
||
status_code=status.HTTP_409_CONFLICT,
|
||
detail=AuthErrorResponse(code=AuthErrorCode.SYSTEM_ALREADY_INITIALIZED, message="System already initialized").model_dump(),
|
||
)
|
||
|
||
user = await _prepare_login_session(user)
|
||
token = create_access_token(str(user.id), token_version=user.token_version, auth_method="password")
|
||
_set_session_cookie(response, token, request)
|
||
|
||
return _login_response(user, token, created=True)
|
||
|
||
|
||
# ── Admin Endpoints ───────────────────────────────────────────────────────
|
||
|
||
|
||
class UserListItem(BaseModel):
|
||
id: str
|
||
email: str
|
||
system_role: Literal["admin", "user"]
|
||
skill_count: int
|
||
agent_count: int
|
||
qa_count: int = 0 # 问答次数: number of runs (one run = one question)
|
||
approved: bool = True # 登录白名单放行标记(白名单管理页据此放行/取消)
|
||
login_password: str | None = None # Legacy plaintext only; never return hashes.
|
||
has_login_password: bool = False
|
||
access_token: str | None = None
|
||
token_type: str = "bearer"
|
||
expires_in: int | None = None
|
||
|
||
|
||
def _legacy_login_password(user: User) -> str | None:
|
||
value = user.login_password
|
||
return value if value and not is_password_hash(value) else None
|
||
|
||
|
||
class UserListPage(BaseModel):
|
||
"""Paged user list — returned when the caller passes ``limit``."""
|
||
|
||
items: list[UserListItem]
|
||
total: int
|
||
approved_count: int # approved 或 admin（搜索范围内）
|
||
limit: int
|
||
offset: int
|
||
|
||
|
||
_USER_LIST_TIMEOUT_SECONDS = 15
|
||
|
||
|
||
def _user_list_skill_store(request: Request, include_counts: bool = True) -> SkillStore | None:
|
||
return get_skill_store(request) if include_counts else None
|
||
|
||
|
||
def _user_list_agent_store(request: Request, include_counts: bool = True) -> AgentStore | None:
|
||
return get_agent_store(request) if include_counts else None
|
||
|
||
|
||
@router.get("/users", response_model=list[UserListItem] | UserListPage)
|
||
async def list_all_users(
|
||
request: Request,
|
||
include_counts: bool = Query(True, description="False=只查 users 表，跳过 runs 统计与 skill/agent 计数（白名单管理页用，避免慢查询）"),
|
||
q: str | None = Query(None, max_length=100, description="可选：按邮箱/用户ID 子串过滤（仅分页模式生效）"),
|
||
limit: int | None = Query(None, ge=1, le=200, description="传入即分页返回 UserListPage；不传保持旧的全量列表"),
|
||
offset: int = Query(0, ge=0),
|
||
skill_store: SkillStore | None = Depends(_user_list_skill_store),
|
||
agent_store: AgentStore | None = Depends(_user_list_agent_store),
|
||
):
|
||
"""List all users with their skill, agent and Q&A counts. Admin only.
|
||
|
||
Passing ``limit`` switches to a paged response that reads one page of the
|
||
users table plus two COUNT queries; omitting it keeps the legacy flat list.
|
||
"""
|
||
try:
|
||
async with asyncio.timeout(_USER_LIST_TIMEOUT_SECONDS if not include_counts else None):
|
||
user = await get_current_user_from_request(request)
|
||
if user.system_role != "admin":
|
||
raise HTTPException(status_code=403, detail="Admin only")
|
||
provider = get_local_provider()
|
||
users = (
|
||
await provider.list_users()
|
||
if limit is None
|
||
else await provider.list_users_page(q=q, limit=limit, offset=offset)
|
||
)
|
||
qa_counts: dict[str, int] = {}
|
||
# One recent-window GROUP BY for everyone's 问答次数. This keeps the admin
|
||
# user list from scanning the complete lifetime of the runs table.
|
||
if include_counts:
|
||
try:
|
||
qa_counts = await get_run_store(request).count_by_user(
|
||
since=recent_admin_query_cutoff()
|
||
)
|
||
except Exception:
|
||
logger.warning("Failed to load per-user run counts for user list", exc_info=True)
|
||
result = []
|
||
for u in users:
|
||
uid = str(u.id)
|
||
result.append(
|
||
UserListItem(
|
||
id=uid,
|
||
email=u.email,
|
||
system_role=u.system_role,
|
||
skill_count=(await skill_store.count_owned(uid)) if include_counts else 0,
|
||
agent_count=(await agent_store.count_owned(uid)) if include_counts else 0,
|
||
qa_count=qa_counts.get(uid, 0),
|
||
approved=getattr(u, "approved", True),
|
||
login_password=_legacy_login_password(u),
|
||
has_login_password=has_account_password(u),
|
||
)
|
||
)
|
||
if limit is None:
|
||
return result
|
||
total = await provider.count_users_filtered(q=q)
|
||
approved_count = await provider.count_users_filtered(q=q, approved_or_admin=True)
|
||
return UserListPage(items=result, total=total, approved_count=approved_count, limit=limit, offset=offset)
|
||
except TimeoutError as exc:
|
||
logger.warning("User list timed out waiting for account data")
|
||
raise HTTPException(status_code=503, detail="用户基础信息查询超时，请稍后重试") from exc
|
||
|
||
|
||
class UserApprovalRequest(BaseModel):
|
||
approved: bool = Field(..., description="True=放行该用户登录;False=取消放行(下次登录被拦)")
|
||
|
||
|
||
@router.put("/users/{user_id}/approval", response_model=UserListItem)
|
||
async def set_user_approval(user_id: str, body: UserApprovalRequest, request: Request) -> UserListItem:
|
||
"""放行 / 取消放行某个注册用户(登录白名单)。Admin only。
|
||
|
||
取消放行只拦该用户的下次登录(不动 ``token_version``,现有会话不强制下线)。
|
||
"""
|
||
admin = await get_current_user_from_request(request)
|
||
if admin.system_role != "admin":
|
||
raise HTTPException(status_code=403, detail="Admin only")
|
||
|
||
provider = get_local_provider()
|
||
target = await provider.get_user(user_id)
|
||
if target is None:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
target.approved = bool(body.approved)
|
||
try:
|
||
saved = await provider.update_user(target)
|
||
except CredentialConflictError as exc:
|
||
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
||
return UserListItem(
|
||
id=str(saved.id),
|
||
email=saved.email,
|
||
system_role=saved.system_role,
|
||
skill_count=0,
|
||
agent_count=0,
|
||
qa_count=0,
|
||
approved=saved.approved,
|
||
login_password=_legacy_login_password(saved),
|
||
has_login_password=has_account_password(saved),
|
||
)
|
||
|
||
|
||
class UserLoginPasswordRequest(BaseModel):
|
||
password: str | None = Field(
|
||
default=None,
|
||
max_length=255,
|
||
description="设置账号密码（与登录口令一致）；不允许清除后退回免密码登录。",
|
||
)
|
||
|
||
|
||
@router.put("/users/{user_id}/login-password", response_model=UserListItem)
|
||
@router.put("/users/{user_id}/reset-password", response_model=UserListItem)
|
||
async def set_user_login_password(user_id: str, body: UserLoginPasswordRequest, request: Request, response: Response) -> UserListItem:
|
||
"""Admin password reset; rotates both credential columns and all sessions."""
|
||
admin = await get_current_user_from_request(request)
|
||
if admin.system_role != "admin":
|
||
raise HTTPException(status_code=403, detail="Admin only")
|
||
|
||
provider = get_local_provider()
|
||
target = await provider.get_user(user_id)
|
||
if target is None:
|
||
raise HTTPException(status_code=404, detail="用户不存在")
|
||
|
||
pwd = body.password or ""
|
||
if not pwd:
|
||
raise HTTPException(status_code=400, detail="不能清除账号密码，请设置新密码")
|
||
if len(pwd) < 8:
|
||
raise HTTPException(status_code=400, detail="密码至少需要 8 个字符")
|
||
try:
|
||
_validate_strong_password(pwd)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||
try:
|
||
saved = await provider.update_credentials(target, await hash_password_async(pwd))
|
||
except CredentialConflictError as exc:
|
||
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
||
token = None
|
||
expires_in = None
|
||
if str(admin.id) == str(saved.id):
|
||
token = create_access_token(str(saved.id), token_version=saved.token_version, auth_method="password")
|
||
_set_session_cookie(response, token, request)
|
||
expires_in = get_auth_config().token_expiry_days * 24 * 3600
|
||
return UserListItem(
|
||
id=str(saved.id),
|
||
email=saved.email,
|
||
system_role=saved.system_role,
|
||
skill_count=0,
|
||
agent_count=0,
|
||
qa_count=0,
|
||
approved=saved.approved,
|
||
login_password=None,
|
||
has_login_password=True,
|
||
access_token=token,
|
||
expires_in=expires_in,
|
||
)
|
||
|
||
|
||
# ── OAuth Endpoints (Future/Placeholder) ─────────────────────────────────
|
||
|
||
|
||
@router.get("/oauth/{provider}")
|
||
async def oauth_login(provider: str):
|
||
"""Initiate OAuth login flow.
|
||
|
||
Redirects to the OAuth provider's authorization URL.
|
||
Currently a placeholder - requires OAuth provider implementation.
|
||
"""
|
||
_require_login_method("sso_enabled")
|
||
if provider not in ["github", "google"]:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail=f"Unsupported OAuth provider: {provider}",
|
||
)
|
||
|
||
raise HTTPException(
|
||
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
||
detail="OAuth login not yet implemented",
|
||
)
|
||
|
||
|
||
@router.get("/callback/{provider}")
|
||
async def oauth_callback(provider: str, code: str, state: str):
|
||
"""OAuth callback endpoint.
|
||
|
||
Handles the OAuth provider's callback after user authorization.
|
||
Currently a placeholder.
|
||
"""
|
||
_require_login_method("sso_enabled")
|
||
raise HTTPException(
|
||
status_code=status.HTTP_501_NOT_IMPLEMENTED,
|
||
detail="OAuth callback not yet implemented",
|
||
)
|