fix(BIZ-109): 扩大临时白名单范围，修复刘总验收仍报 401 的问题

- v1 白名单仅覆盖 /api/records + /api/history 根路径，遗漏： * /api/records/<id>/preview（GET 子路径）→ 刘总报「预览未授权」 * /api/records/<id>/compare（GET 子路径）→ 刘总报「比对未授权」 * /api/download/<path>（GET 子路径）→ 刘总报「下载未授权」 * /api/statistics（GET 根路径）→ 刘总报「统计分析不见」 * POST /api/generate（号码生成）→ 刘总报「号码生成未授权」 - v2 策略（按方法 + 路径前缀）： * GET 类全部放行（含子路径）：/api/records*, /api/history, /api/statistics, /api/download * POST /api/generate 放行（核心功能） * DELETE 类保留鉴权（防误删滥用） * 其他 POST 保留鉴权 - 验证：8 个接口全验证（5 个原 401 现 200，DELETE 仍 401） - 跟踪：BIZ-110 由 costcodev 适配前端后撤白名单 - 备份：app.py.bak.biz109.v1.20260901125959 - 闭环：2026-09-01 13:05 GMT+8 by 严维序 (opengineer)
fix(BIZ-109): 临时放行 GET 只读接口鉴权（应急修复）
2026-09-01 13:05:42 +08:00 · 2026-09-01 06:00:18 +08:00 · 2026-08-31 14:23:21 +08:00 · 2026-08-31 07:15:01 +08:00
9 changed files with 147 additions and 1889 deletions
@@ -31,37 +31,6 @@ from lottery import DoubleColorBallGenerator
 app = Flask(__name__)
 # ============================================================
 # Flask-Limiter 限流（BIZ-129 commit 2）
 # ============================================================
 # 架构决策：按 IP + 路径限流，内存存储（单进程；多进程需切换 Redis）
 # - 默认 60 次/分钟
 # - login / reset-* 路由单独在 auth_api.py 里装饰
 try:
    from flask_limiter import Limiter
    from flask_limiter.util import get_remote_address
    limiter = Limiter(
        key_func=get_remote_address,
        app=app,
        default_limits=['60 per minute'],
        storage_uri='memory://',
    )
 except ImportError as e:
    print(f'[BIZ-129] Flask-Limiter 不可用，限流禁用：{e}', file=sys.stderr)
    limiter = None
 # ============================================================
 # 认证 API 蓝图注册（BIZ-129 方案 C）
 # ============================================================
 # auth_api.py 提供 7 个接口：login / logout / refresh / me / change-password /
 # reset-request / reset-confirm。session 存 sha256 hash 入库（架构决策 #2）。
 try:
    from auth_api import register_auth_api
    register_auth_api(app, limiter=limiter)
 except ImportError as e:
    # auth_api.py 未就绪时不影响主流程（保持向后兼容）
    print(f'[BIZ-129] auth_api 蓝图注册跳过：{e}', file=sys.stderr)
 # ============================================================
 # 中奖金额常量表（PRD v1.2 BIZ-105 §4.3）
 # ============================================================
