#!/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()