From fdcee9c349f2ee06aa54e755d48a49d26328ea96 Mon Sep 17 00:00:00 2001 From: flym Date: Tue, 8 Sep 2026 23:42:28 +0800 Subject: [PATCH] feat: implement Phase 2 auth and quota foundation --- app/account/page.tsx | 11 +- components/account-panel.tsx | 60 +++++ components/auth-form.tsx | 65 +++++- infra/README.md | 19 ++ infra/caddy/Caddyfile | 16 ++ infra/systemd/kaotings-api.service | 19 ++ infra/systemd/kaotings-web.service | 19 ++ services/api/.env.example | 11 + services/api/app/__init__.py | 0 services/api/app/config.py | 46 ++++ services/api/app/db.py | 14 ++ services/api/app/init_admin.py | 34 +++ services/api/app/main.py | 283 ++++++++++++++++++++++++ services/api/app/migrate.py | 25 +++ services/api/app/schemas.py | 69 ++++++ services/api/app/security.py | 127 +++++++++++ services/api/migrations/001_initial.sql | 115 ++++++++++ services/api/requirements.txt | 6 + 18 files changed, 924 insertions(+), 15 deletions(-) create mode 100644 components/account-panel.tsx create mode 100644 infra/README.md create mode 100644 infra/caddy/Caddyfile create mode 100644 infra/systemd/kaotings-api.service create mode 100644 infra/systemd/kaotings-web.service create mode 100644 services/api/.env.example create mode 100644 services/api/app/__init__.py create mode 100644 services/api/app/config.py create mode 100644 services/api/app/db.py create mode 100644 services/api/app/init_admin.py create mode 100644 services/api/app/main.py create mode 100644 services/api/app/migrate.py create mode 100644 services/api/app/schemas.py create mode 100644 services/api/app/security.py create mode 100644 services/api/migrations/001_initial.sql create mode 100644 services/api/requirements.txt diff --git a/app/account/page.tsx b/app/account/page.tsx index a02d28a..298bb48 100644 --- a/app/account/page.tsx +++ b/app/account/page.tsx @@ -1,6 +1,4 @@ -import Link from "next/link"; -import { ArrowRight } from "@/components/icons"; -import { PreviewNotice, StatusCard } from "@/components/ui"; +import { AccountPanel } from "@/components/account-panel"; export const metadata = { title: "账户中心" }; @@ -8,12 +6,7 @@ export default function AccountPage() { return (

Account

账户中心

未登录状态预览
-
账户数据尚未接入。以下字段只用于检查信息层级，不代表真实用户状态。
-
-
-

Profile

基本资料

邮箱
未登录
手机号
未绑定
角色
user / 待服务返回
前往登录
-

History

TTS 历史

打开工作台

暂无可显示的历史

登录和 TTS 任务服务完成后，历史记录将在此按用户归属显示。

-
+
); } diff --git a/components/account-panel.tsx b/components/account-panel.tsx new file mode 100644 index 0000000..573f184 --- /dev/null +++ b/components/account-panel.tsx @@ -0,0 +1,60 @@ +"use client"; + +import Link from "next/link"; +import { FormEvent, useEffect, useState } from "react"; +import { ArrowRight } from "@/components/icons"; +import { PreviewNotice, StatusCard } from "@/components/ui"; + +const apiBase = process.env.NEXT_PUBLIC_API_BASE_URL ?? "/api/v1"; + +type User = { email: string; role: "user" | "admin"; plan: "free" | "vip"; status: string; email_verified: boolean; phone_verified: boolean }; +type Usage = { period_start: string; period_end: string; limit: number; adjustment: number; used: number; reserved: number; available: number; plan: string }; + +export function AccountPanel() { + const [csrf, setCsrf] = useState(""); + const [user, setUser] = useState(null); + const [usage, setUsage] = useState(null); + const [error, setError] = useState(""); + const [passwordMessage, setPasswordMessage] = useState(""); + + async function load() { + const csrfResponse = await fetch(`${apiBase}/auth/csrf`, { credentials: "include" }); + const csrfBody = await csrfResponse.json(); + setCsrf(csrfBody.csrf_token ?? ""); + const [meResponse, usageResponse] = await Promise.all([ + fetch(`${apiBase}/auth/me`, { credentials: "include" }), + fetch(`${apiBase}/account/usage`, { credentials: "include" }), + ]); + if (!meResponse.ok) throw new Error("请先登录后查看账户"); + setUser(await meResponse.json()); + if (usageResponse.ok) setUsage(await usageResponse.json()); + } + + useEffect(() => { load().catch((loadError) => setError(loadError instanceof Error ? loadError.message : "无法加载账户")); }, []); + + async function logout() { + await fetch(`${apiBase}/auth/logout`, { method: "POST", credentials: "include", headers: { "X-CSRF-Token": csrf } }); + window.location.assign("/login"); + } + + async function changePassword(event: FormEvent) { + event.preventDefault(); + setPasswordMessage(""); + const data = new FormData(event.currentTarget); + const response = await fetch(`${apiBase}/auth/password/change`, { method: "POST", credentials: "include", headers: { "Content-Type": "application/json", "X-CSRF-Token": csrf }, body: JSON.stringify({ current_password: data.get("current_password"), new_password: data.get("new_password") }) }); + const body = await response.json().catch(() => ({})); + if (!response.ok) setPasswordMessage(body.error?.message ?? "密码修改失败"); + else { setPasswordMessage("密码已修改，其他会话已撤销"); event.currentTarget.reset(); } + } + + if (error) return
{error}。前往登录
; + if (!user || !usage) return
; + + return <> +
+
+

Profile

基本资料

邮箱
{user.email}
角色
{user.role}
额度周期
{new Date(usage.period_start).toLocaleDateString()} 至 {new Date(usage.period_end).toLocaleDateString()}
+

Usage

额度明细

打开工作台

Security

修改密码

{passwordMessage ?

{passwordMessage}

: null}
+
+ ; +} diff --git a/components/auth-form.tsx b/components/auth-form.tsx index 26ec2b5..a1c8717 100644 --- a/components/auth-form.tsx +++ b/components/auth-form.tsx @@ -1,16 +1,69 @@ +"use client"; + import Link from "next/link"; +import { FormEvent, useEffect, useState } from "react"; import { PreviewNotice } from "@/components/ui"; +const apiBase = process.env.NEXT_PUBLIC_API_BASE_URL ?? "/api/v1"; + export function AuthForm({ mode }: { mode: "login" | "register" }) { const isLogin = mode === "login"; + const [csrf, setCsrf] = useState(""); + const [email, setEmail] = useState(""); + const [password, setPassword] = useState(""); + const [confirmation, setConfirmation] = useState(""); + const [message, setMessage] = useState(""); + const [error, setError] = useState(""); + const [busy, setBusy] = useState(false); + + useEffect(() => { + fetch(`${apiBase}/auth/csrf`, { credentials: "include" }) + .then((response) => response.json()) + .then((body) => setCsrf(body.csrf_token ?? "")) + .catch(() => setError("暂时无法连接认证服务")); + }, []); + + async function submit(event: FormEvent) { + event.preventDefault(); + setError(""); + setMessage(""); + if (!isLogin && password !== confirmation) { + setError("两次输入的密码不一致"); + return; + } + setBusy(true); + try { + const response = await fetch(`${apiBase}/auth/${isLogin ? "login" : "register"}`, { + method: "POST", + credentials: "include", + headers: { "Content-Type": "application/json", ...(csrf ? { "X-CSRF-Token": csrf } : {}) }, + body: JSON.stringify({ email, password }), + }); + const body = await response.json().catch(() => ({})); + if (!response.ok) throw new Error(body.error?.message ?? "请求失败，请稍后重试"); + if (isLogin) { + window.location.assign("/account"); + } else { + setMessage("注册成功，请使用新账户登录"); + window.setTimeout(() => window.location.assign("/login"), 500); + } + } catch (submitError) { + setError(submitError instanceof Error ? submitError.message : "请求失败，请稍后重试"); + } finally { + setBusy(false); + } + } + return (
- 认证服务尚未接入。当前表单仅用于检查布局，不会提交或创建账户。 -
-
-
- {!isLogin ?
: null} - + 认证服务已接入测试环境。邮箱注册、登录和会话行为使用测试数据库，验证渠道仍未启用。 + +
setEmail(event.target.value)} placeholder="name@example.com" required type="email" value={email} />
+
setPassword(event.target.value)} placeholder="至少 8 位字符" required type="password" value={password} />
+ {!isLogin ?
setConfirmation(event.target.value)} placeholder="再次输入密码" required type="password" value={confirmation} />
: null} + {error ?

{error}

: null} + {message ?

{message}

: null} +

