diff --git a/tests/test_register.py b/tests/test_register.py
new file mode 100644
index 0000000..b26d073
--- /dev/null
+++ b/tests/test_register.py
@@ -0,0 +1,120 @@
+# -*- coding: utf-8 -*-
+"""注册/找回/卡密端点的沙箱测试（不触碰生产 data/site.db）。
+
+直接向沙箱库 email_codes 表播种验证码，绕过 SMTP，覆盖：
+- 注册协议门槛、用户名/密码长度校验
+- 注册成功路径：建号 + 赠送次数 + 签发可用会话 Cookie
+- 验证码错误分支
+- 找回密码：重置后可用新密码登录、旧会话失效
+- 卡密充值
+"""
+import time
+
+import pytest
+from fastapi.testclient import TestClient
+
+import auth_system as auth
+import core.db as db
+
+SECRET = 'register-test-secret'
+
+
+@pytest.fixture()
+def env(tmp_path, monkeypatch):
+    db_path = tmp_path / 'site.db'
+    monkeypatch.setattr(auth, 'DATA_DIR', tmp_path)
+    monkeypatch.setattr(auth, 'DB_PATH', db_path)
+    monkeypatch.setattr(auth, 'BACKUP_DIR', tmp_path / 'backups')
+    monkeypatch.setattr(auth, 'SECRET_KEY_PATH', tmp_path / 'secret.key')
+    monkeypatch.setattr(db, 'DATA_DIR', tmp_path)
+    monkeypatch.setattr(db, 'DB_PATH', db_path)
+    auth.init_db()
+    monkeypatch.setattr('web.deps.secret_key', lambda: SECRET)
+    monkeypatch.setattr('web.routers.auth.secret_key', lambda: SECRET)
+    monkeypatch.setattr('web.routers.register.secret_key', lambda: SECRET)
+    import auth_routes
+    auth_routes._SEND_CODE_IP.clear()
+    from web.app import create_app
+    return TestClient(create_app())
+
+
+def _seed_code(email: str, code: str, purpose: str = 'register'):
+    conn = auth.get_conn()
+    conn.execute('DELETE FROM email_codes WHERE email = ? AND purpose = ?', (email, purpose))
+    conn.execute('INSERT INTO email_codes (email, code, expires_at, created_at, purpose) '
+                 'VALUES (?, ?, ?, ?, ?)',
+                 (email, code, int(time.time()) + 600, int(time.time()), purpose))
+    conn.commit()
+    conn.close()
+
+
+def test_register_requires_agreement(env):
+    r = env.post('/api/v1/auth/register', json={
+        'username': 'newuser', 'password': 'Passw0rd12', 'email': 'n@example.com',
+        'code': '000000', 'agree': '0'})
+    assert r.json()['message'].startswith('请先阅读并同意')
+
+
+def test_register_success(env):
+    _seed_code('new@example.com', '123456')
+    r = env.post('/api/v1/auth/register', json={
+        'username': 'newuser', 'password': 'Passw0rd12', 'email': 'new@example.com',
+        'code': '123456', 'agree': '1', 'device_id': 'dev-1'})
+    assert r.status_code == 200
+    body = r.json()
+    assert body['success'] is True
+    assert '已赠送' in body['message']
+    assert 'session' in r.cookies
+
+    # 会话可用
+    assert env.post('/api/v1/auth/heartbeat', json={}).status_code == 200
+
+    # 用户名/邮箱已被占用
+    _seed_code('new@example.com', '222222')
+    r2 = env.post('/api/v1/auth/register', json={
+        'username': 'newuser', 'password': 'Passw0rd12', 'email': 'new@example.com',
+        'code': '222222', 'agree': '1'})
+    assert '已被注册' in r2.json()['message']
+
+
+def test_register_bad_code(env):
+    r = env.post('/api/v1/auth/register', json={
+        'username': 'another', 'password': 'Passw0rd12', 'email': 'a@example.com',
+        'code': '999999', 'agree': '1'})
+    assert r.json()['message'] == '邮箱验证码错误或已过期'
+
+
+def test_forgot_password_resets(env):
+    # 建一个用户
+    auth.create_user('forgetme', 'OldPass123', email='f@example.com', credits=5)
+    _seed_code('f@example.com', '654321', purpose='reset')
+    r = env.post('/api/v1/auth/forgot_password', json={
+        'email': 'f@example.com', 'code': '654321', 'password': 'NewPass456'})
+    assert r.status_code == 200
+    assert r.json()['success'] is True
+    # 新密码可登录
+    r2 = env.post('/api/v1/auth/login', json={'username': 'forgetme', 'password': 'NewPass456'})
+    assert r2.status_code == 200 and r2.json()['success'] is True
+
+
+def test_forgot_password_admin_blocked(env):
+    conn = auth.get_conn()
+    conn.execute("UPDATE users SET role='admin', email='boss@example.com' WHERE id=1")
+    conn.commit()
+    conn.close()
+    _seed_code('boss@example.com', '111111', purpose='reset')
+    r = env.post('/api/v1/auth/forgot_password', json={
+        'email': 'boss@example.com', 'code': '111111', 'password': 'NewPass456'})
+    # 管理员不走自助找回：统一防枚举文案
+    assert r.json()['message'] == '邮箱验证码错误或已过期'
+
+
+def test_redeem_card(env):
+    # 造一张卡密并让用户登录
+    auth.create_user('buyer', 'Passw0rd12', email='b@example.com', credits=0)
+    code = auth.generate_cards(1, 10, note='test')[0]
+    env.post('/api/v1/auth/login', json={'username': 'buyer', 'password': 'Passw0rd12'})
+    r = env.post('/api/v1/auth/redeem', json={'code': code})
+    assert r.status_code == 200
+    assert r.json()['success'] is True
+    assert r.json()['data']['credits'] == 10
diff --git a/web/app.py b/web/app.py
index 5292941..7f31be8 100644
--- a/web/app.py
+++ b/web/app.py
@@ -31,6 +31,7 @@ from web.routers import delivery as delivery_router
 from web.routers import index as index_router
 from web.routers import music as music_router
 from web.routers import podcast as podcast_router
