Archived
feat(routes): 迁移认证端点(login/logout/heartbeat/ann_seen)
- web/routers/auth.py
· login:邮箱/用户名登录、超长密码短路、失败锁定、封禁/禁用分支、
管理员转 2FA 票据;成功签发 core.session 会话 Cookie(与 Flask 同格式)
· logout:递增 login_seq 顶号 + 删 Cookie
· heartbeat:次数校验(401/expired 语义同 Flask)
· ann_seen:记录已读公告版本
· 失败锁定与 2FA 票据表复用 auth_routes 模块级状态(跨栈共享,防绕锁定)
· PBKDF2 与库操作走 asyncio.to_thread,不阻塞事件循环
测试:tests/test_auth_routes.py 5 项,全部使用**沙箱库**(pytest tmp_path +
monkeypatch auth_system/core.db 路径),不触碰生产 data/site.db。
全套:pytest 56/56;ruff All checks passed;pyright 0 errors;奇偶校验 28/28。
This commit is contained in:
1 parent
18c77296c2
commit
91c032a1c3
3 files changed
+339
No files matched your search
@@ -0,0 +1,112 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""登录/会话端点测试(沙箱库,不触碰生产 data/site.db)。
|
||||
|
||||
覆盖:密码错误、正常用户登录签发可用 Cookie、管理员转 2FA 票据、
|
||||
退出登录递增 login_seq、heartbeat 次数校验。
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
import auth_system as auth
|
||||
import core.db as db
|
||||
|
||||
SECRET = 'auth-test-secret'
|
||||
USER_PW = 'UserPass123'
|
||||
ADMIN_PW = 'AdminPass123'
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def app_env(tmp_path, monkeypatch):
|
||||
"""沙箱站点库 + 独立密钥,隔离生产数据。"""
|
||||
db_path = tmp_path / 'site.db'
|
||||
monkeypatch.setattr(auth, 'DATA_DIR', tmp_path)
|
||||
monkeypatch.setattr(auth, 'DB_PATH', db_path)
|
||||
monkeypatch.setattr(auth, 'BACKUP_DIR', tmp_path / 'backups')
|
||||
monkeypatch.setattr(auth, 'SECRET_KEY_PATH', tmp_path / 'secret.key')
|
||||
monkeypatch.setattr(db, 'DATA_DIR', tmp_path)
|
||||
monkeypatch.setattr(db, 'DB_PATH', db_path)
|
||||
auth.init_db()
|
||||
|
||||
conn = auth.get_conn()
|
||||
conn.execute('INSERT INTO users (username, password_hash, salt, role, duration_seconds, '
|
||||
'status, login_seq, created_at) VALUES (?,?,?,?,?,1,0,?)',
|
||||
('normaluser', auth._hash_new(USER_PW), '', 'user', 97, 1730000000))
|
||||
conn.execute('INSERT INTO users (username, password_hash, salt, role, duration_seconds, '
|
||||
'status, login_seq, created_at, email) VALUES (?,?,?,?,?,1,0,?,?)',
|
||||
('bossuser', auth._hash_new(ADMIN_PW), '', 'admin', -1, 1730000000,
|
||||
'boss@example.com'))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
monkeypatch.setattr('web.deps.secret_key', lambda: SECRET)
|
||||
monkeypatch.setattr('web.routers.auth.secret_key', lambda: SECRET)
|
||||
|
||||
# 清空登录失败锁定/2FA 的模块级内存态,避免测试间串扰
|
||||
import auth_routes
|
||||
auth_routes._LOGIN_FAILS.clear()
|
||||
auth_routes._LOGIN_FAIL_LOGGED.clear()
|
||||
auth_routes._ADMIN_2FA.clear()
|
||||
|
||||
from web.app import create_app
|
||||
return TestClient(create_app())
|
||||
|
||||
|
||||
def test_wrong_password(app_env):
|
||||
r = app_env.post('/api/v1/auth/login', json={'username': 'normaluser', 'password': 'nope'})
|
||||
assert r.status_code == 400
|
||||
assert r.json()['success'] is False
|
||||
assert 'session' not in r.cookies
|
||||
|
||||
|
||||
def test_user_login_issues_working_cookie(app_env):
|
||||
r = app_env.post('/api/v1/auth/login', json={'username': 'normaluser', 'password': USER_PW})
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body['success'] is True
|
||||
assert body['data']['redirect'] == 'user'
|
||||
assert 'session' in r.cookies
|
||||
|
||||
# 用签发的 Cookie 访问受保护端点,应放行
|
||||
r2 = app_env.post('/api/v1/auth/heartbeat', json={})
|
||||
assert r2.status_code == 200
|
||||
assert r2.json()['data']['duration_text'] == '97 次'
|
||||
|
||||
|
||||
def test_admin_login_returns_2fa_ticket(app_env):
|
||||
r = app_env.post('/api/v1/auth/login', json={'username': 'bossuser', 'password': ADMIN_PW})
|
||||
assert r.status_code == 200
|
||||
data = r.json()['data']
|
||||
assert data['admin_2fa'] is True
|
||||
assert data['ticket']
|
||||
assert 'session' not in r.cookies # 管理员不在此建立会话
|
||||
|
||||
|
||||
def test_logout_bumps_login_seq_and_clears(app_env):
|
||||
app_env.post('/api/v1/auth/login', json={'username': 'normaluser', 'password': USER_PW})
|
||||
conn = auth.get_conn()
|
||||
seq_before = conn.execute(
|
||||
"SELECT login_seq FROM users WHERE username='normaluser'").fetchone()[0]
|
||||
conn.close()
|
||||
r = app_env.post('/api/v1/auth/logout', json={})
|
||||
assert r.status_code == 200
|
||||
conn = auth.get_conn()
|
||||
seq_after = conn.execute(
|
||||
"SELECT login_seq FROM users WHERE username='normaluser'").fetchone()[0]
|
||||
conn.close()
|
||||
assert seq_after == seq_before + 1
|
||||
# 旧 Cookie 失效
|
||||
r2 = app_env.post('/api/v1/auth/heartbeat', json={})
|
||||
assert r2.status_code == 401
|
||||
|
||||
|
||||
def test_heartbeat_expired_credits(app_env):
|
||||
conn = auth.get_conn()
|
||||
conn.execute("UPDATE users SET duration_seconds=0 WHERE username='normaluser'")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
r = app_env.post('/api/v1/auth/login', json={'username': 'normaluser', 'password': USER_PW})
|
||||
assert r.status_code == 200
|
||||
r2 = app_env.post('/api/v1/auth/heartbeat', json={})
|
||||
assert r2.status_code == 401
|
||||
assert r2.json().get('error_code') == 'expired'
|
||||
Reference in new issue
Block a user