costcodev
111b56f2ee
测试覆盖： - login: 成功 / 错密码 / 缺失字段 / 未知用户 (4) - me: 未登录 / 已登录 / fake cookie → session_expired (3) - logout: 清 session (1) - change-password: 成功（强制重新登录）/ 旧密码错 / 密码过短 (3) - reset-request: 已存在用户 / 未知用户（防枚举）/ 完整重置流程 / 无效 token (4) - 限流: login 6/min 触发 429 + reset-request 4/min 触发 429 (2) - 安全约束: session 存 sha256 hash + bcrypt $2b$12$ (2) 修复 bug： - reset-confirm 把 consume_reset_token 返回的 dict 当 user_id 传给 update_password → 修复为 user_info['user_id'] 测试结果：19 passed, 65 warnings in 18.34s 依赖：b002e92（7 路由）+ 44f46a0（限流）+ 1df6e7f（前端） Co-authored-by: multica-agent <github@multica.ai>
349 lines
14 KiB
Python
349 lines
14 KiB
Python
#!/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()
|