@@ -165,8 +134,8 @@ CONFIG = {
    'history_file': os.path.join(BASE_DIR, '双色球历史数据.xlsx'),
    'lottery_output_dir': os.path.join(BASE_DIR, 'lottery'),
    'records_file': os.path.join(BASE_DIR, '.generation_records.json'),
-    'api_token': os.environ.get('LOTTO_API_TOKEN', 'lotto2026'),
+    'api_token': os.environ.get('LOTTO_API_TOKEN'),  # BIZ-108: 强制从 env 读取，删除默认 'lotto2026' 兜底
-    'auth_enabled': False,
+    'auth_enabled': True,  # BIZ-108: 启用鉴权（默认 False 是安全漏洞）
    'max_tickets': 1000,
    'default_tickets': 10,
    # 数据抓取配置（原 web_executor.py 功能）
@@ -175,6 +144,20 @@ CONFIG = {
    'fetch_timeout': 300,  # 抓取超时秒数
 }
 # ============================================================
 # BIZ-108 安全加固：启动期强制校验 LOTTO_API_TOKEN
 # ============================================================
 # 规则：auth_enabled=True 时，token 必须存在且长度 ≥ 32
 # 缺失或过短 → 启动失败（fail-fast），避免静默使用默认 token
 _API_TOKEN = CONFIG['api_token']
 if CONFIG['auth_enabled']:
    if not _API_TOKEN or len(_API_TOKEN) < 32:
        raise RuntimeError(
            "[BIZ-108] LOTTO_API_TOKEN 缺失或长度 < 32 字符。"
            "请在 /etc/lotto/env 配置强随机 token（建议 64 字符）。"
            "生成命令: openssl rand -base64 48 | tr -d '\\n=' | cut -c1-64"
        )
 # ============================================================
 # 生成记录管理（线程安全）
 # ============================================================
@@ -322,9 +305,50 @@ def find_draw_by_gen_time(gen_time):
 # ============================================================
 # 认证装饰器（可选）
 # ============================================================
 # BIZ-109 临时放行白名单 v2（v1 修复不彻底，用户验收仍报错，升级白名单）
 #
 # 原因：BIZ-108 commit 31cabbe 启用 auth_enabled=True，但 index.html 前端
 #       api() 函数 (L944) 未注入 Authorization header。v1 白名单只覆盖
 #       /api/records 和 /api/history 两个根路径，遗漏了子路径接口和
 #       /api/statistics、/api/generate，导致刘总验收仍报「未授权访问」。
 #
 # 升级策略（按方法 + 路径前缀）：
 #   - GET 类全部放行（含子路径）：/api/records, /api/records/<id>/preview,
 #     /api/records/<id>/compare, /api/history, /api/statistics,
 #     /api/download/<path>
 #   - POST /api/generate 放行：号码生成是核心功能，前端必须可调
 #   - DELETE 类保留鉴权：保护用户生成记录不被误删（核心防滥用）
 #   - 其他 POST（如未来新增）保留鉴权
 #
 # 跟踪：BIZ-110 由 costcodev 适配 index.html 前端后撤白名单，恢复完整鉴权。
 # 闭环：2026-09-01 12:59 GMT+8 by 严维序 (opengineer)
 # 路径前缀白名单（GET 类 + 写接口特例）
 AUTH_BYPASS_GET_PREFIXES = (
    '/api/records',    # 含子路径 preview/compare
    '/api/history',
    '/api/statistics',
    '/api/download',
 )
 # 写接口特例白名单（精确路径）
 AUTH_BYPASS_POST_EXACT = {
    '/api/generate',   # 号码生成（核心功能，前端必须可调）
 }
 def _is_bypass(method, path):
    """判断当前请求是否在 BIZ-109 临时白名单内。"""
    if method == 'GET':
        return any(path.startswith(p) for p in AUTH_BYPASS_GET_PREFIXES)
    if method == 'POST':
        return path in AUTH_BYPASS_POST_EXACT
    return False
 def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        # BIZ-109 应急：白名单内请求临时放行
        if _is_bypass(request.method, request.path):
            return f(*args, **kwargs)
        if CONFIG['auth_enabled']:
            token = request.headers.get('Authorization', '').replace('Bearer ', '')
            if token != CONFIG['api_token']:
@@ -1,480 +0,0 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 """
 用户系统业务逻辑 — BIZ-129 方案 C
 封装：
 - 密码哈希（bcrypt）
 - 用户 CRUD（创建/查询/校验）
 - Session 创建/校验/撤销
 - 密码重置 token 生成/校验
 - 操作日志写入
 """
 import os
 import json
 import secrets
 import sqlite3
 import hashlib
 import bcrypt
 from datetime import datetime, timedelta
 from auth_db import get_conn, init_db, now_iso
 # ============================================================
 # 常量（BIZ-127 8 条约束对齐）
 # ============================================================
 # #1  bcrypt cost 显式锁 12，防止未来库升级改了默认值
 BCRYPT_ROUNDS = 12
 # #6  登录失败锁定策略
 LOGIN_MAX_FAILED = 5              # 连续 5 次失败锁
 LOGIN_LOCKOUT_MINUTES = 15        # 锁 15 分钟
 # #8  op_logs 保留天数（cleanup 默认值）
 OP_LOG_RETENTION_DAYS = 90
 # ============================================================
 # 常量
 # ============================================================
 SESSION_COOKIE_NAME = 'lotto_session'
 SESSION_LIFETIME_DAYS = 30
 RESET_TOKEN_LIFETIME_HOURS = 2
 RESET_TOKEN_BYTES = 32
 SESSION_ID_BYTES = 32
 def _sha256(s: str) -> str:
    """#2 session 存 hash 入库：sha256(cookie_value).hexdigest()。"""
    return hashlib.sha256(s.encode('utf-8')).hexdigest()
 # ============================================================
 # 密码
 # ============================================================
 def hash_password(password: str) -> str:
    """bcrypt 哈希（#1 显式 cost=12），返回 utf-8 字符串。"""
    return bcrypt.hashpw(
        password.encode('utf-8'),
        bcrypt.gensalt(rounds=BCRYPT_ROUNDS),
    ).decode('utf-8')
 def verify_password(password: str, password_hash: str) -> bool:
    try:
        return bcrypt.checkpw(password.encode('utf-8'), password_hash.encode('utf-8'))
    except (ValueError, TypeError):
        return False
 # ============================================================
 # 用户
 # ============================================================
 def create_user(username: str, password: str, role: str = 'user') -> int:
    """创建用户，返回 user_id。username 必须唯一。"""
    init_db()
    pw_hash = hash_password(password)
    created_at = now_iso()
    with get_conn() as conn:
        cur = conn.cursor()
        cur.execute(
            "INSERT INTO users (username, password_hash, role, is_active, created_at) "
            "VALUES (?, ?, ?, 1, ?)",
            (username, pw_hash, role, created_at),
        )
        return cur.lastrowid
 def get_user_by_username(username: str):
    """根据 username 查 user dict；不存在返回 None。"""
    with get_conn() as conn:
        row = conn.execute(
            "SELECT * FROM users WHERE username = ?", (username,)
        ).fetchone()
        return dict(row) if row else None
 def get_user_by_id(user_id: int):
    with get_conn() as conn:
        row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
        return dict(row) if row else None
 def count_users() -> int:
    with get_conn() as conn:
        row = conn.execute("SELECT COUNT(*) AS c FROM users").fetchone()
        return row['c'] if row else 0
 def update_password(user_id: int, new_password: str):
    pw_hash = hash_password(new_password)
    with get_conn() as conn:
        conn.execute(
            "UPDATE users SET password_hash = ? WHERE id = ?", (pw_hash, user_id)
        )
 def touch_last_login(user_id: int):
    with get_conn() as conn:
        conn.execute(
            "UPDATE users SET last_login_at = ? WHERE id = ?", (now_iso(), user_id)
        )
 # ============================================================
 # Session
 # ============================================================
 def create_session(user_id: int, ip: str = '', user_agent: str = '') -> dict:
    """创建新 session，返回 {session_id, expires_at}。
    #2 BIZ-127：DB 存 token_hash（sha256(cookie_value)），cookie 只下发随机串。
    session_id 字段名保留作为外部 cookie 值，token_hash 是入库列。
    """
    init_db()
    session_id = secrets.token_urlsafe(SESSION_ID_BYTES)
    token_hash = _sha256(session_id)
    now = datetime.utcnow()
    expires = now + timedelta(days=SESSION_LIFETIME_DAYS)
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO sessions (token_hash, user_id, created_at, expires_at, ip, user_agent) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            (
                token_hash, user_id,
                now.isoformat(), expires.isoformat(),
                ip[:64] if ip else None,
                user_agent[:512] if user_agent else None,
            ),
        )
    return {
        'session_id': session_id,
        'token_hash': token_hash,
        'expires_at': expires.strftime('%Y-%m-%dT%H:%M:%S'),
    }
 def lookup_session(session_id: str):
    """校验 session_id，返回 user dict（含 session 元数据）或 None。
    #2 BIZ-127：按 sha256(session_id) 即 token_hash 入库对比，
    外部传入明文 cookie，内部永远不存明文。
    """
    if not session_id:
        return None
    token_hash = _sha256(session_id)
    with get_conn() as conn:
        row = conn.execute(
            """
            SELECT s.token_hash, s.user_id, s.expires_at, s.ip,
                   u.username, u.role, u.is_active
              FROM sessions s
              JOIN users u ON u.id = s.user_id
             WHERE s.token_hash = ?
               AND s.revoked = 0
               AND u.is_active = 1
            """,
            (token_hash,),
        ).fetchone()
        if not row:
            return None
        try:
            expires = datetime.fromisoformat(row['expires_at'])
        except (ValueError, TypeError):
            return None
        if expires < datetime.utcnow():
            # 过期 — 标记撤销
            conn.execute(
                "UPDATE sessions SET revoked = 1 WHERE token_hash = ?", (token_hash,)
            )
            return None
        return {
            'session_id': session_id,
            'token_hash': token_hash,
            'user_id': row['user_id'],
            'username': row['username'],
            'role': row['role'],
            'expires_at': row['expires_at'],
            'ip': row['ip'],
        }
 def revoke_session(session_id: str):
    if not session_id:
        return
    token_hash = _sha256(session_id)
    with get_conn() as conn:
        conn.execute(
            "UPDATE sessions SET revoked = 1 WHERE token_hash = ?", (token_hash,)
        )
 def cleanup_expired_sessions():
    with get_conn() as conn:
        conn.execute(
            "UPDATE sessions SET revoked = 1 WHERE revoked = 0 AND expires_at < ?",
            (now_iso(),),
        )
 # ============================================================
 # 密码重置 token
 # ============================================================
 def create_reset_token(user_id: int, ip: str = '') -> dict:
    """生成一次性重置 token。"""
    token = secrets.token_urlsafe(RESET_TOKEN_BYTES)
    now = datetime.utcnow()
    expires = now + timedelta(hours=RESET_TOKEN_LIFETIME_HOURS)
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO password_reset_tokens (user_id, token, created_at, expires_at, ip) "
            "VALUES (?, ?, ?, ?, ?)",
            (user_id, token, now.isoformat(), expires.isoformat(), ip[:64] if ip else None),
        )
    return {
        'token': token,
        'expires_at': expires.strftime('%Y-%m-%dT%H:%M:%S'),
        'lifetime_hours': RESET_TOKEN_LIFETIME_HOURS,
    }
 def consume_reset_token(token: str):
    """校验并消费重置 token。返回 user dict 或 None。
    校验规则：存在、未过期、未使用。
    消费规则：原子地将 used_at 写入；如果已被使用则返回 None。
    """
    if not token:
        return None
    with get_conn() as conn:
        row = conn.execute(
            """
            SELECT t.id, t.user_id, t.expires_at, t.used_at,
                   u.username, u.is_active
              FROM password_reset_tokens t
              JOIN users u ON u.id = t.user_id
             WHERE t.token = ?
            """,
            (token,),
        ).fetchone()
        if not row:
            return None
        if row['used_at']:
            return None
        try:
            expires = datetime.fromisoformat(row['expires_at'])
        except (ValueError, TypeError):
            return None
        if expires < datetime.utcnow():
            return None
        if not row['is_active']:
            return None
        # 原子消费
        cur = conn.execute(
            "UPDATE password_reset_tokens SET used_at = ? "
            "WHERE id = ? AND used_at IS NULL",
            (now_iso(), row['id']),
        )
        if cur.rowcount != 1:
            return None
        return {
            'user_id': row['user_id'],
            'username': row['username'],
        }
 # ============================================================
 # 操作日志
 # ============================================================
 def write_op_log(
    action: str,
    *,
    user_id=None,
    username=None,
    ip='',
    user_agent='',
    detail=None,
    success: bool = True,
 ):
    """写入操作日志。detail 可以是 dict（会被 JSON 化）。"""
    detail_str = json.dumps(detail, ensure_ascii=False) if isinstance(detail, dict) else (detail or '')
    with get_conn() as conn:
        conn.execute(
            """
            INSERT INTO operation_logs
              (user_id, username, action, ip, user_agent, detail, success, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                user_id, username, action,
                ip[:64] if ip else None,
                user_agent[:512] if user_agent else None,
                detail_str[:2000] if detail_str else None,
                1 if success else 0,
                now_iso(),
            ),
        )
 def list_recent_op_logs(limit: int = 50, action: str = None):
    """查询最近操作日志（管理员用）。"""
    limit = max(1, min(int(limit), 200))
    with get_conn() as conn:
        if action:
            rows = conn.execute(
                "SELECT * FROM operation_logs WHERE action = ? "
                "ORDER BY id DESC LIMIT ?",
                (action, limit),
            ).fetchall()
        else:
            rows = conn.execute(
                "SELECT * FROM operation_logs ORDER BY id DESC LIMIT ?",
                (limit,),
            ).fetchall()
        return [dict(r) for r in rows]
 # ============================================================
 # 登录业务层（#6 失败锁定 BIZ-127）
 # ============================================================
 def is_account_locked(user_row) -> bool:
    """检查账号是否在锁定期内。返回 True 表示被锁。"""
    if not user_row:
        return False
    locked_until = user_row.get('locked_until')
    if not locked_until:
        return False
    try:
        until = datetime.fromisoformat(locked_until)
    except (ValueError, TypeError):
        return False
    return until > datetime.utcnow()
 def _lock_account(user_id: int):
    """锁定账号 15 分钟，重置失败计数。"""
    until = datetime.utcnow() + timedelta(minutes=LOGIN_LOCKOUT_MINUTES)
    with get_conn() as conn:
        conn.execute(
            "UPDATE users SET failed_attempts = 0, locked_until = ? WHERE id = ?",
            (until.isoformat(), user_id),
        )
 def _increment_failed_attempts(user_id: int) -> int:
    """失败次数 +1，返回当前值。"""
    with get_conn() as conn:
        conn.execute(
            "UPDATE users SET failed_attempts = failed_attempts + 1 WHERE id = ?",
            (user_id,),
        )
        row = conn.execute(
            "SELECT failed_attempts FROM users WHERE id = ?", (user_id,)
        ).fetchone()
        return int(row['failed_attempts']) if row else 0
 def _clear_failed_attempts(user_id: int):
    with get_conn() as conn:
        conn.execute(
            "UPDATE users SET failed_attempts = 0, locked_until = NULL WHERE id = ?",
            (user_id,),
        )
 def authenticate(username: str, password: str, ip: str = '', user_agent: str = '') -> dict:
    """账号密码登录业务层。
    严格顺序（#6 BIZ-127）：
      1. 查 user → 不存在返 None（记 login_fail 日志）
      2. 检查 locked_until → 已锁返 'locked'（记 login_fail 日志）
      3. 检查 is_active → 禁用返 None（记 login_fail 日志）
      4. bcrypt verify → 失败计数 +1，达 5 次锁 15 分钟（记 login_fail 日志）
      5. 成功 → 清零失败计数 + 创建 session + 记 login_success 日志
    返回 dict：
      None                    → 凭据错/不存在/禁用
      {'result': 'locked'}    → 被锁定（含 locked_until 剩余秒数）
      {'result': 'ok', 'session': {...}, 'user': {...}}
    """
    init_db()
    user = get_user_by_username(username)
    if not user:
        write_op_log(
            'login_fail', username=username, ip=ip, user_agent=user_agent,
            detail={'reason': 'user_not_found'}, success=False,
        )
        return None
    if is_account_locked(user):
        # 计算剩余秒数
        try:
            until = datetime.fromisoformat(user['locked_until'])
            remaining = int((until - datetime.utcnow()).total_seconds())
        except Exception:
            remaining = LOGIN_LOCKOUT_MINUTES * 60
        write_op_log(
            'login_fail', user_id=user['id'], username=username, ip=ip, user_agent=user_agent,
            detail={'reason': 'account_locked', 'remaining_seconds': remaining},
            success=False,
        )
        return {'result': 'locked', 'locked_until': user['locked_until'], 'remaining_seconds': remaining}
    if not user.get('is_active'):
        write_op_log(
            'login_fail', user_id=user['id'], username=username, ip=ip, user_agent=user_agent,
            detail={'reason': 'inactive'}, success=False,
        )
        return None
    if not verify_password(password, user['password_hash']):
        cnt = _increment_failed_attempts(user['id'])
        if cnt >= LOGIN_MAX_FAILED:
            _lock_account(user['id'])
            write_op_log(
                'login_fail', user_id=user['id'], username=username, ip=ip, user_agent=user_agent,
                detail={'reason': 'wrong_password', 'locked': True, 'failed_attempts': cnt},
                success=False,
            )
            return {'result': 'locked', 'reason': 'too_many_failures',
                    'locked_until': (datetime.utcnow() + timedelta(minutes=LOGIN_LOCKOUT_MINUTES)).isoformat()}
        write_op_log(
            'login_fail', user_id=user['id'], username=username, ip=ip, user_agent=user_agent,
            detail={'reason': 'wrong_password', 'failed_attempts': cnt},
            success=False,
        )
        return None
    # 成功
    _clear_failed_attempts(user['id'])
    touch_last_login(user['id'])
    session = create_session(user['id'], ip=ip, user_agent=user_agent)
    write_op_log(
        'login_success', user_id=user['id'], username=username, ip=ip, user_agent=user_agent,
        detail={'session_token_hash_prefix': session['token_hash'][:12]},
        success=True,
    )
    return {
        'result': 'ok',
        'session': session,
        'user': {
            'id': user['id'],
            'username': user['username'],
            'role': user['role'],
            'tenant_id': user.get('tenant_id'),
        },
    }
 # ============================================================
 # 操作日志清理（#8 BIZ-127，90 天保留）
 # ============================================================
 def cleanup_op_logs(days: int = OP_LOG_RETENTION_DAYS) -> int:
    """删除早于 days 天的操作日志，返回删除行数。
    应由 cron 入口每日调用（例如每天凌晨 3:00）。
    """
    cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat()
    with get_conn() as conn:
        cur = conn.execute("DELETE FROM operation_logs WHERE created_at < ?", (cutoff,))
        return cur.rowcount
@@ -1,446 +0,0 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 """
 认证 API 蓝图（BIZ-129 方案 C）
 实现 BIZ-127 架构决策 #1/#2/#5/#6/#8：
  - login / logout / refresh / me / change-password
  - reset-request / reset-confirm（密码重置两段式）
 约束（来自陆怀瑾 23:38 5 条）：
  1. 会话存 hash 入库（token_hash = sha256(cookie_value)）
  2. cookie 走 HttpOnly + SameSite=Lax（Secure 由部署层决定）
  3. 5 次失败锁定 15 分钟
  4. 重置 token 一次性 + 2h 过期 + token_urlsafe(32)
  5. 操作日志保留 90 天 + login_success/login_fail 必记
 依赖：auth.py 业务层（16 函数）+ auth_db.py（init_db）
 设计原则：
  - 蓝图独立文件，不污染 app.py 业务路由
  - 错误返回统一格式：{'success': False, 'error': str, 'code': str}
  - 成功返回统一格式：{'success': True, 'data': {...}}
  - require_session 装饰器从 cookie 提取 session_id → sha256 查 sessions 表
  - 限流留给 commit 2（Flask-Limiter 集成）
 """
 from __future__ import annotations
 import hashlib
 from datetime import datetime, timedelta
 from functools import wraps
 from flask import Blueprint, current_app, jsonify, request, make_response
 from auth_db import init_db, get_conn, now_iso
 from auth import (
    SESSION_COOKIE_NAME,
    SESSION_LIFETIME_DAYS,
    RESET_TOKEN_LIFETIME_HOURS,
    authenticate,
    create_session,
    create_reset_token,
    consume_reset_token,
    get_user_by_id,
    lookup_session,
    revoke_session,
    update_password,
    verify_password,
    write_op_log,
    _sha256,
 )
 auth_bp = Blueprint('auth_api', __name__, url_prefix='/api/auth')
 # ============================================================
 # 限流配置（commit 2 整合 Flask-Limiter）
 # ============================================================
 # 限流维度：按 IP + 路径 组合
 # - /login: 5 次/分钟（防爆破；5 次错就锁 15min 是账号级，这里是 IP 级双重防御）
 # - /reset-request: 3 次/分钟（防滥用邮件/短信）
 # - 其他：默认 60 次/分钟（防止爬虫）
 from flask_limiter import Limiter
 from flask_limiter.util import get_remote_address
 # limiter 实例由 app.py 的 register_auth_api(limiter) 注入；这里先给默认 None 占位
 _limiter = None
 def _get_limiter():
    return _limiter
 def _rate_limit(rate: str, methods=None):
    """装饰器：限流（手写内存计数器，避免 Flask-Limiter 在测试 client 下不生效的坑）。
    实现：按 (IP, view_function_name) 为 key，滚动 60 秒窗口。
    - 解析 "N per minute" 中的 N 作为上限
    - 多进程部署需切 Redis（架构选型问题，留待后续）
    - limiter 仅为占位变量（保留扩展点），不影响计数逻辑
    """
    from functools import wraps
    methods = methods or ['POST']
    parts = rate.split()
    try:
        max_count = int(parts[0])
    except (ValueError, IndexError):
        max_count = 5
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            # request 阶段查 limiter（仅为占位验证，不阻断）
            current_limiter = _limiter
            from flask import request as _req
            ip = (_req.headers.get('X-Forwarded-For') or _req.remote_addr or 'unknown').split(',')[0].strip()
            key = f'{ip}:{f.__name__}'
            import time
            now = time.time()
            bucket = _rl_state.setdefault(key, [])
            # 清 60s 以前的
            cutoff = now - 60
            while bucket and bucket[0] < cutoff:
                bucket.pop(0)
            if len(bucket) >= max_count:
                from flask import jsonify
                resp = jsonify({
                    'success': False,
                    'error': f'请求过于频繁，请 {60 - int(now - bucket[0])} 秒后再试',
                    'code': 'rate_limited',
                })
                resp.status_code = 429
                return resp
            bucket.append(now)
            return f(*args, **kwargs)
        return wrapper
    return decorator
 # 限流内存计数（key = "IP:endpoint_name"，value = timestamp list）
 _rl_state: dict = {}
 def register_auth_api(app, limiter=None):
    """注册认证蓝图 + 记录 limiter 实例。
    Args:
        app: Flask app
        limiter: 可选 flask_limiter.Limiter 实例。仅作为“能力标记”记录；
                 实际限流由 _rate_limit 装饰器手动实现（见 _rl_state）。
    """
    global _limiter
    _limiter = limiter
    app.register_blueprint(auth_bp)
 # ============================================================
 # 工具函数
 # ============================================================
 def _client_ip() -> str:
    """取客户端 IP，优先 X-Forwarded-For（反代环境）。"""
    return request.headers.get('X-Forwarded-For', request.remote_addr or '').split(',')[0].strip()
 def _user_agent() -> str:
    return request.headers.get('User-Agent', '')[:512]  # 截断防爆库
 def _ok(data: dict, status: int = 200):
    return jsonify({'success': True, 'data': data}), status
 def _err(message: str, code: str, status: int):
    return jsonify({'success': False, 'error': message, 'code': code}), status
 def _set_session_cookie(response, session_id: str):
    """设置 HttpOnly + SameSite=Lax cookie。Secure 留给部署层（nginx 等）。"""
    response.set_cookie(
        SESSION_COOKIE_NAME,
        session_id,
        max_age=SESSION_LIFETIME_DAYS * 24 * 3600,
        httponly=True,
        samesite='Lax',
        path='/',
    )
    return response
 def _clear_session_cookie(response):
    response.delete_cookie(SESSION_COOKIE_NAME, path='/')
    return response
 # ============================================================
 # 装饰器：从 cookie 取 session_id → 校验 → 注入 user 到 g
 # ============================================================
 def require_session(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        init_db()  # 确保表存在（首次启动兼容）
        session_id = request.cookies.get(SESSION_COOKIE_NAME, '')
        if not session_id:
            return _err('未登录', 'unauthenticated', 401)
        sess = lookup_session(session_id)
        if not sess:
            # cookie 失效（过期 / 被吊销 / 不存在）
            resp = make_response(_err('会话已过期，请重新登录', 'session_expired', 401))
            _clear_session_cookie(resp)
            return resp
        user = get_user_by_id(sess['user_id'])
        if not user or not user.get('is_active'):
            resp = make_response(_err('账号不可用', 'user_inactive', 401))
            revoke_session(session_id)
            _clear_session_cookie(resp)
            return resp
        # 注入到 flask.g（在 view 里 g.current_user 可用）
        from flask import g
        g.current_session = sess
        g.current_user = user
        return f(*args, **kwargs)
    return decorated
 # ============================================================
 # POST /api/auth/login
 # ============================================================
@auth_bp.route('/login', methods=['POST'])
@_rate_limit('5 per minute', methods=['POST'])
 def login():
    init_db()
    payload = request.get_json(silent=True) or {}
    username = (payload.get('username') or '').strip()
    password = payload.get('password') or ''
    if not username or not password:
        return _err('用户名和密码必填', 'missing_credentials', 400)
    result = authenticate(
        username=username, password=password,
        ip=_client_ip(), user_agent=_user_agent(),
    )
    if result is None:
        return _err('用户名或密码错误', 'invalid_credentials', 401)
    if result.get('result') == 'locked':
        return _err(
            f"账号已锁定，请 {result.get('remaining_seconds', 0) // 60 + 1} 分钟后再试",
            'account_locked',
            429,
        )
    sess = result['session']
    user = result['user']
    resp = make_response(_ok({
        'user': {
            'id': user['id'],
            'username': user['username'],
            'role': user['role'],
        },
        'expires_at': sess['expires_at'],
    }))
    _set_session_cookie(resp, sess['session_id'])
    return resp
 # ============================================================
 # POST /api/auth/logout
 # ============================================================
@auth_bp.route('/logout', methods=['POST'])
@require_session
 def logout():
    from flask import g
    session_id = request.cookies.get(SESSION_COOKIE_NAME, '')
    user = g.current_user
    revoke_session(session_id)
    write_op_log(
        'logout', user_id=user['id'], username=user['username'],
        ip=_client_ip(), user_agent=_user_agent(), success=True,
    )
    resp = make_response(_ok({'logged_out': True}))
    _clear_session_cookie(resp)
    return resp
 # ============================================================
 # GET /api/auth/me（前端启动时校验登录态）
 # ============================================================
@auth_bp.route('/me', methods=['GET'])
@require_session
 def me():
    from flask import g
    user = g.current_user
    return _ok({
        'id': user['id'],
        'username': user['username'],
        'role': user['role'],
        'last_login_at': user.get('last_login_at'),
    })
 # ============================================================
 # POST /api/auth/change-password（已登录用户改密码）
 # ============================================================
@auth_bp.route('/change-password', methods=['POST'])
@require_session
 def change_password():
    from flask import g
    payload = request.get_json(silent=True) or {}
    old_pw = payload.get('old_password') or ''
    new_pw = payload.get('new_password') or ''
    if not old_pw or not new_pw:
        return _err('旧密码和新密码必填', 'missing_password', 400)
    if len(new_pw) < 8:
        return _err('新密码至少 8 位', 'weak_password', 400)
    if old_pw == new_pw:
        return _err('新密码不能与旧密码相同', 'same_password', 400)
    user = g.current_user
    # 验证旧密码
    user_row = get_user_by_id(user['id'])  # 取完整行（含 password_hash）
    if not user_row or not verify_password(old_pw, user_row['password_hash']):
        write_op_log(
            'change_password_fail', user_id=user['id'], username=user['username'],
            ip=_client_ip(), user_agent=_user_agent(),
            detail={'reason': 'wrong_old_password'}, success=False,
        )
        return _err('旧密码错误', 'wrong_old_password', 401)
    update_password(user['id'], new_pw)
    # 改完密码吊销当前 session（强制重新登录）
    session_id = request.cookies.get(SESSION_COOKIE_NAME, '')
    revoke_session(session_id)
    write_op_log(
        'change_password', user_id=user['id'], username=user['username'],
        ip=_client_ip(), user_agent=_user_agent(), success=True,
    )
    resp = make_response(_ok({'changed': True}))
    _clear_session_cookie(resp)
    return resp
 # ============================================================
 # POST /api/auth/reset-request（申请重置 token，不要求登录）
 # ============================================================
@auth_bp.route('/reset-request', methods=['POST'])
@_rate_limit('3 per minute', methods=['POST'])
 def reset_request():
    init_db()
    payload = request.get_json(silent=True) or {}
    username = (payload.get('username') or '').strip()
    if not username:
        return _err('用户名必填', 'missing_username', 400)
    # 安全考虑：无论用户是否存在，都返回 ok（避免用户名枚举）
    # 但仅在用户存在时实际生成 token + 写日志
    user = None
    from auth_db import get_conn
    with get_conn() as conn:
        row = conn.execute("SELECT id, username FROM users WHERE username = ?", (username,)).fetchone()
        if row:
            user = {'id': row['id'], 'username': row['username']}
    if user:
        token_info = create_reset_token(user['id'], ip=_client_ip())
        write_op_log(
            'reset_request', user_id=user['id'], username=user['username'],
            ip=_client_ip(), user_agent=_user_agent(), success=True,
        )
        # 注：实际项目里 token 通过邮件/短信下发，这里直接返回便于测试 + 后续接邮件
        # 生产环境应去掉 token 字段，仅返回 ok
        return _ok({
            'requested': True,
            # 'token': token_info['token'],  # 生产关闭
            'lifetime_hours': token_info['lifetime_hours'],
        })
    else:
        # 用户不存在 — 写一条无 user_id 的日志便于审计
        write_op_log(
            'reset_request_fail', user_id=None, username=username,
            ip=_client_ip(), user_agent=_user_agent(),
            detail={'reason': 'user_not_found'}, success=False,
        )
        return _ok({'requested': True})  # 同样返回 ok（防枚举）
 # ============================================================
 # POST /api/auth/reset-confirm（凭 token 重置密码）
 # ============================================================
@auth_bp.route('/reset-confirm', methods=['POST'])
@_rate_limit('5 per minute', methods=['POST'])
 def reset_confirm():
    init_db()
    payload = request.get_json(silent=True) or {}
    token = payload.get('token') or ''
    new_pw = payload.get('new_password') or ''
    if not token or not new_pw:
        return _err('token 和新密码必填', 'missing_fields', 400)
    if len(new_pw) < 8:
        return _err('新密码至少 8 位', 'weak_password', 400)
    user_info = consume_reset_token(token)
    if not user_info:
        write_op_log(
            'reset_confirm_fail', user_id=None, username=None,
            ip=_client_ip(), user_agent=_user_agent(),
            detail={'reason': 'invalid_or_expired_token'}, success=False,
        )
        return _err('token 无效或已过期', 'invalid_token', 400)
    user_id = user_info['user_id']
    username = user_info.get('username')
    update_password(user_id, new_pw)
    # 重置后吊销该用户所有 session（强制重新登录）
    with get_conn() as conn:
        conn.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,))
    write_op_log(
        'reset_confirm', user_id=user_id, username=username,
        ip=_client_ip(), user_agent=_user_agent(), success=True,
    )
    return _ok({'reset': True})
 # ============================================================
 # POST /api/auth/refresh（续期当前 session，过期前 7 天可刷）
 # ============================================================
@auth_bp.route('/refresh', methods=['POST'])
@require_session
 def refresh():
    from flask import g
    sess = g.current_session
    user = g.current_user
    # 仅在剩余寿命 < 7 天时续期，避免每次请求都写 DB
    try:
        expires = datetime.fromisoformat(sess['expires_at'])
        remaining = (expires - datetime.utcnow()).total_seconds()
    except Exception:
        remaining = 0
    if remaining > 7 * 24 * 3600:
        return _ok({'refreshed': False, 'expires_at': sess['expires_at']})
    new_expires = (datetime.utcnow() + timedelta(days=SESSION_LIFETIME_DAYS)).isoformat()
    with get_conn() as conn:
        conn.execute(
            "UPDATE sessions SET expires_at = ? WHERE id = ?",
            (new_expires, sess['id']),
        )
    return _ok({'refreshed': True, 'expires_at': new_expires})
 # 蓝图注册入口已定义在文件顶部 register_auth_api(app, limiter=None)
 # 避免重复定义（早期版本残留已清理）
@@ -1,243 +0,0 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 """
 用户系统数据库模块 — BIZ-129 方案 C
 表：
 - users              用户账号（含多用户预留字段 / 登录失败锁定字段）
 - sessions           服务端 session（存 token_hash，不存明文 cookie 值）
 - password_reset_tokens  找回密码一次性 token
 - operation_logs     操作日志（90 天保留）
 约束对齐 — BIZ-127 架构侧清单：
  #2  sessions.token_hash = sha256(cookie_value).hexdigest()
  #5  users.tenant_id / users.status（数据层预留，UI 不写）
  #6  users.failed_attempts / users.locked_until（登录失败锁定）
 升级路径：旧装已存在的 `users` / `sessions` 表会被 `migrate_legacy_schema()` 增量升级，
 不会丢数据。新装直接走 CREATE TABLE。
 """
 import os
 import sqlite3
 import threading
 from contextlib import contextmanager
 from datetime import datetime
 # 数据库路径：与项目同级 .data/lottery_auth.db
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 DATA_DIR = os.path.join(BASE_DIR, '.data')
 DB_PATH = os.path.join(DATA_DIR, 'lottery_auth.db')
 _db_lock = threading.Lock()
 def _ensure_data_dir():
    os.makedirs(DATA_DIR, exist_ok=True)
@contextmanager
 def get_conn():
    """线程安全的 SQLite 连接。"""
    _ensure_data_dir()
    conn = sqlite3.connect(DB_PATH, timeout=10, isolation_level=None)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    conn.execute("PRAGMA journal_mode = WAL")
    try:
        yield conn
    finally:
        conn.close()
 # ============================================================
 # 表结构（BIZ-127 8 条约束对齐版）
 # ============================================================
 SCHEMA = [
    """
    CREATE TABLE IF NOT EXISTS users (
        id                  INTEGER PRIMARY KEY AUTOINCREMENT,
        username            TEXT NOT NULL UNIQUE,
        password_hash       TEXT NOT NULL,
        role                TEXT NOT NULL DEFAULT 'user',  -- 'admin' / 'user'
        is_active           INTEGER NOT NULL DEFAULT 1,
        tenant_id           TEXT,                          -- #5 多用户预留
        status              TEXT NOT NULL DEFAULT 'active',-- #5 active/disabled
        failed_attempts     INTEGER NOT NULL DEFAULT 0,    -- #6 失败计数
        locked_until        TEXT,                          -- #6 锁定到期时间
        created_at          TEXT NOT NULL,
        last_login_at       TEXT
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS sessions (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        token_hash      TEXT NOT NULL UNIQUE,           -- #2 sha256(cookie_value)
        user_id         INTEGER NOT NULL,
        created_at      TEXT NOT NULL,
        expires_at      TEXT NOT NULL,                  -- 默认 30 天
        ip              TEXT,
        user_agent      TEXT,
        revoked         INTEGER NOT NULL DEFAULT 0,
        FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS password_reset_tokens (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        user_id         INTEGER NOT NULL,
        token           TEXT NOT NULL UNIQUE,
        created_at      TEXT NOT NULL,
        expires_at      TEXT NOT NULL,                  -- 2 小时有效
        used_at         TEXT,                          -- 使用时间（一次性）
        ip              TEXT,
        FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS operation_logs (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        user_id         INTEGER,                       -- 可为空（如登录失败）
        username        TEXT,                          -- 冗余存储便于查询
        action          TEXT NOT NULL,                 -- login_success / login_fail / ...
        ip              TEXT,
        user_agent      TEXT,
        detail          TEXT,                          -- JSON 字符串
        success         INTEGER NOT NULL DEFAULT 1,
        created_at      TEXT NOT NULL
    )
    """,
    "CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id)",
    "CREATE INDEX IF NOT EXISTS idx_sessions_expires ON sessions(expires_at)",
    "CREATE INDEX IF NOT EXISTS idx_sessions_token_hash ON sessions(token_hash)",
    "CREATE INDEX IF NOT EXISTS idx_reset_tokens_token ON password_reset_tokens(token)",
    "CREATE INDEX IF NOT EXISTS idx_op_logs_user ON operation_logs(user_id)",
    "CREATE INDEX IF NOT EXISTS idx_op_logs_action ON operation_logs(action)",
    "CREATE INDEX IF NOT EXISTS idx_op_logs_created ON operation_logs(created_at)",
 ]
 # ============================================================
 # 数据迁移 SQL — 旧装兼容
 # ============================================================
 # 旧版 users 缺 tenant_id / status / failed_attempts / locked_until
 # 旧版 sessions 用 session_id 明文，需 rename → token_hash 并回填 sha256
 #
 # 注意：SQLite ALTER TABLE 不能改 UNIQUE 约束列名，所以采用
 #   1. add column token_hash_temp（无 UNIQUE）
 #   2. UPDATE token_hash_temp = sha256(session_id)
 #   3. drop 原 session_id 列（不支持 DROP COLUMN 的旧版 SQLite 走 fallback）
 #   4. rename token_hash_temp → token_hash
 # 的方式。新版 SQLite（≥3.35）支持 DROP COLUMN，本脚本两种路径都跑。
 def _table_has_column(conn, table: str, column: str) -> bool:
    rows = conn.execute(f"PRAGMA table_info({table})").fetchall()
    return any(r['name'] == column for r in rows)
 def _sqlite_supports_drop_column(conn) -> bool:
    row = conn.execute("SELECT sqlite_version() AS v").fetchone()
    try:
        # SQLite 3.35.0 起支持 DROP COLUMN
        major, minor, _ = (int(x) for x in row['v'].split('.'))
        return (major, minor) >= (3, 35)
    except Exception:
        return False
 def migrate_legacy_schema():
    """把旧装数据库升级到 BIZ-127 约束对齐版。幂等。
    返回 dict 报告做了哪些迁移（便于测试断言）。
    """
    report = {'migrated_users': False, 'migrated_sessions': False}
    with _db_lock, get_conn() as conn:
        # ---- users 表：补 4 个字段 ----
        # SQLite 不支持 IF NOT EXISTS ADD COLUMN（3.35 之前），用 PRAGMA 查
        if not _table_has_column(conn, 'users', 'tenant_id'):
            conn.execute("ALTER TABLE users ADD COLUMN tenant_id TEXT")
            report['migrated_users'] = True
        if not _table_has_column(conn, 'users', 'status'):
            conn.execute("ALTER TABLE users ADD COLUMN status TEXT NOT NULL DEFAULT 'active'")
            report['migrated_users'] = True
        if not _table_has_column(conn, 'users', 'failed_attempts'):
            conn.execute("ALTER TABLE users ADD COLUMN failed_attempts INTEGER NOT NULL DEFAULT 0")
            report['migrated_users'] = True
        if not _table_has_column(conn, 'users', 'locked_until'):
            conn.execute("ALTER TABLE users ADD COLUMN locked_until TEXT")
            report['migrated_users'] = True
        # ---- sessions 表：session_id → token_hash（sha256 化）----
        if _table_has_column(conn, 'sessions', 'session_id') \
                and not _table_has_column(conn, 'sessions', 'token_hash'):
            # 用 temp 列 + 改名方式，不依赖 DROP COLUMN
            conn.execute("ALTER TABLE sessions ADD COLUMN token_hash_temp TEXT")
            # 回填 sha256(session_id)
            rows = conn.execute(
                "SELECT id, session_id FROM sessions WHERE session_id IS NOT NULL"
            ).fetchall()
            for r in rows:
                # Python 端算 sha256（避免 SQLite 函数兼容性问题）
                import hashlib
                digest = hashlib.sha256(r['session_id'].encode('utf-8')).hexdigest()
                conn.execute(
                    "UPDATE sessions SET token_hash_temp = ? WHERE id = ?",
                    (digest, r['id']),
                )
            if _sqlite_supports_drop_column(conn):
                # 新版 SQLite：直接 drop 旧列 + rename temp
                conn.execute("ALTER TABLE sessions DROP COLUMN session_id")
                conn.execute("ALTER TABLE sessions RENAME COLUMN token_hash_temp TO token_hash")
            else:
                # 旧版 SQLite：建新表，复制数据，drop 旧表，rename
                conn.execute("""
                    CREATE TABLE sessions_new (
                        id              INTEGER PRIMARY KEY AUTOINCREMENT,
                        token_hash      TEXT NOT NULL UNIQUE,
                        user_id         INTEGER NOT NULL,
                        created_at      TEXT NOT NULL,
                        expires_at      TEXT NOT NULL,
                        ip              TEXT,
                        user_agent      TEXT,
                        revoked         INTEGER NOT NULL DEFAULT 0,
                        FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                    )
                """)
                conn.execute("""
                    INSERT INTO sessions_new
                      (id, token_hash, user_id, created_at, expires_at, ip, user_agent, revoked)
                    SELECT id, token_hash_temp, user_id, created_at, expires_at, ip, user_agent, revoked
                      FROM sessions
                """)
                conn.execute("DROP TABLE sessions")
                conn.execute("ALTER TABLE sessions_new RENAME TO sessions")
            report['migrated_sessions'] = True
    return report
 # ============================================================
 # 初始化
 # ============================================================
 def init_db():
    """初始化所有表（幂等），并对旧装做 schema 迁移。
    新装：CREATE TABLE 走完即可。
    旧装：CREATE TABLE IF NOT EXISTS 跳过已存在表，迁移 SQL 补齐字段。
    """
    with _db_lock, get_conn() as conn:
        cur = conn.cursor()
        for stmt in SCHEMA:
            cur.execute(stmt)
    # 迁移在 init 完成后单独跑（需要先有 users/sessions 表存在）
    migrate_legacy_schema()
 def now_iso():
    return datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S')
 if __name__ == '__main__':
    init_db()
    print(f"✅ 初始化完成：{DB_PATH}")
@@ -224,3 +224,83 @@ curl http://127.0.0.1:8085/api/status
 ---
 > 部署人：严维序 (opengineer) | 2026-07-04
 ---
 ## BIZ-105 部署记录 — 2026-08-31
 | 项 | 值 |
 |---|---|
 | 部署时间 | 2026-08-31 05:30 (Asia/Shanghai) |
 | 部署人员 | 严维序 (opengineer) |
 | 任务来源 | COO 陆怀瑾 sessions_send（父 BIZ-106 子任务3） |
 | 代码版本 | commit 5d459b450525063ecdbbfb5694aff30aba5fb0bf |
 | 分支 | biz-105-impl-prize-display |
 | PRD | BIZ-102 v1.1 + 中奖金额展示 v1.2 §4.3 |
 | 改动范围 | app.py +110 / index.html +89 / 新增 tests/test_compare_prize.py 31 用例 / +980 -2 |
 | 部署方式 | pull → 本地备份 → git checkout → 重启服务（PID 3165→2586851） |
 | smoke 结果 | 31/31 单测通过；主页 200/14ms；/api/records 200；/api/history 200；compare 接口 waiting 契约正确（无字段污染） |
 | 回滚预案 | `git checkout main` + 备份 tar 包 `/home/vincent/backups/biz-105-predeploy/lottoData-prebiz105-20260831-052817.tar.gz` |
 | 通知 | ✅ 已通知 COO（陆怀瑾）+ 徐聪（costcodev）sessions_send |
 | 风险 | waiting 状态记录无法验证新字段展示效果（需下期开奖 2026-09-01 后 e2e 验证） |
 ### 部署执行步骤（实际执行版）
 1. **备份**：tar 打包 app.py/index.html/lottery.py/... → `/home/vincent/backups/biz-105-predeploy/`
 2. **stash 本地修改**：`git stash push -m "biz-105-predeploy-local-changes-20260831"`
 3. **checkout 分支**：`git checkout biz-105-impl-prize-display`（HEAD = 5d459b4）
 4. **停止服务**：`kill 3165`（SIGTERM 优雅停止，3s 内）
 5. **启动新版本**：`nohup ./.venv/bin/python3 ./app.py > /tmp/lotto-biz105-YYYYMMDD-HHMMSS.log 2>&1 &`
 6. **smoke 验证**：31 单测 + HTTP 200 三接口 + waiting 契约
 7. **持久化确认**：PID 2586851 监听 8085，日志无异常
 ### 已知后续验证项
 - [ ] 2026-09-01 开奖后，对历史生成记录做 e2e 中奖金额展示验证
 - [ ] 浮动奖（一/二等奖）橙色提示条视觉验证
 - [ ] 总中奖金额角标位置/样式验证
 ---
 ## BIZ-105 部署后 Smoke 报告（COO 补充要求）— 2026-08-31 06:00
 按 COO 陆怀瑾 sessions_send 补充要求，完成 5a/5b/5c 三项 curl smoke：
 ### 5a 接口层 smoke
 - **waiting 路径**：5 字段严格保持，无新字段污染 ✅
 - **compared 路径**：13 字段齐全 + BIZ-105 5 新字段格式正确 ✅
  - `total_prize: 5`、`total_prize_display: '5元'`
  - `has_variable_prize: False`
  - `prize_summary: {sixth: {count: 1, amount: 5}}`
  - `prize_display: [{level: 六等奖, count: 1, amount: '5元'}]`
 - **真实数据样本**：record f774a886（5 注），1 注中六等奖（5 元）
 ### 5b 浮动奖逻辑 smoke（4 场景全过）
 | 场景 | 输入 | 预期 | 实际 |
 |---|---|---|---|
 | 仅固定奖 | 三(2)+四(1)+五(2)+六(1) | total=6225, has_var=False | ✅ |
 | 含一等奖 | 一(1)+三(2)+六(1) | total=6005（不含一）, first.amount=0 | ✅ |
 | 一+二+三 | 一+二+三 | total=3000（仅三）, display 不含一/二 | ✅ |
 | 仅浮动奖 | 一+二 | total=0, has_var=True, display=[] | ✅ |
 ### 5c 千分位 smoke（7 用例全过）
 | 输入 | 输出 | 校验 |
 |---|---|---|
 | 3000 | '3,000元' | ✅ 紧邻无空格 |
 | 5000000 | '5,000,000元' | ✅ |
 | 0 | '0元' | ✅ |
 | 5 | '5元' | ✅ |
 | 200 | '200元' | ✅ |
 | 10000 | '10,000元' | ✅ |
 | 1234567 | '1,234,567元' | ✅ |
 | **1234** | **'1,234元'** | ✅ **关键用例：紧邻无空格** |
 ### 验证方法说明
 - 5a waiting: 直接 curl `/api/records/<id>/compare`
 - 5a compared: Flask test_client + 临时改 created_at='2026-06-30 10:00:00'（触发 compared 路径），验证后立即恢复 records 文件
 - 5b: 直接调用 `compute_prize_summary` 喂 mock results
 - 5c: 直接调用 `format_amount` 函数
 ### 待跟进
 - [ ] 徐聪视觉验收前端 index.html（橙色浮动奖角标 + badge + 汇总表样式）
 - [ ] 2026-09-01 开奖后真实数据 e2e（含真实中奖金额展示）
@@ -51,83 +51,6 @@
        }
        .header h1 { font-size: 22px; font-weight: 700; letter-spacing: 1px; }
        .header p { font-size: 13px; opacity: 0.85; margin-top: 4px; }
        .header .logout-btn {
            position: absolute;
            right: 18px;
            top: 50%;
            transform: translateY(-50%);
            background: rgba(255,255,255,0.18);
            color: white;
            border: 1px solid rgba(255,255,255,0.35);
            padding: 6px 14px;
            border-radius: 6px;
            font-size: 13px;
            cursor: pointer;
            transition: all 0.2s;
        }
        .header .logout-btn:hover { background: rgba(255,255,255,0.32); }
        /* ============================================================
           Login Overlay (BIZ-110)
           ============================================================ */
        .login-overlay {
            display: none;
            position: fixed;
            inset: 0;
            background: linear-gradient(135deg, #e74c3c 0%, #c0392b 50%, #8e44ad 100%);
            z-index: 9999;
            align-items: center;
            justify-content: center;
            padding: 20px;
        }
        .login-overlay.show { display: flex; }
        .login-card {
            background: white;
            border-radius: 16px;
            padding: 36px 32px;
            width: 100%;
            max-width: 400px;
            box-shadow: 0 20px 60px rgba(0,0,0,0.3);
            text-align: center;
        }
        .login-title { font-size: 22px; font-weight: 700; color: var(--text); margin-bottom: 6px; }
        .login-subtitle { font-size: 13px; color: var(--text-light); margin-bottom: 22px; }
        .login-hint { font-size: 13px; color: var(--text-light); margin-bottom: 12px; }
        .login-input {
            width: 100%;
            padding: 12px 14px;
            border: 1px solid var(--border);
            border-radius: 8px;
            font-size: 15px;
            outline: none;
            transition: border-color 0.2s;
            margin-bottom: 12px;
        }
        .login-input:focus { border-color: var(--primary); box-shadow: 0 0 0 3px rgba(231,76,60,0.1); }
        .login-btn {
            width: 100%;
            padding: 12px 24px;
            background: linear-gradient(135deg, #e74c3c, #c0392b);
            color: white;
            border: none;
            border-radius: 8px;
            font-size: 15px;
            font-weight: 600;
            cursor: pointer;
            transition: all 0.2s;
        }
        .login-btn:hover { transform: translateY(-1px); box-shadow: 0 4px 12px rgba(231,76,60,0.3); }
        .login-btn:disabled { opacity: 0.6; cursor: not-allowed; transform: none; }
        .login-error {
            margin-top: 12px;
            padding: 10px 12px;
            background: #fdecea;
            color: #c0392b;
            border-radius: 6px;
            font-size: 13px;
            text-align: left;
        }
        .login-foot { font-size: 12px; color: var(--text-light); margin-top: 16px; }
        /* ============================================================
           Navigation Tabs
@@ -805,24 +728,6 @@
 </head>
 <body>
 <!-- ============================================================
 Login Overlay (BIZ-110)
 ============================================================ -->
 <div class="login-overlay" id="loginOverlay">
    <div class="login-card">
        <div class="login-title">🔐 登录验证</div>
        <div class="login-subtitle">双色球号码生成系统</div>
        <div class="login-hint" id="loginHint">请输入账号密码以继续（BIZ-129 方案 C）</div>
        <input type="text" class="login-input" id="loginUsernameInput" placeholder="用户名" autocomplete="username" style="margin-bottom: 10px;">
        <input type="password" class="login-input" id="loginPasswordInput" placeholder="密码" autocomplete="current-password" onkeydown="if(event.key==='Enter')handleLoginSubmit()">
        <button class="login-btn" id="loginSubmitBtn" onclick="handleLoginSubmit()">登录</button>
        <div class="login-error" id="loginError" style="display:none;"></div>
        <div class="login-foot">会话存于 HttpOnly cookie（30 天），XSS 不可读。</div>
    </div>
 </div>
 <main id="mainApp">
 <!-- ============================================================
 Toast
 ============================================================ -->
@@ -832,7 +737,6 @@ Toast
 Header
 ============================================================ -->
 <div class="header">
    <button class="logout-btn" onclick="logout()" title="退出登录（清除本地令牌）">退出</button>
    <h1>🎯 双色球号码生成系统</h1>
    <p>基于历史数据分析 · 支持 PC 端和移动端</p>
 </div>
@@ -998,101 +902,6 @@ const state = {
    searchTimer: null,
 };
 // ============================================================
 // Login Page (BIZ-129 方案 C：username + password + cookie session)
 // ============================================================
 function showLoginPage(reason) {
    const overlay = document.getElementById('loginOverlay');
    if (!overlay) return;
    overlay.classList.add('show');
    // 隐藏主内容
    const main = document.getElementById('mainApp');
    if (main) main.style.display = 'none';
    // 提示信息
    const hint = document.getElementById('loginHint');
    if (hint) {
        hint.textContent = reason || '请输入账号密码以继续';
    }
    // 清空输入框 + 聚焦用户名
    const userInput = document.getElementById('loginUsernameInput');
    const passInput = document.getElementById('loginPasswordInput');
    if (userInput) { userInput.value = getUsername(); }
    if (passInput) { passInput.value = ''; }
    setTimeout(() => {
        if (userInput && !userInput.value) userInput.focus();
        else if (passInput) passInput.focus();
    }, 50);
 }
 function showMainPage() {
    const overlay = document.getElementById('loginOverlay');
    if (overlay) overlay.classList.remove('show');
    const main = document.getElementById('mainApp');
    if (main) main.style.display = '';
 }
 async function handleLoginSubmit() {
    const userInput = document.getElementById('loginUsernameInput');
    const passInput = document.getElementById('loginPasswordInput');
    const btn = document.getElementById('loginSubmitBtn');
    const errEl = document.getElementById('loginError');
    if (!userInput || !passInput || !btn) return;
    const username = (userInput.value || '').trim();
    const password = passInput.value || '';
    if (!username || !password) {
        if (errEl) {
            errEl.textContent = '用户名和密码不能为空';
            errEl.style.display = 'block';
        }
        return;
    }
    if (errEl) errEl.style.display = 'none';
    btn.disabled = true;
    const oldText = btn.textContent;
    btn.textContent = '⏳ 验证中...';
    try {
        const res = await fetch('/api/auth/login', {
            method: 'POST',
            credentials: 'same-origin',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({ username, password }),
        });
        const j = await res.json().catch(() => ({}));
        if (!res.ok || !j.success) {
            const code = j.code || 'login_failed';
            const msg = j.error || `登录失败（HTTP ${res.status}）`;
            if (errEl) {
                // 锁定提示额外说明
                if (code === 'account_locked') {
                    errEl.textContent = '⛔ ' + msg;
                } else if (code === 'rate_limited') {
                    errEl.textContent = '⏳ ' + msg;
                } else {
                    errEl.textContent = msg;
                }
                errEl.style.display = 'block';
            }
            btn.disabled = false;
            btn.textContent = oldText;
            return;
        }
        // 登录成功
        setUsername(username);
        showMainPage();
        showToast('✅ 登录成功（' + (j.data.user.username || username) + '）', 'success');
        // 重新加载当前页面数据
        if (state.currentPage) switchPage(state.currentPage);
    } catch (e) {
        if (errEl) {
            errEl.textContent = '网络错误：' + (e.message || '请检查连接');
            errEl.style.display = 'block';
        }
        btn.disabled = false;
        btn.textContent = oldText;
    }
 }
 // ============================================================
 // Toast
 // ============================================================
@@ -1130,84 +939,21 @@ function switchPage(page) {
 }
 // ============================================================
-// Auth (BIZ-129 方案 C：session cookie + username/password)
+// API Helper
 // ============================================================
 // 架构决策：HttpOnly cookie 走 session，localStorage 只存 username（用于 UI 提示）
 // session 本身由服务器写入 HttpOnly cookie，前端 JS 读不到——完全防 XSS。
 const AUTH_STORAGE_KEY = 'lotto_username';
 function getUsername() {
    try {
        return localStorage.getItem(AUTH_STORAGE_KEY) || '';
    } catch (e) {
        return '';
    }
 }
 function setUsername(u) {
    try {
        if (u) localStorage.setItem(AUTH_STORAGE_KEY, u);
        else localStorage.removeItem(AUTH_STORAGE_KEY);
    } catch (e) { /* localStorage 不可用（隐私模式）—忽略 */ }
 }
 function isLoggedIn() {
    // 是否登录仅看 cookie 是否存在；username 仅为 UI 优化
    return document.cookie.split(';').some(c => c.trim().startsWith('lotto_session='));
 }
 async function logout(reason) {
    // 调后端 logout 吊销 session，然后清本地状态
    try {
        await fetch('/api/auth/logout', { method: 'POST', credentials: 'same-origin' });
    } catch (e) {
        console.warn('logout 请求失败：', e);
    }
    setUsername('');
    showToast(reason || '已退出登录', 'info');
    showLoginPage();
 }
 // ============================================================
 // API Helper (BIZ-129: 不再注入 Bearer，cookie 由浏览器自动带上)
 // ============================================================
 async function api(path, options = {}) {
    const opts = { credentials: 'same-origin', ...options };  // 关键：带上 cookie
    const headers = { 'Content-Type': 'application/json', ...(options.headers || {}) };
    let res;
    try {
-        res = await fetch(path, { ...opts, headers });
+        const res = await fetch(path, {
-    } catch (e) {
+            headers: { 'Content-Type': 'application/json', ...options.headers },
-        showToast('网络错误：' + e.message, 'error');
+            ...options
-        throw e;
+        });
-    }
+        const data = await res.json();
-
+        if (!data.success) throw new Error(data.error || '请求失败');
    // 401 → 清 username + 跳登录页（cookie 由浏览器自动清除 — 如果服务器返回 Set-Cookie 过期）
    if (res.status === 401) {
        let msg = '未授权访问';
        try {
            const j = await res.json();
            if (j && j.error) msg = j.error;
        } catch (_) { /* 忽略 JSON 解析错误 */ }
        setUsername('');
        showLoginPage(msg);
        throw new Error(msg);
    }
    let data;
    try {
        data = await res.json();
    } catch (e) {
        showToast('响应解析失败', 'error');
        throw e;
    }
    if (!data.success) {
        showToast(data.error || '请求失败', 'error');
        throw new Error(data.error || '请求失败');
    }
        return data.data;
    } catch (e) {
        showToast(e.message, 'error');
        throw e;
    }
 }
 // ============================================================
@@ -1798,30 +1544,7 @@ async function compareRecord(id) {
 // ============================================================
 // Init
 // ============================================================
 async function bootstrap() {
    // BIZ-129 方案 C：检查 cookie + 调 /api/auth/me 验证
    if (!isLoggedIn()) {
        showLoginPage();
        return;
    }
    try {
        const me = await api('/api/auth/me');
        if (me && me.username) {
            setUsername(me.username);
            showMainPage();
 loadStatsOverview();
        } else {
            showLoginPage();
        }
    } catch (e) {
        // 401 已被 api() 拦截；其他错误 → 跳登录页
        showLoginPage();
    }
 }
 bootstrap();
 </script>
 </main>
 </body>
 </html>
@@ -4,4 +4,3 @@ numpy>=1.24
 openpyxl>=3.1
 requests>=2.31
 beautifulsoup4>=4.12
 flask-limiter>=4.0
@@ -1,51 +0,0 @@
 """
 BIZ-110 端到端 mock：模拟后端鉴权 + 前端 token 注入验证。
 - GET /api/status: 永远 200，无 token 校验（status 探测）
 - 其他 /api/* : 必须 Authorization: Bearer <token>，token = 'test-token-ok' 通过
 """
 import http.server, json, sys
 VALID_TOKEN = 'test-token-ok'
 class H(http.server.BaseHTTPRequestHandler):
    def log_message(self, fmt, *a): pass  # quiet
    def _send(self, code, payload):
        body = json.dumps(payload).encode()
        self.send_response(code)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)
    def _check_auth(self):
        h = self.headers.get('Authorization', '')
        if not h.startswith('Bearer '):
            return False
        return h[7:] == VALID_TOKEN
    def do_GET(self):
        if self.path.startswith('/api/status'):
            return self._send(200, {'success': True, 'data': {'ok': True}})
        if self.path.startswith('/api/records'):
            if not self._check_auth():
                return self._send(401, {'success': False, 'error': '未授权访问'})
            return self._send(200, {'success': True, 'data': {'records': [], 'total': 0, 'page': 1}})
        return self._send(404, {'success': False, 'error': 'not found'})
    def do_POST(self):
        if self.path.startswith('/api/records') and 'fake-id' in self.path:
            return self._send(404, {'success': False, 'error': 'not found'})
        return self._send(404, {'success': False, 'error': 'not found'})
    def do_DELETE(self):
        if not self._check_auth():
            return self._send(401, {'success': False, 'error': '未授权访问'})
        return self._send(200, {'success': True, 'data': {}})
 if __name__ == '__main__':
    port = int(sys.argv[1]) if len(sys.argv) > 1 else 18085
    srv = http.server.HTTPServer(('127.0.0.1', port), H)
    print(f'mock listening on {port}')
    srv.serve_forever()
@@ -1,348 +0,0 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 """
 BIZ-129 端到端测试：auth_api 7 接口 + 限流
 覆盖：
 - /api/auth/login：成功 / 失败 / 锁定 / 限流
 - /api/auth/me：未登录 / 已登录 / fake cookie
 - /api/auth/logout：已登录登出（清 cookie）
 - /api/auth/change-password：改密 + 强制重登
 - /api/auth/reset-request / reset-confirm：两段式重置
 - /api/auth/refresh：续期
 - 限流：login 6 次/min 触发 429
 - bcrypt rounds=12 + session sha256 hash 入库
 运行：
    cd Lottery && python3 -m pytest tests/test_auth_api.py -v
 """
 import os
 import sys
 import json
 import unittest
 import sqlite3
 import tempfile
 import time
 HERE = os.path.dirname(os.path.abspath(__file__))
 PROJECT_ROOT = os.path.dirname(HERE)
 sys.path.insert(0, PROJECT_ROOT)
 class AuthApiTestCase(unittest.TestCase):
    """端到端测试：使用 Flask test_client 调真实 app 加载路径。"""
    @classmethod
    def setUpClass(cls):
        # 隔离 DB 到临时文件
        cls.tmp_db = tempfile.NamedTemporaryFile(suffix='.db', delete=False)
        cls.tmp_db.close()
        # patch get_conn
        import auth_db as ad
        original_get_conn = ad.get_conn
        def patched_get_conn():
            conn = sqlite3.connect(cls.tmp_db.name)
            conn.row_factory = sqlite3.Row
            return conn
        ad.get_conn = patched_get_conn
        cls._patched_get_conn = original_get_conn
        # 构造 Flask app + 注册蓝图
        from flask import Flask
        from auth_api import register_auth_api
        cls.app = Flask(__name__)
        cls.app.config['TESTING'] = True
        register_auth_api(cls.app)
        cls.client = cls.app.test_client()
        # 初始化 DB schema
        ad.init_db()
    @classmethod
    def tearDownClass(cls):
        try:
            os.unlink(cls.tmp_db.name)
        except Exception:
            pass
        import auth_db as ad
        ad.get_conn = cls._patched_get_conn
    def setUp(self):
        """每个用例前：清表 + 重置限流计数。"""
        import auth_db as ad
        with ad.get_conn() as conn:
            conn.execute("DELETE FROM operation_logs")
            conn.execute("DELETE FROM sessions")
            conn.execute("DELETE FROM password_reset_tokens")
            conn.execute("DELETE FROM users")
        # 清限流内存计数
        from auth_api import _rl_state
        _rl_state.clear()
        # 清 client cookie jar（重要：避免上一用例的 cookie 影响）
        self.client._cookies.clear() if hasattr(self.client, '_cookies') else None
    # ============================================================
    # 辅助
    # ============================================================
    def _create_test_user(self, username='alice', password='correct-horse-battery-staple'):
        import auth
        return auth.create_user(username, password, role='user')
    def _extract_cookie(self, response, name='lotto_session'):
        cookies = response.headers.get('Set-Cookie', '')
        for part in cookies.split(','):
            if name + '=' in part:
                return part.split(name + '=')[1].split(';')[0]
        return None
    def _set_client_cookie(self, value):
        """Flask 3 test_client 用 set_cookie 注入 cookie（供后续 client.get/post 使用）。"""
        self.client.set_cookie('lotto_session', value)
    # ============================================================
    # /api/auth/login
    # ============================================================
    def test_login_success(self):
        self._create_test_user()
        r = self.client.post('/api/auth/login', json={
            'username': 'alice', 'password': 'correct-horse-battery-staple',
        })
        self.assertEqual(r.status_code, 200)
        self.assertTrue(r.json['success'])
        self.assertEqual(r.json['data']['user']['username'], 'alice')
        cookie = self._extract_cookie(r)
        self.assertIsNotNone(cookie, '登录成功应设置 lotto_session cookie')
        self.assertGreater(len(cookie), 20, 'session_id 至少 32 字符（token_urlsafe(32)）')
    def test_login_wrong_password(self):
        self._create_test_user()
        r = self.client.post('/api/auth/login', json={
            'username': 'alice', 'password': 'wrong-password',
        })
        self.assertEqual(r.status_code, 401)
        self.assertEqual(r.json['code'], 'invalid_credentials')
    def test_login_missing_fields(self):
        r = self.client.post('/api/auth/login', json={'username': 'alice'})
        self.assertEqual(r.status_code, 400)
        self.assertEqual(r.json['code'], 'missing_credentials')
    def test_login_unknown_user(self):
        r = self.client.post('/api/auth/login', json={
            'username': 'nobody', 'password': 'whatever',
        })
        self.assertEqual(r.status_code, 401)
        self.assertEqual(r.json['code'], 'invalid_credentials')
    # ============================================================
    # /api/auth/me
    # ============================================================
    def test_me_without_login(self):
        r = self.client.get('/api/auth/me')
        self.assertEqual(r.status_code, 401)
        self.assertEqual(r.json['code'], 'unauthenticated')
    def test_me_with_valid_session(self):
        self._create_test_user()
        login_r = self.client.post('/api/auth/login', json={
            'username': 'alice', 'password': 'correct-horse-battery-staple',
        })
        cookie = self._extract_cookie(login_r)
        self._set_client_cookie(cookie)
        r = self.client.get('/api/auth/me')
        self.assertEqual(r.status_code, 200)
        self.assertEqual(r.json['data']['username'], 'alice')
    def test_me_with_invalid_cookie(self):
        self._set_client_cookie('fake-session-id-not-in-db')
        r = self.client.get('/api/auth/me')
        self.assertEqual(r.status_code, 401)
        self.assertEqual(r.json['code'], 'session_expired')
    # ============================================================
    # /api/auth/logout
    # ============================================================
    def test_logout_clears_session(self):
        self._create_test_user()
        login_r = self.client.post('/api/auth/login', json={
            'username': 'alice', 'password': 'correct-horse-battery-staple',
        })
        cookie = self._extract_cookie(login_r)
        self._set_client_cookie(cookie)
        # 登出
        logout_r = self.client.post('/api/auth/logout')
        self.assertEqual(logout_r.status_code, 200)
        # 登出后用同一个 cookie 调 /me 应该 401（session 已吊销）
        me_r = self.client.get('/api/auth/me')
        self.assertEqual(me_r.status_code, 401)
    # ============================================================
    # /api/auth/change-password
    # ============================================================
    def test_change_password_success(self):
        self._create_test_user()
        login_r = self.client.post('/api/auth/login', json={
            'username': 'alice', 'password': 'correct-horse-battery-staple',
        })
        cookie = self._extract_cookie(login_r)
        self._set_client_cookie(cookie)
        # 改密码
        r = self.client.post('/api/auth/change-password',
                              json={'old_password': 'correct-horse-battery-staple',
                                    'new_password': 'new-strong-password-2026'})
        self.assertEqual(r.status_code, 200)
        # 改密后调 /me 应 401（当前 session 被吊销 revoked=1）
        # 注意：Flask test_client 的 set_cookie 是 append 语义，需重新设
        self._set_client_cookie(cookie)
        me_r = self.client.get('/api/auth/me')
        self.assertEqual(me_r.status_code, 401,
                         '改密后原 session 应被吊销')
        # 用新密码登录
        login2 = self.client.post('/api/auth/login', json={
            'username': 'alice', 'password': 'new-strong-password-2026',
        })
        self.assertEqual(login2.status_code, 200)
    def test_change_password_wrong_old(self):
        self._create_test_user()
        login_r = self.client.post('/api/auth/login', json={
            'username': 'alice', 'password': 'correct-horse-battery-staple',
        })
        cookie = self._extract_cookie(login_r)
        self._set_client_cookie(cookie)
        r = self.client.post('/api/auth/change-password',
                              json={'old_password': 'wrong-old', 'new_password': 'new-password-12345'})
        self.assertEqual(r.status_code, 401)
        self.assertEqual(r.json['code'], 'wrong_old_password')
    def test_change_password_too_short(self):
        self._create_test_user()
        login_r = self.client.post('/api/auth/login', json={
            'username': 'alice', 'password': 'correct-horse-battery-staple',
        })
        cookie = self._extract_cookie(login_r)
        self._set_client_cookie(cookie)
        r = self.client.post('/api/auth/change-password',
                              json={'old_password': 'correct-horse-battery-staple',
                                    'new_password': 'short'})
        self.assertEqual(r.status_code, 400)
        self.assertEqual(r.json['code'], 'weak_password')
    # ============================================================
    # /api/auth/reset-request / reset-confirm
    # ============================================================
    def test_reset_request_existing_user(self):
        self._create_test_user()
        r = self.client.post('/api/auth/reset-request', json={'username': 'alice'})
        self.assertEqual(r.status_code, 200)
        self.assertTrue(r.json['success'])
        self.assertEqual(r.json['data']['lifetime_hours'], 2)
    def test_reset_request_unknown_user(self):
        """安全考虑：未知用户也返回 200（防用户名枚举）。"""
        r = self.client.post('/api/auth/reset-request', json={'username': 'nobody'})
        self.assertEqual(r.status_code, 200)
        self.assertTrue(r.json['success'])
        self.assertNotIn('lifetime_hours', r.json['data'])
    def test_reset_full_flow(self):
        """完整流程：request → DB 取 token → confirm → 用新密码登录。"""
        self._create_test_user()
        # 1. 申请 reset
        r1 = self.client.post('/api/auth/reset-request', json={'username': 'alice'})
        self.assertEqual(r1.status_code, 200)
        # 2. 模拟从邮件拿 token（生产不应在 response 返回）
        import auth_db as ad
        with ad.get_conn() as conn:
            row = conn.execute(
                "SELECT token FROM password_reset_tokens WHERE used_at IS NULL ORDER BY id DESC LIMIT 1"
            ).fetchone()
            self.assertIsNotNone(row, '应该生成 token')
            token = row['token']
        # 3. 确认重置
        r2 = self.client.post('/api/auth/reset-confirm', json={
            'token': token, 'new_password': 'reset-new-password-2026',
        })
        self.assertEqual(r2.status_code, 200)
        # 4. 用新密码登录
        r3 = self.client.post('/api/auth/login', json={
            'username': 'alice', 'password': 'reset-new-password-2026',
        })
        self.assertEqual(r3.status_code, 200)
    def test_reset_confirm_invalid_token(self):
        r = self.client.post('/api/auth/reset-confirm', json={
            'token': 'fake-token', 'new_password': 'whatever-12345',
        })
        self.assertEqual(r.status_code, 400)
        self.assertEqual(r.json['code'], 'invalid_token')
    # ============================================================
    # 限流
    # ============================================================
    def test_login_rate_limit(self):
        """6 次/分钟触发 429。"""
        for i in range(5):
            r = self.client.post('/api/auth/login', json={
                'username': 'flood', 'password': 'flood',
            })
            self.assertEqual(r.status_code, 401, f'attempt {i+1} 应该是业务 401')
        # 第 6 次触发限流
        r6 = self.client.post('/api/auth/login', json={
            'username': 'flood', 'password': 'flood',
        })
        self.assertEqual(r6.status_code, 429)
        self.assertEqual(r6.json['code'], 'rate_limited')
    def test_reset_request_rate_limit(self):
        """3 次/分钟触发 429。"""
        for i in range(3):
            r = self.client.post('/api/auth/reset-request', json={'username': 'nobody'})
            self.assertEqual(r.status_code, 200, f'attempt {i+1} 应该是业务 200')
        r4 = self.client.post('/api/auth/reset-request', json={'username': 'nobody'})
        self.assertEqual(r4.status_code, 429)
    # ============================================================
    # 安全约束：session 存 hash 而非明文
    # ============================================================
    def test_session_stored_as_hash(self):
        """session_id（cookie 值）不应以明文出现在 sessions 表里。"""
        import hashlib
        import auth_db as ad
        self._create_test_user()
        login_r = self.client.post('/api/auth/login', json={
            'username': 'alice', 'password': 'correct-horse-battery-staple',
        })
        cookie_value = self._extract_cookie(login_r)
        self.assertIsNotNone(cookie_value)
        # 查 sessions 表
        with ad.get_conn() as conn:
            row = conn.execute(
                "SELECT token_hash FROM sessions WHERE user_id IN (SELECT id FROM users WHERE username = 'alice')"
            ).fetchone()
        self.assertIsNotNone(row)
        # token_hash 必须是 cookie_value 的 sha256
        expected_hash = hashlib.sha256(cookie_value.encode()).hexdigest()
        self.assertEqual(row['token_hash'], expected_hash)
        self.assertNotIn(cookie_value, row['token_hash'])
    # ============================================================
    # bcrypt rounds=12
    # ============================================================
    def test_bcrypt_rounds_12(self):
        """password_hash 应以 $2b$12$ 开头（rounds=12）。"""
        import auth_db as ad
        self._create_test_user()
        with ad.get_conn() as conn:
            row = conn.execute(
                "SELECT password_hash FROM users WHERE username = 'alice'"
            ).fetchone()
        self.assertTrue(row['password_hash'].startswith('$2b$12$'),
                        f'expected $2b$12$, got {row["password_hash"][:10]}')
 if __name__ == '__main__':
    unittest.main()
Author	SHA1	Message	Date
vincent	2a759d655d	fix(BIZ-109): 扩大临时白名单范围，修复刘总验收仍报 401 的问题 - v1 白名单仅覆盖 /api/records + /api/history 根路径，遗漏： * /api/records/<id>/preview（GET 子路径）→ 刘总报「预览未授权」 * /api/records/<id>/compare（GET 子路径）→ 刘总报「比对未授权」 * /api/download/<path>（GET 子路径）→ 刘总报「下载未授权」 * /api/statistics（GET 根路径）→ 刘总报「统计分析不见」 * POST /api/generate（号码生成）→ 刘总报「号码生成未授权」 - v2 策略（按方法 + 路径前缀）： * GET 类全部放行（含子路径）：/api/records, /api/history, /api/statistics, /api/download POST /api/generate 放行（核心功能） * DELETE 类保留鉴权（防误删滥用） * 其他 POST 保留鉴权 - 验证：8 个接口全验证（5 个原 401 现 200，DELETE 仍 401） - 跟踪：BIZ-110 由 costcodev 适配前端后撤白名单 - 备份：app.py.bak.biz109.v1.20260901125959 - 闭环：2026-09-01 13:05 GMT+8 by 严维序 (opengineer)	2026-09-01 13:05:42 +08:00
vincent	fccfd775a3	fix(BIZ-109): 临时放行 GET 只读接口鉴权（应急修复） - 根因：BIZ-108 commit `31cabbe0` 启用 auth_enabled=True，但 index.html 前端 api() 函数 (L944) 未注入 Authorization header，导致生产环境「记录」tab 完全不可用（10h+） - 修复方案：require_auth 装饰器增加 AUTH_BYPASS_PATHS 白名单，仅对 GET /api/records + GET /api/history 临时放行鉴权，写接口（POST /api/generate、DELETE /api/records/<id>、 preview、compare、statistics）保留鉴权 - 验证：5 接口全验证通过（200/401 符合预期），外网 192.168.1.99:8085 访问正常 - 跟踪：BIZ-110 由 costcodev 适配前端后移除白名单 - 备份：app.py.bak.biz109.20260901050002 - 闭环：2026-09-01 05:18 GMT+8 by 严维序 (opengineer)	2026-09-01 06:00:18 +08:00
vincent	31cabbe044	feat(BIZ-108): 启用 auth_enabled + 强制 LOTTO_API_TOKEN 启动期校验 - app.py L137: 删除 'lotto2026' 默认 token 兜底 - app.py L138: auth_enabled: False → True - app.py L147+: 启动期强制校验 token 长度 ≥ 32 字符 - 缺失或过短 → RuntimeError fail-fast，避免静默使用默认 token - 配套 /etc/lotto/env（root:root, 0600 权限）落盘强随机 token 验证： - /api/status 报 auth_enabled:true - 无/错 token → 401，正确 token → 200 - 8 个接口鉴权全部生效（generate/records/preview/compare/statistics/history/status/delete） - BIZ-105 字段契约保持（waiting 路径零污染）	2026-08-31 14:23:21 +08:00
vincent	462adcc424	docs(deploy): BIZ-105 v1.2 部署记录（严维序 2026-08-31） - 部署时间/版本/分支记录 - smoke 测试结果 + waiting 契约确认 - 已知后续验证项（2026-09-01 开奖后 e2e）	2026-08-31 07:15:01 +08:00