+from web.routers import register as register_router
 from web.routers import site as site_router
 from web.routers import stream as stream_router
 from web.routers import system as system_router
@@ -95,6 +96,7 @@ def create_app(serve_flask_fallback: bool = False) -> FastAPI:
     app.include_router(auth_router.router, prefix='/api/v1', tags=['auth'])
     app.include_router(user_router.router, prefix='/api/v1', tags=['user'])
     app.include_router(admin_router.router, prefix='/api/v1', tags=['admin'])
+    app.include_router(register_router.router, prefix='/api/v1', tags=['auth'])
     # 首页与静态资源（迁移完成形态；fallback 模式下同路径也能由 Flask 提供）
     app.include_router(index_router.router, tags=['index'])
 
diff --git a/web/routers/register.py b/web/routers/register.py
new file mode 100644
index 0000000..075cf42
--- /dev/null
+++ b/web/routers/register.py
@@ -0,0 +1,243 @@
+# -*- coding: utf-8 -*-
+"""注册 / 找回密码 / 卡密 / 验证码（FastAPI 栈）——与 Flask 侧响应契约逐字段对齐。
+
+已迁移：
+- POST /api/v1/auth/send_code        ↔ Flask /auth/send_code
+- POST /api/v1/auth/register         ↔ Flask /auth/register
+- POST /api/v1/auth/trial            ↔ Flask /auth/trial
+- POST /api/v1/auth/redeem           ↔ Flask /auth/redeem
+- POST /api/v1/auth/forgot_password  ↔ Flask /auth/forgot_password
+
+说明：
+- 同步的 auth_system 调用（SQLite 事务、邮件 SMTP、PBKDF2）经 asyncio.to_thread 执行，
+  不阻塞事件循环；行为与 Flask 完全一致。
+- 发码 IP 限流复用 auth_routes 的模块级状态 `_SEND_CODE_IP`（跨栈共享计数）。
+- 失败锁定/验证码消费等安全口径逐字保留。
+"""
+import asyncio
+import os
+import sqlite3
+import time
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import JSONResponse
+
+import auth_system as auth
+from core import db, repo
+from core import session as cs
+from web import ratelimit
+from web.deps import SiteUser, current_user, secret_key
+from web.response import error, success
+
+router = APIRouter()
+
+SESSION_COOKIE = 'session'
+_SESSION_MAX_AGE = 7 * 86400
+
+
+async def _request_data(request: Request) -> dict[str, Any]:
+    data: dict[str, Any] = {}
+    if request.method == 'GET':
+        data.update(dict(request.query_params))
+        return data
+    try:
+        body = await request.json()
+        if isinstance(body, dict):
+            data.update(body)
+    except Exception:
+        pass
+    if not data:
+        try:
+            form = await request.form()
+            data.update(dict(form.items()))
+        except Exception:
+            pass
+    return data
+
+
+def _user_public(row: dict) -> dict:
+    return {
+        'id': row['id'],
+        'username': row['username'],
+        'email': row.get('email'),
+        'role': row['role'],
+        'duration_seconds': row['duration_seconds'],
+        'created_at': row.get('created_at'),
+        'last_login': row.get('last_login'),
+        'ann_seen': row.get('ann_seen'),
+    }
+
+
+def _login_cookie_response(payload: dict, message: str, uid: int, seq: int,
+                           secure: bool) -> JSONResponse:
+    resp = JSONResponse(content={'status': 200, 'success': True, 'message': message, 'data': payload})
+    cookie = cs.dumps(secret_key(), {'site_uid': uid, 'site_uid_seq': seq,
+                                     'site_uid_ts': int(time.time())})
+    resp.set_cookie(SESSION_COOKIE, cookie, max_age=_SESSION_MAX_AGE, httponly=True,
+                    samesite='lax', path='/', secure=secure)
+    return resp
+
+
+async def _safe_register_credits() -> int:
+    try:
+        credits = int(await repo.get_setting('register_credits', str(auth.DEFAULT_REGISTER_CREDITS)) or 0)
+    except (TypeError, ValueError):
+        return auth.DEFAULT_REGISTER_CREDITS
+    return max(0, credits)
+
+
+@router.post('/auth/send_code')
+async def send_code(request: Request):
+    """发送邮箱验证码（注册/找回用途隔离）。"""
+    from auth_routes import _send_code_ip_check
+    ip = ratelimit.client_ip(request)
+    if not _send_code_ip_check(ip):
+        return error('验证码请求过于频繁，请稍后再试')
+    data = await _request_data(request)
+    email = str(data.get('email') or '').strip()
+    purpose = str(data.get('purpose') or 'register').strip()
+    if purpose not in ('register', 'reset'):
+        return error('验证码用途不正确')
+    try:
+        await asyncio.to_thread(auth.send_verify_code, email, purpose)
+        return success(message='验证码已发送，请查收邮箱')
+    except (ValueError, RuntimeError) as e:
+        return error(str(e))
+    except Exception:
+        return error('验证码发送失败，请稍后重试', 500)
+
+
+@router.post('/auth/trial')
+async def trial():
+    """试用已并入注册：注册即送 N 次。"""
+    return error(f'注册即送 {await _safe_register_credits()} 次免费下载次数，请直接注册账号')
+
+
+@router.post('/auth/redeem')
+async def redeem(user: Annotated[SiteUser | None, Depends(current_user)], request: Request):
+    """卡密充值。"""
+    if user is None:
+        return error('请先登录', 401, error_code='unauthorized')
+    ip = ratelimit.client_ip(request)
+    data = await _request_data(request)
+    code = str(data.get('code') or '')
+    try:
+        credits, remaining = await asyncio.to_thread(auth.redeem_card, code, user.username)
+        await asyncio.to_thread(auth.add_log, 'success', user.username, ip, '卡密充值',
+                                f'{code[:4]}**** · +{credits} 次')
+        return success({'credits': credits, 'duration_seconds': remaining,
+                        'duration_text': auth.format_credits(remaining)},
+                       f'充值成功，+{credits} 次')
+    except ValueError as e:
+        await asyncio.to_thread(auth.add_log, 'fail', user.username, ip, '卡密充值', str(e))
+        return error(str(e))
+    except Exception:
+        return error('充值失败，请稍后重试', 500)
+
+
+@router.post('/auth/forgot_password')
+async def forgot_password(request: Request):
+    """自助找回密码：邮箱验证码 → 重置密码并踢下全部会话（响应防枚举）。"""
+    data = await _request_data(request)
+    email = auth._normalize_email(str(data.get('email') or ''))
+    code = str(data.get('code') or '').strip()
+    newpwd = str(data.get('password') or '')
+    if len(newpwd) < 8 or len(newpwd) > 128:
+        return error('新密码长度需在 8-128 位之间')
+    if not email:
+        return error('缺少邮箱')
+    try:
+        if not await asyncio.to_thread(auth.check_verify_code, email, code, 'reset'):
+            return error('邮箱验证码错误或已过期')
+        user = await asyncio.to_thread(auth.get_user_by_email, email)
+        if not user:
+            return error('邮箱验证码错误或已过期')
+        if user['role'] == 'admin':
+            # 管理员不走自助找回（2FA 保护，统一防枚举文案）
+            return error('邮箱验证码错误或已过期')
+        await asyncio.to_thread(auth.reset_password, user['id'], newpwd)
+        await asyncio.to_thread(auth.add_log, 'success', user['username'],
+                                ratelimit.client_ip(request), '自助找回密码', '邮箱验证码重置')
+        return success(message='密码已重置，请使用新密码登录（所有设备已下线）')
+    except Exception:
+        return error('重置失败，请稍后重试', 500)
+
+
+@router.post('/auth/register')
+async def register(request: Request):
+    """注册：验证码校验 → 用户名/邮箱查重 → 防刷双闸 → 建号赠送 → 建立会话。"""
+    from auth_routes import re_username_ok
+
+    data = await _request_data(request)
+    username = str(data.get('username') or '').strip()
+    password = str(data.get('password') or '')
+    email = auth._normalize_email(str(data.get('email') or ''))
+    code = str(data.get('code') or '').strip()
+    device_id = str(data.get('device_id') or '').strip()[:64]
+    ip = ratelimit.client_ip(request)
+
+    if str(data.get('agree') or '').strip() != '1':
+        return error('请先阅读并同意《用户协议》与《隐私政策》')
+    if not (3 <= len(username) <= 20 and re_username_ok(username)):
+        return error('用户名需 3-20 位，仅支持中文、字母、数字、下划线，且不能纯下划线')
+    if len(password) < 8 or len(password) > 128:
+        return error('密码长度需在 8-128 位之间')
+    if len(email) > 254:
+        return error('邮箱格式不正确')
+
+    try:
+        code_ttl = await asyncio.to_thread(auth.check_verify_code, email, code, 'register')
+        if not code_ttl:
+            return error('邮箱验证码错误或已过期')
+        if await asyncio.to_thread(auth.get_user_by_name, username):
+            await asyncio.to_thread(auth.restore_verify_code, email, code, code_ttl, 'register')
+            return error('用户名已被注册')
+        if email and await asyncio.to_thread(auth.get_user_by_email, email):
+            await asyncio.to_thread(auth.restore_verify_code, email, code, code_ttl, 'register')
+            return error('该邮箱已被注册，请更换邮箱或联系管理员')
+
+        device_used_flag = bool(device_id) and await asyncio.to_thread(auth.device_used, device_id)
+        if ip and await asyncio.to_thread(auth.register_ip_exceeded, ip):
+            await asyncio.to_thread(auth.restore_verify_code, email, code, code_ttl, 'register')
+            await asyncio.to_thread(auth.add_log, 'fail', username, ip, '注册', '同 IP 24h 注册次数超限')
+            return error('当前网络注册过于频繁，请 24 小时后再试或联系管理员')
+
+        credits = await _safe_register_credits()
+        if credits < 0:
+            credits = 0
+        if device_used_flag:
+            credits = 0  # 同设备重复注册：账号照建，不再赠送
+
+        try:
+            user = await asyncio.to_thread(
+                auth.create_user, username, password, email=email, credits=credits,
+                device_id=device_id, ip=ip,
+                ua=str(request.headers.get('user-agent') or ''),
+                register_ip_limit=3, register_device_once=True)
+        except sqlite3.IntegrityError:
+            await asyncio.to_thread(auth.restore_verify_code, email, code, code_ttl, 'register')
+            return error('用户名或邮箱已被注册，请换一个再试')
+        except ValueError as e:
+            await asyncio.to_thread(auth.restore_verify_code, email, code, code_ttl, 'register')
+            await asyncio.to_thread(auth.add_log, 'fail', username, ip, '注册', str(e))
+            return error(str(e))
+
+        # 建立会话（login_seq 递增 + 签发 Cookie）
+        uid = int(user['id'])
+        now = int(time.time())
+        await db.execute('UPDATE users SET last_active=?, last_login=?, '
+                         'login_seq=COALESCE(login_seq,0)+1 WHERE id=?', (now, now, uid))
+        fresh = await repo.get_user_by_id(uid) or {}
+        seq = int(fresh.get('login_seq') or 1)
+        credits = min(credits, max(0, int(fresh.get('duration_seconds') or 0)))
+        await asyncio.to_thread(auth.add_log, 'success', username, ip, '注册账号',
+                                f'赠送 {credits} 次' if credits else '未赠送次数')
+        secure = os.environ.get('ENABLE_SECURE_COOKIE', '').strip().lower() in ('1', 'true', 'yes')
+        return _login_cookie_response({
+            'user': _user_public(fresh),
+            'duration_text': auth.format_credits(int(fresh.get('duration_seconds') or 0)),
+            'redirect': 'admin' if fresh.get('role') == 'admin' else 'user',
+        }, '注册成功' + (f'，已赠送 {credits} 次下载次数' if credits else ''), uid, seq, secure)
+    except Exception:
+        return error('注册失败，请稍后重试', 500)