Lottery/auth_api.py

#!/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_id = consume_reset_token(token)
    if not user_id:
        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)

    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=None,
        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)
# 避免重复定义（早期版本残留已清理）