{isLogin ? "还没有账户？" : "已有账户？"}{" "}{isLogin ? "去注册" : "去登录"}

diff --git a/infra/README.md b/infra/README.md new file mode 100644 index 0000000..67f0a0a --- /dev/null +++ b/infra/README.md @@ -0,0 +1,19 @@ +# Phase 2 基础设施 + +本目录只保存不含秘密的部署模板： + +- `caddy/Caddyfile`：测试机 IP 的内部 CA HTTPS、Web/API 反向代理。 +- `systemd/kaotings-api.service`：FastAPI 自动重启服务。 +- `systemd/kaotings-web.service`：Next.js 自动重启服务。 + +数据库账号密码、`CSRF_SECRET` 和 Caddy 私有 CA 不进入 Git。服务器上的 `/etc/kaotings/api.env` 必须由 root 拥有且权限为 `0600`。 + +Phase 2 测试入口： + +- HTTPS：`https://192.168.199.22/` +- API：通过同源 `https://192.168.199.22/api/v1/*` +- API 回环端口：`127.0.0.1:8000`，不对局域网开放 +- Web 回环端口：`127.0.0.1:3000`，不对局域网开放 +- PostgreSQL：`127.0.0.1:5432`，不对局域网开放 + +Caddy `tls internal` 生成的根证书需要在测试浏览器/设备中导入后，HTTPS 才会显示为受信任。根证书可以从服务器的 Caddy 数据目录受控导出；私钥不得导出或提交。 diff --git a/infra/caddy/Caddyfile b/infra/caddy/Caddyfile new file mode 100644 index 0000000..6106607 --- /dev/null +++ b/infra/caddy/Caddyfile @@ -0,0 +1,16 @@ +https://192.168.199.22 { + tls internal + encode gzip + + handle /api/* { + reverse_proxy 127.0.0.1:8000 + } + + handle /healthz { + reverse_proxy 127.0.0.1:8000 + } + + handle { + reverse_proxy 127.0.0.1:3000 + } +} diff --git a/infra/systemd/kaotings-api.service b/infra/systemd/kaotings-api.service new file mode 100644 index 0000000..17edde1 --- /dev/null +++ b/infra/systemd/kaotings-api.service @@ -0,0 +1,19 @@ +[Unit] +Description=Kaotings business API +After=network-online.target postgresql.service +Wants=network-online.target + +[Service] +Type=simple +User=flym +Group=flym +WorkingDirectory=/home/flym/kaotings-api +EnvironmentFile=/etc/kaotings/api.env +ExecStart=/home/flym/kaotings-api/.venv/bin/uvicorn app.main:app --host 127.0.0.1 --port 8000 +Restart=always +RestartSec=3 +NoNewPrivileges=true +PrivateTmp=true + +[Install] +WantedBy=multi-user.target diff --git a/infra/systemd/kaotings-web.service b/infra/systemd/kaotings-web.service new file mode 100644 index 0000000..af1e3b7 --- /dev/null +++ b/infra/systemd/kaotings-web.service @@ -0,0 +1,19 @@ +[Unit] +Description=Kaotings website skeleton +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +User=flym +Group=flym +WorkingDirectory=/home/flym/kaotings-www +Environment=NODE_ENV=production +ExecStart=/usr/local/bin/npm run start -- -H 127.0.0.1 -p 3000 +Restart=always +RestartSec=3 +NoNewPrivileges=true +PrivateTmp=true + +[Install] +WantedBy=multi-user.target diff --git a/services/api/.env.example b/services/api/.env.example new file mode 100644 index 0000000..8e8c05e --- /dev/null +++ b/services/api/.env.example @@ -0,0 +1,11 @@ +DATABASE_URL=postgresql://kaotings_app:change-me@127.0.0.1:5432/kaotings +APP_ENV=test +ALLOWED_ORIGINS=http://192.168.199.22:3000,https://192.168.199.22 +SESSION_COOKIE_NAME=kaotings_session +SESSION_COOKIE_SECURE=false +SESSION_TTL_SECONDS=2592000 +CSRF_SECRET=replace-with-a-random-secret +EMAIL_VERIFICATION_ENABLED=false +PHONE_VERIFICATION_ENABLED=false +TEST_FREE_PERIOD_LIMIT=10000 +TEST_VIP_PERIOD_LIMIT=100000 diff --git a/services/api/app/__init__.py b/services/api/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/services/api/app/config.py b/services/api/app/config.py new file mode 100644 index 0000000..9b650b2 --- /dev/null +++ b/services/api/app/config.py @@ -0,0 +1,46 @@ +import os +from dataclasses import dataclass + + +def as_bool(value: str | None, default: bool = False) -> bool: + if value is None: + return default + return value.lower() in {"1", "true", "yes", "on"} + + +@dataclass(frozen=True) +class Settings: + database_url: str + app_env: str + allowed_origins: tuple[str, ...] + session_cookie_name: str + session_cookie_secure: bool + session_ttl_seconds: int + csrf_secret: str + email_verification_enabled: bool + phone_verification_enabled: bool + test_free_period_limit: int + test_vip_period_limit: int + + @classmethod + def from_env(cls) -> "Settings": + database_url = os.getenv("DATABASE_URL") + if not database_url: + raise RuntimeError("DATABASE_URL is required") + origins = tuple(item.strip().rstrip("/") for item in os.getenv("ALLOWED_ORIGINS", "").split(",") if item.strip()) + return cls( + database_url=database_url, + app_env=os.getenv("APP_ENV", "test"), + allowed_origins=origins, + session_cookie_name=os.getenv("SESSION_COOKIE_NAME", "kaotings_session"), + session_cookie_secure=as_bool(os.getenv("SESSION_COOKIE_SECURE")), + session_ttl_seconds=int(os.getenv("SESSION_TTL_SECONDS", "2592000")), + csrf_secret=os.getenv("CSRF_SECRET", "development-only-change-me"), + email_verification_enabled=as_bool(os.getenv("EMAIL_VERIFICATION_ENABLED")), + phone_verification_enabled=as_bool(os.getenv("PHONE_VERIFICATION_ENABLED")), + test_free_period_limit=int(os.getenv("TEST_FREE_PERIOD_LIMIT", "10000")), + test_vip_period_limit=int(os.getenv("TEST_VIP_PERIOD_LIMIT", "100000")), + ) + + +settings = Settings.from_env() diff --git a/services/api/app/db.py b/services/api/app/db.py new file mode 100644 index 0000000..5a69bc7 --- /dev/null +++ b/services/api/app/db.py @@ -0,0 +1,14 @@ +from collections.abc import Generator + +import psycopg +from psycopg.rows import dict_row + +from .config import settings + + +def get_connection() -> Generator[psycopg.Connection, None, None]: + connection = psycopg.connect(settings.database_url, row_factory=dict_row) + try: + yield connection + finally: + connection.close() diff --git a/services/api/app/init_admin.py b/services/api/app/init_admin.py new file mode 100644 index 0000000..a708212 --- /dev/null +++ b/services/api/app/init_admin.py @@ -0,0 +1,34 @@ +import os +import sys + +import psycopg + +from .config import settings +from .security import hash_password + + +def main() -> None: + email = os.getenv("ADMIN_EMAIL", "").strip().lower() + password = os.getenv("ADMIN_PASSWORD", "") + if not email or not password: + raise SystemExit("ADMIN_EMAIL and ADMIN_PASSWORD must be provided through the process environment") + if len(password) < 8: + raise SystemExit("ADMIN_PASSWORD must be at least 8 characters") + with psycopg.connect(settings.database_url) as connection: + existing = connection.execute("SELECT id FROM users WHERE email = %s", (email,)).fetchone() + if existing: + raise SystemExit("admin email already exists; refusing to overwrite") + row = connection.execute( + """ + INSERT INTO users(email, password_hash, role, plan, email_verified) + VALUES (%s, %s, 'admin', 'free', true) + RETURNING id + """, + (email, hash_password(password)), + ).fetchone() + connection.commit() + print(f"created admin {row[0]}") + + +if __name__ == "__main__": + main() diff --git a/services/api/app/main.py b/services/api/app/main.py new file mode 100644 index 0000000..9de701a --- /dev/null +++ b/services/api/app/main.py @@ -0,0 +1,283 @@ +from datetime import datetime, timedelta, timezone +from typing import Any +from uuid import UUID + +import psycopg +from fastapi import Depends, FastAPI, HTTPException, Request, Response, status +from fastapi.responses import JSONResponse +from psycopg import Connection + +from .config import settings +from .db import get_connection +from .schemas import ( + AuthResponse, + LoginRequest, + MembershipRequest, + PasswordChangeRequest, + QuotaAdjustmentRequest, + RegisterRequest, + StatusRequest, + UserPublic, + VerificationConfirmRequest, + VerificationSendRequest, + normalize_email, +) +from .security import ( + check_origin, + clear_session_cookie, + code_digest, + current_user, + hash_password, + new_csrf_token, + new_token, + password_hasher, + rate_limiter, + require_admin, + require_csrf, + set_csrf_cookie, + set_session_cookie, + token_digest, + utc_now, + verify_password, +) + +app = FastAPI(title="Kaotings Business API", version="0.2.0", docs_url=None if settings.app_env == "production" else "/docs") + + +def error(code: str, message: str, http_status: int) -> HTTPException: + return HTTPException(status_code=http_status, detail={"code": code, "message": message}) + + +def period_bounds(now: datetime | None = None) -> tuple[datetime, datetime]: + current = now or utc_now() + start = current.replace(day=1, hour=0, minute=0, second=0, microsecond=0) + if start.month == 12: + end = start.replace(year=start.year + 1, month=1) + else: + end = start.replace(month=start.month + 1) + return start, end + + +def effective_plan(connection: Connection, user: dict) -> str: + if user["role"] == "admin": + return user["plan"] + grant = connection.execute( + """ + SELECT plan FROM membership_grants + WHERE user_id = %s AND revoked_at IS NULL AND starts_at <= now() + AND (expires_at IS NULL OR expires_at > now()) + ORDER BY starts_at DESC LIMIT 1 + """, + (user["id"],), + ).fetchone() + return grant["plan"] if grant else "free" + + +def user_public(connection: Connection, user: dict) -> UserPublic: + return UserPublic( + id=user["id"], email=user["email"], phone=user["phone"], role=user["role"], plan=effective_plan(connection, user), + status=user["status"], email_verified=user["email_verified"], phone_verified=user["phone_verified"], + created_at=user["created_at"], last_login_at=user["last_login_at"], + ) + + +def ensure_quota(connection: Connection, user_id: UUID, plan: str) -> dict: + start, end = period_bounds() + policy = connection.execute("SELECT period_limit FROM plan_policies WHERE code = %s", (plan,)).fetchone() + if not policy: + raise error("POLICY_MISSING", "权益策略未配置", 500) + connection.execute( + """ + INSERT INTO quota_accounts(user_id, period_start, period_end, limit_snapshot) + VALUES (%s, %s, %s, %s) + ON CONFLICT (user_id, period_start, period_end) DO NOTHING + """, + (user_id, start, end, policy["period_limit"]), + ) + row = connection.execute( + "SELECT * FROM quota_accounts WHERE user_id = %s AND period_start = %s AND period_end = %s", + (user_id, start, end), + ).fetchone() + return row + + +def usage_view(quota: dict) -> dict[str, Any]: + available = max(0, quota["limit_snapshot"] + quota["adjustment"] - quota["used"] - quota["reserved"]) + return { + "period_start": quota["period_start"], "period_end": quota["period_end"], "limit": quota["limit_snapshot"], + "adjustment": quota["adjustment"], "used": quota["used"], "reserved": quota["reserved"], "available": available, + } + + +@app.exception_handler(HTTPException) +async def http_error_handler(_: Request, exc: HTTPException): + detail = exc.detail if isinstance(exc.detail, dict) else {"code": "REQUEST_FAILED", "message": str(exc.detail)} + return JSONResponse(status_code=exc.status_code, content={"error": detail}) + + +@app.get("/healthz") +def health(connection: Connection = Depends(get_connection)): + connection.execute("SELECT 1") + return {"status": "ok", "database": "ok", "service": "api"} + + +@app.get("/api/v1/auth/csrf") +def csrf(response: Response, request: Request): + check_origin(request) + token = request.cookies.get("kaotings_csrf") or new_csrf_token() + set_csrf_cookie(response, token) + return {"csrf_token": token} + + +@app.post("/api/v1/auth/register", response_model=AuthResponse, status_code=201) +def register(payload: RegisterRequest, request: Request, connection: Connection = Depends(get_connection)): + check_origin(request) + key = f"register:{request.client.host if request.client else 'unknown'}" + if not rate_limiter.allow(key, 5, 3600): + raise error("RATE_LIMITED", "请求过于频繁", 429) + email = normalize_email(str(payload.email)) + try: + user = connection.execute( + """ + INSERT INTO users(email, password_hash) VALUES (%s, %s) + RETURNING * + """, + (email, hash_password(payload.password)), + ).fetchone() + ensure_quota(connection, user["id"], "free") + connection.commit() + except psycopg.errors.UniqueViolation: + connection.rollback() + raise error("REGISTRATION_FAILED", "注册信息不可用", 409) + return {"user": user_public(connection, user)} + + +@app.post("/api/v1/auth/login", response_model=AuthResponse) +def login(payload: LoginRequest, request: Request, response: Response, connection: Connection = Depends(get_connection)): + check_origin(request) + key = f"login:{request.client.host if request.client else 'unknown'}:{normalize_email(str(payload.email))}" + if not rate_limiter.allow(key, 10, 300): + raise error("RATE_LIMITED", "请求过于频繁", 429) + user = connection.execute("SELECT * FROM users WHERE email = %s", (normalize_email(str(payload.email)),)).fetchone() + if not user or not verify_password(payload.password, user["password_hash"]): + raise error("LOGIN_FAILED", "邮箱或密码错误", 401) + if user["status"] != "active": + raise error("ACCOUNT_DISABLED", "账户不可用", 403) + token = new_token() + connection.execute("INSERT INTO sessions(user_id, token_hash, expires_at) VALUES (%s, %s, %s)", (user["id"], token_digest(token), utc_now() + timedelta(seconds=settings.session_ttl_seconds))) + user = connection.execute("UPDATE users SET last_login_at = now(), updated_at = now() WHERE id = %s RETURNING *", (user["id"],)).fetchone() + ensure_quota(connection, user["id"], effective_plan(connection, user)) + connection.commit() + set_session_cookie(response, token) + return {"user": user_public(connection, user)} + + +@app.post("/api/v1/auth/logout", status_code=204, dependencies=[Depends(require_csrf)]) +def logout(response: Response, request: Request, connection: Connection = Depends(get_connection)): + token = request.cookies.get(settings.session_cookie_name) + if token: + connection.execute("UPDATE sessions SET revoked_at = now() WHERE token_hash = %s", (token_digest(token),)) + connection.commit() + clear_session_cookie(response) + return Response(status_code=204) + + +@app.get("/api/v1/auth/me", response_model=UserPublic) +def me(user: dict = Depends(current_user), connection: Connection = Depends(get_connection)): + return user_public(connection, user) + + +@app.post("/api/v1/auth/password/change", dependencies=[Depends(require_csrf)]) +def change_password(payload: PasswordChangeRequest, request: Request, connection: Connection = Depends(get_connection), user: dict = Depends(current_user)): + if not verify_password(payload.current_password, user["password_hash"]): + raise error("PASSWORD_INVALID", "当前密码错误", 400) + connection.execute("UPDATE users SET password_hash = %s, updated_at = now() WHERE id = %s", (hash_password(payload.new_password), user["id"])) + connection.execute("UPDATE sessions SET revoked_at = now() WHERE user_id = %s AND id <> %s", (user["id"], user["session_id"])) + connection.commit() + return {"status": "ok"} + + +@app.get("/api/v1/account/usage") +def account_usage(user: dict = Depends(current_user), connection: Connection = Depends(get_connection)): + quota = ensure_quota(connection, user["id"], effective_plan(connection, user)) + connection.commit() + return {"plan": effective_plan(connection, user), **usage_view(quota)} + + +@app.post("/api/v1/auth/verification/send") +def verification_send(payload: VerificationSendRequest, request: Request, user: dict = Depends(current_user)): + check_origin(request) + enabled = settings.email_verification_enabled if payload.channel == "email" else settings.phone_verification_enabled + if not enabled: + raise error("VERIFICATION_NOT_ENABLED", "验证渠道尚未启用", 503) + raise error("VERIFICATION_PROVIDER_UNAVAILABLE", "验证渠道尚未配置", 503) + + +@app.post("/api/v1/auth/verification/confirm", dependencies=[Depends(require_csrf)]) +def verification_confirm(payload: VerificationConfirmRequest, request: Request, connection: Connection = Depends(get_connection), user: dict = Depends(current_user)): + check_origin(request) + challenge = connection.execute("SELECT * FROM verification_challenges WHERE id = %s AND user_id = %s", (payload.challenge_id, user["id"])).fetchone() + if not challenge or challenge["consumed_at"] or challenge["expires_at"] <= utc_now() or challenge["attempt_count"] >= 5: + raise error("VERIFICATION_INVALID", "验证挑战无效或已过期", 400) + if not __import__("hmac").compare_digest(challenge["code_digest"], code_digest(payload.code)): + connection.execute("UPDATE verification_challenges SET attempt_count = attempt_count + 1 WHERE id = %s", (payload.challenge_id,)) + connection.commit() + raise error("VERIFICATION_INVALID", "验证码错误", 400) + connection.execute("UPDATE verification_challenges SET consumed_at = now() WHERE id = %s", (payload.challenge_id,)) + if challenge["channel"] == "email": + connection.execute("UPDATE users SET email_verified = true, updated_at = now() WHERE id = %s", (user["id"],)) + else: + connection.execute("UPDATE users SET phone_verified = true, updated_at = now() WHERE id = %s", (user["id"],)) + connection.commit() + return {"status": "ok"} + + +@app.get("/api/v1/admin/users") +def admin_users(user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)): + rows = connection.execute("SELECT id, email, phone, role, plan, status, email_verified, phone_verified, created_at, last_login_at FROM users ORDER BY created_at DESC LIMIT 100").fetchall() + return {"items": rows} + + +@app.patch("/api/v1/admin/users/{user_id}/status", dependencies=[Depends(require_csrf)]) +def admin_status(user_id: UUID, payload: StatusRequest, connection: Connection = Depends(get_connection), actor: dict = Depends(require_admin)): + target = connection.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone() + if not target: + raise error("NOT_FOUND", "用户不存在", 404) + if target["role"] == "admin" and payload.status == "disabled": + count = connection.execute("SELECT count(*) FROM users WHERE role = 'admin' AND status = 'active'").fetchone()["count"] + if count <= 1: + raise error("LAST_ADMIN_PROTECTED", "不能禁用最后一个可用管理员", 409) + updated = connection.execute("UPDATE users SET status = %s, updated_at = now() WHERE id = %s RETURNING id, status", (payload.status, user_id)).fetchone() + connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, before_value, after_value, reason) VALUES (%s, 'user_status', 'user', %s, %s, %s, %s)", (actor["id"], user_id, {"status": target["status"]}, {"status": updated["status"]}, payload.reason)) + connection.commit() + return updated + + +@app.put("/api/v1/admin/users/{user_id}/membership", dependencies=[Depends(require_csrf)]) +def admin_membership(user_id: UUID, payload: MembershipRequest, connection: Connection = Depends(get_connection), actor: dict = Depends(require_admin)): + target = connection.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone() + if not target: + raise error("NOT_FOUND", "用户不存在", 404) + starts_at = payload.starts_at or utc_now() + connection.execute("UPDATE membership_grants SET revoked_at = now() WHERE user_id = %s AND revoked_at IS NULL", (user_id,)) + grant = connection.execute("INSERT INTO membership_grants(user_id, plan, starts_at, expires_at, created_by, reason) VALUES (%s, %s, %s, %s, %s, %s) RETURNING id, plan, starts_at, expires_at", (user_id, payload.plan, starts_at, payload.expires_at, actor["id"], payload.reason)).fetchone() + connection.execute("UPDATE users SET plan = %s, updated_at = now() WHERE id = %s", (payload.plan, user_id)) + connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, before_value, after_value, reason) VALUES (%s, 'membership', 'user', %s, %s, %s, %s)", (actor["id"], user_id, {"plan": target["plan"]}, {"plan": payload.plan, "expires_at": payload.expires_at.isoformat() if payload.expires_at else None}, payload.reason)) + connection.commit() + return grant + + +@app.post("/api/v1/admin/users/{user_id}/quota-adjustments", dependencies=[Depends(require_csrf)]) +def admin_quota(user_id: UUID, payload: QuotaAdjustmentRequest, connection: Connection = Depends(get_connection), actor: dict = Depends(require_admin)): + target = connection.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone() + if not target: + raise error("NOT_FOUND", "用户不存在", 404) + quota = ensure_quota(connection, user_id, effective_plan(connection, target)) + existing = connection.execute("SELECT id FROM usage_records WHERE user_id = %s AND idempotency_key = %s", (user_id, payload.idempotency_key)).fetchone() + if existing: + raise error("IDEMPOTENCY_CONFLICT", "该调整已提交", 409) + connection.execute("UPDATE quota_accounts SET adjustment = adjustment + %s, version = version + 1 WHERE id = %s", (payload.amount, quota["id"])) + connection.execute("INSERT INTO usage_records(user_id, quota_account_id, type, amount, idempotency_key) VALUES (%s, %s, 'adjust', %s, %s)", (user_id, quota["id"], payload.amount, payload.idempotency_key)) + connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, after_value, reason) VALUES (%s, 'quota_adjustment', 'user', %s, %s, %s)", (actor["id"], user_id, {"amount": payload.amount}, payload.reason)) + connection.commit() + return {"status": "ok", "amount": payload.amount} diff --git a/services/api/app/migrate.py b/services/api/app/migrate.py new file mode 100644 index 0000000..4f7fc87 --- /dev/null +++ b/services/api/app/migrate.py @@ -0,0 +1,25 @@ +from pathlib import Path + +import psycopg + +from .config import settings + + +def main() -> None: + migration_dir = Path(__file__).resolve().parent.parent / "migrations" + with psycopg.connect(settings.database_url) as connection: + connection.execute("CREATE TABLE IF NOT EXISTS schema_migrations (version TEXT PRIMARY KEY, applied_at TIMESTAMPTZ NOT NULL DEFAULT now())") + connection.commit() + for path in sorted(migration_dir.glob("*.sql")): + version = path.name + applied = connection.execute("SELECT 1 FROM schema_migrations WHERE version = %s", (version,)).fetchone() + if applied: + continue + connection.execute(path.read_text(encoding="utf-8")) + connection.execute("INSERT INTO schema_migrations(version) VALUES (%s)", (version,)) + connection.commit() + print(f"applied {version}") + + +if __name__ == "__main__": + main() diff --git a/services/api/app/schemas.py b/services/api/app/schemas.py new file mode 100644 index 0000000..d07b9be --- /dev/null +++ b/services/api/app/schemas.py @@ -0,0 +1,69 @@ +from datetime import datetime +from typing import Any, Literal +from uuid import UUID + +from pydantic import BaseModel, EmailStr, Field, field_validator + + +class RegisterRequest(BaseModel): + email: EmailStr + password: str = Field(min_length=8, max_length=128) + + +class LoginRequest(BaseModel): + email: EmailStr + password: str = Field(min_length=1, max_length=128) + + +class PasswordChangeRequest(BaseModel): + current_password: str = Field(min_length=1, max_length=128) + new_password: str = Field(min_length=8, max_length=128) + + +class UserPublic(BaseModel): + id: UUID + email: str | None + phone: str | None + role: Literal["user", "admin"] + plan: Literal["free", "vip"] + status: Literal["active", "disabled"] + email_verified: bool + phone_verified: bool + created_at: datetime + last_login_at: datetime | None + + +class AuthResponse(BaseModel): + user: UserPublic + + +class MembershipRequest(BaseModel): + plan: Literal["free", "vip"] + starts_at: datetime | None = None + expires_at: datetime | None = None + reason: str = Field(min_length=1, max_length=500) + + +class StatusRequest(BaseModel): + status: Literal["active", "disabled"] + reason: str = Field(min_length=1, max_length=500) + + +class QuotaAdjustmentRequest(BaseModel): + amount: int + reason: str = Field(min_length=1, max_length=500) + idempotency_key: str = Field(min_length=8, max_length=120) + + +class VerificationConfirmRequest(BaseModel): + challenge_id: UUID + code: str = Field(min_length=4, max_length=12) + + +class VerificationSendRequest(BaseModel): + channel: Literal["email", "phone"] + purpose: Literal["registration", "contact_binding", "contact_change", "password_reset"] + + +def normalize_email(value: str) -> str: + return value.strip().lower() diff --git a/services/api/app/security.py b/services/api/app/security.py new file mode 100644 index 0000000..50d51f0 --- /dev/null +++ b/services/api/app/security.py @@ -0,0 +1,127 @@ +import hashlib +import hmac +import secrets +import threading +import time +from collections import defaultdict, deque +from datetime import datetime, timezone + +from argon2 import PasswordHasher +from argon2.exceptions import InvalidHashError, VerificationError +from fastapi import Depends, HTTPException, Request, status +from psycopg import Connection + +from .config import settings +from .db import get_connection + +password_hasher = PasswordHasher() + + +def hash_password(password: str) -> str: + return password_hasher.hash(password) + + +def verify_password(password: str, password_hash: str) -> bool: + try: + return password_hasher.verify(password_hash, password) + except (InvalidHashError, VerificationError): + return False + + +def token_digest(token: str) -> str: + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + +def new_token() -> str: + return secrets.token_urlsafe(32) + + +def new_csrf_token() -> str: + return secrets.token_urlsafe(32) + + +def code_digest(code: str) -> str: + return hmac.new(settings.csrf_secret.encode(), code.encode(), hashlib.sha256).hexdigest() + + +def utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def check_origin(request: Request) -> None: + origin = request.headers.get("origin") + if origin and origin.rstrip("/") not in settings.allowed_origins: + raise HTTPException(status_code=403, detail={"code": "ORIGIN_NOT_ALLOWED", "message": "请求来源不被允许"}) + + +def require_csrf(request: Request) -> None: + check_origin(request) + header = request.headers.get("x-csrf-token") + cookie = request.cookies.get("kaotings_csrf") + if not header or not cookie or not hmac.compare_digest(header, cookie): + raise HTTPException(status_code=403, detail={"code": "CSRF_FAILED", "message": "请求校验失败"}) + + +def set_session_cookie(response, token: str) -> None: + response.set_cookie( + settings.session_cookie_name, + token, + max_age=settings.session_ttl_seconds, + httponly=True, + secure=settings.session_cookie_secure, + samesite="lax", + path="/", + ) + + +def clear_session_cookie(response) -> None: + response.delete_cookie(settings.session_cookie_name, path="/") + + +def set_csrf_cookie(response, token: str) -> None: + response.set_cookie("kaotings_csrf", token, max_age=settings.session_ttl_seconds, httponly=False, secure=settings.session_cookie_secure, samesite="lax", path="/") + + +def current_user(request: Request, connection: Connection = Depends(get_connection)) -> dict: + token = request.cookies.get(settings.session_cookie_name) + if not token: + raise HTTPException(status_code=401, detail={"code": "AUTH_REQUIRED", "message": "请先登录"}) + user = connection.execute( + """ + SELECT u.*, s.id AS session_id + FROM sessions s JOIN users u ON u.id = s.user_id + WHERE s.token_hash = %s AND s.revoked_at IS NULL AND s.expires_at > now() + """, + (token_digest(token),), + ).fetchone() + if not user: + raise HTTPException(status_code=401, detail={"code": "AUTH_REQUIRED", "message": "会话已失效"}) + if user["status"] != "active": + raise HTTPException(status_code=403, detail={"code": "ACCOUNT_DISABLED", "message": "账户不可用"}) + return user + + +def require_admin(user: dict = Depends(current_user)) -> dict: + if user["role"] != "admin": + raise HTTPException(status_code=403, detail={"code": "FORBIDDEN", "message": "无权执行此操作"}) + return user + + +class RateLimiter: + def __init__(self) -> None: + self._events: dict[str, deque[float]] = defaultdict(deque) + self._lock = threading.Lock() + + def allow(self, key: str, limit: int, window_seconds: int) -> bool: + now = time.monotonic() + with self._lock: + events = self._events[key] + while events and events[0] <= now - window_seconds: + events.popleft() + if len(events) >= limit: + return False + events.append(now) + return True + + +rate_limiter = RateLimiter() diff --git a/services/api/migrations/001_initial.sql b/services/api/migrations/001_initial.sql new file mode 100644 index 0000000..efcc686 --- /dev/null +++ b/services/api/migrations/001_initial.sql @@ -0,0 +1,115 @@ +CREATE EXTENSION IF NOT EXISTS pgcrypto; + +CREATE TABLE IF NOT EXISTS schema_migrations ( + version TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS users ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + email TEXT UNIQUE, + phone TEXT UNIQUE, + password_hash TEXT NOT NULL, + role TEXT NOT NULL DEFAULT 'user' CHECK (role IN ('user', 'admin')), + plan TEXT NOT NULL DEFAULT 'free' CHECK (plan IN ('free', 'vip')), + status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'disabled')), + email_verified BOOLEAN NOT NULL DEFAULT false, + phone_verified BOOLEAN NOT NULL DEFAULT false, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + last_login_at TIMESTAMPTZ, + CHECK (email IS NOT NULL OR phone IS NOT NULL) +); + +CREATE TABLE IF NOT EXISTS sessions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + token_hash TEXT NOT NULL UNIQUE, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + expires_at TIMESTAMPTZ NOT NULL, + revoked_at TIMESTAMPTZ, + last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE INDEX IF NOT EXISTS sessions_user_idx ON sessions(user_id, expires_at); + +CREATE TABLE IF NOT EXISTS plan_policies ( + code TEXT PRIMARY KEY CHECK (code IN ('free', 'vip')), + period_limit BIGINT NOT NULL CHECK (period_limit >= 0), + max_text_length INTEGER NOT NULL CHECK (max_text_length >= 0), + max_concurrency INTEGER NOT NULL CHECK (max_concurrency >= 0), + rate_limit INTEGER NOT NULL CHECK (rate_limit >= 0), + allowed_voices JSONB NOT NULL DEFAULT '[]'::jsonb, + version INTEGER NOT NULL DEFAULT 1, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS membership_grants ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + plan TEXT NOT NULL CHECK (plan IN ('free', 'vip')), + starts_at TIMESTAMPTZ NOT NULL, + expires_at TIMESTAMPTZ, + revoked_at TIMESTAMPTZ, + created_by UUID REFERENCES users(id), + reason TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CHECK (expires_at IS NULL OR expires_at > starts_at) +); +CREATE INDEX IF NOT EXISTS membership_active_idx ON membership_grants(user_id, starts_at, expires_at, revoked_at); + +CREATE TABLE IF NOT EXISTS quota_accounts ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + period_start TIMESTAMPTZ NOT NULL, + period_end TIMESTAMPTZ NOT NULL, + limit_snapshot BIGINT NOT NULL CHECK (limit_snapshot >= 0), + adjustment BIGINT NOT NULL DEFAULT 0, + used BIGINT NOT NULL DEFAULT 0 CHECK (used >= 0), + reserved BIGINT NOT NULL DEFAULT 0 CHECK (reserved >= 0), + version BIGINT NOT NULL DEFAULT 0, + UNIQUE (user_id, period_start, period_end), + CHECK (period_end > period_start) +); + +CREATE TABLE IF NOT EXISTS usage_records ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + quota_account_id UUID NOT NULL REFERENCES quota_accounts(id) ON DELETE CASCADE, + task_id UUID, + type TEXT NOT NULL CHECK (type IN ('reserve', 'consume', 'release', 'adjust')), + amount BIGINT NOT NULL, + idempotency_key TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE UNIQUE INDEX IF NOT EXISTS usage_adjustment_idem_idx ON usage_records(user_id, idempotency_key) WHERE idempotency_key IS NOT NULL; + +CREATE TABLE IF NOT EXISTS verification_challenges ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID REFERENCES users(id) ON DELETE CASCADE, + channel TEXT NOT NULL CHECK (channel IN ('email', 'phone')), + purpose TEXT NOT NULL CHECK (purpose IN ('registration', 'contact_binding', 'contact_change', 'password_reset')), + target TEXT NOT NULL, + code_digest TEXT NOT NULL, + expires_at TIMESTAMPTZ NOT NULL, + attempt_count INTEGER NOT NULL DEFAULT 0, + consumed_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS admin_audit_logs ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + actor_id UUID NOT NULL REFERENCES users(id), + action TEXT NOT NULL, + target_type TEXT NOT NULL, + target_id UUID NOT NULL, + before_value JSONB, + after_value JSONB, + reason TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +INSERT INTO plan_policies(code, period_limit, max_text_length, max_concurrency, rate_limit, allowed_voices) +VALUES + ('free', 10000, 5000, 1, 10, '[]'::jsonb), + ('vip', 100000, 10000, 3, 30, '[]'::jsonb) +ON CONFLICT (code) DO NOTHING; diff --git a/services/api/requirements.txt b/services/api/requirements.txt new file mode 100644 index 0000000..6962010 --- /dev/null +++ b/services/api/requirements.txt @@ -0,0 +1,6 @@ +fastapi==0.115.14 +uvicorn[standard]==0.34.3 +psycopg[binary]==3.2.9 +argon2-cffi==25.1.0 +pydantic==2.11.7 +email-validator==2.2.0