#!/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') # ============================================================ # 工具函数 # ============================================================ 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']) 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']) 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']) 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}) # ============================================================ # 蓝图注册入口（供 app.py 调用） # ============================================================ def register_auth_api(app): """在 app.py 里 app.register_blueprint(auth_bp) 即可。""" app.register_blueprint(auth_bp)