Simon
814 lines
32 KiB
Python
814 lines
32 KiB
Python
# =============================================================================
|
||
# 企微IT智能服务台 — OTP 首次绑定与管理后台清除功能 全链路测试
|
||
# =============================================================================
|
||
# 验证增量 PRD (05-增量PRD-OTP首次绑定与重置.md) 的所有行为变更：
|
||
#
|
||
# Part A: 登录行为变更
|
||
# A1. mfa_enabled=False → require_otp_bind: true（不再直发 token）
|
||
# A2. mfa_enabled=True 无 OTP → require_otp: true（回归验证）
|
||
# A3. mfa_enabled=True 有 OTP → 签发 token（回归验证）
|
||
#
|
||
# Part B: OTP 首次绑定验证 (verify_otp)
|
||
# B1. 首次绑定场景：校验通过 → verified=true + token + is_first_bind
|
||
# B2. 首次绑定场景：校验失败 → verified=false
|
||
# B3. 已绑定场景：校验通过 → verified=true（无 token，回归）
|
||
# B4. 无 secret 场景 → verified=false
|
||
#
|
||
# Part C: 管理后台端点
|
||
# C1. GET /auth/otp-admin-users → 返回列表含 mfa_enabled/mfa_bound_at
|
||
# C2. 非 admin 访问 → 403
|
||
# C3. POST /auth/otp-admin-reset/{id} → 成功清除绑定
|
||
#
|
||
# Part D: 身份验证缺口（Auth Gap）探查
|
||
# D1. require_otp_bind 后能否直接调用 otp-bind（需 token）
|
||
# =============================================================================
|
||
|
||
import json
|
||
import pyotp
|
||
import pytest
|
||
import pytest_asyncio
|
||
from datetime import datetime
|
||
from unittest.mock import AsyncMock, patch
|
||
from sqlalchemy import select
|
||
|
||
from app.models.agent import Agent
|
||
from app.models.role import Role
|
||
from app.models.user_role import UserRole
|
||
from app.services.mfa_service import MFA_VERIFIED_TTL_SECONDS, MFAService
|
||
from app.services.token_service import TokenService
|
||
from tests.conftest import create_test_agent, MockRedis
|
||
|
||
|
||
# =============================================================================
|
||
# OTP 专用 client fixture — 扩展 base client, 同时覆盖 otp.py 的 _get_redis
|
||
# =============================================================================
|
||
# 原因: otp.py 的端点使用 Depends(_get_redis) 注入 Redis, 但 client fixture 仅
|
||
# 覆盖了 app.api.agents._get_redis 和 dep_redis, 未覆盖 app.api.otp._get_redis。
|
||
# 本 fixture 在 base client 之上额外设置 app.dependency_overrides,
|
||
# 确保 otp 端点也使用 mock_redis。
|
||
# =============================================================================
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def otp_client(client, mock_redis: MockRedis):
|
||
"""返回已覆盖 otp._get_redis 依赖的 HTTP 测试客户端。
|
||
|
||
用法: 所有需要 otp 端点读写 Redis 的测试使用 otp_client 代替 client。
|
||
"""
|
||
from app.main import create_app
|
||
from app.database import get_db
|
||
from app.api.otp import _get_redis as otp_get_redis
|
||
|
||
# 从 client fixture 获取 app 实例（通过 client 的 transport）
|
||
app = client._transport.app
|
||
|
||
# 覆盖 otp._get_redis
|
||
app.dependency_overrides[otp_get_redis] = lambda: mock_redis
|
||
|
||
yield client
|
||
|
||
# 清理
|
||
if otp_get_redis in app.dependency_overrides:
|
||
del app.dependency_overrides[otp_get_redis]
|
||
|
||
|
||
# =============================================================================
|
||
# 辅助函数
|
||
# =============================================================================
|
||
|
||
def _bearer(token: str) -> dict:
|
||
"""构造 Bearer 认证头。"""
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
async def _create_token_in_redis(
|
||
mock_redis: MockRedis,
|
||
employee_id: str,
|
||
name: str = "",
|
||
roles: list = None,
|
||
login_source: str = "agent",
|
||
) -> str:
|
||
"""直接在 mock_redis 中创建 token 记录，绕过登录流程。
|
||
|
||
用于测试"半认证"状态下的端点（如首次绑定场景中 otp-bind/otp-verify 需要 token，
|
||
但 agent_login 对 mfa_enabled=False 不签发 token）。
|
||
|
||
Args:
|
||
mock_redis: 模拟 Redis
|
||
employee_id: 用户标识
|
||
name: 用户名
|
||
roles: 角色列表
|
||
login_source: 登录来源
|
||
|
||
Returns:
|
||
str: 生成的 token 字符串
|
||
"""
|
||
import secrets
|
||
token = secrets.token_urlsafe(32)
|
||
token_data = json.dumps({
|
||
"employee_id": employee_id,
|
||
"name": name,
|
||
"roles": roles or ["agent"],
|
||
"current_role": "agent",
|
||
"login_source": login_source,
|
||
})
|
||
await mock_redis.setex(f"user:token:{token}", 28800, token_data)
|
||
return token
|
||
|
||
|
||
async def _seed_admin_role(db_session, employee_id: str):
|
||
"""为用户分配 admin 角色。"""
|
||
import uuid
|
||
|
||
stmt = select(Role).where(Role.name == "admin")
|
||
role = (await db_session.execute(stmt)).scalars().first()
|
||
if not role:
|
||
role = Role(
|
||
id=str(uuid.uuid4()),
|
||
name="admin",
|
||
display_name="管理员",
|
||
is_default=False,
|
||
permissions=[],
|
||
)
|
||
db_session.add(role)
|
||
await db_session.flush()
|
||
|
||
stmt = select(UserRole).where(
|
||
UserRole.employee_id == employee_id,
|
||
UserRole.role_id == role.id,
|
||
)
|
||
existing = (await db_session.execute(stmt)).scalars().first()
|
||
if not existing:
|
||
user_role = UserRole(
|
||
id=str(uuid.uuid4()),
|
||
employee_id=employee_id,
|
||
role_id=role.id,
|
||
source="manual",
|
||
assigned_at=datetime.now(),
|
||
)
|
||
db_session.add(user_role)
|
||
await db_session.flush()
|
||
|
||
|
||
async def _login_and_get_token(client, user_id: str, name: str, otp_code: str = None):
|
||
"""调用 /agents/login 并返回响应 data。
|
||
|
||
注意：此函数在新行为下对 mfa_enabled=False 的 agent 会返回
|
||
{require_otp_bind: true} 而非 token。
|
||
"""
|
||
payload = {"user_id": user_id, "name": name}
|
||
if otp_code:
|
||
payload["otp_code"] = otp_code
|
||
|
||
response = await client.post("/agents/login", json=payload)
|
||
assert response.status_code == 200, f"登录失败: {response.text}"
|
||
body = response.json()
|
||
assert body.get("code") == 0, f"登录业务码非 0: {body}"
|
||
return body["data"]
|
||
|
||
|
||
# =============================================================================
|
||
# Part A: 登录行为变更
|
||
# =============================================================================
|
||
|
||
class TestLoginRequireOtpBind:
|
||
"""A1: mfa_enabled=False → require_otp_bind: true"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_new_agent_login_returns_require_otp_bind(self, client, db_session):
|
||
"""新坐席（mfa_enabled=False）登录应返回 require_otp_bind + 半认证 token。"""
|
||
agent = create_test_agent(user_id="new_bind_001", name="新坐席001")
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
data = await _login_and_get_token(client, "new_bind_001", "新坐席001")
|
||
|
||
assert data.get("require_otp_bind") is True, \
|
||
f"期望 require_otp_bind=true，实际: {data}"
|
||
assert "token" in data, \
|
||
f"BUG-001 修复: require_otp_bind 应携带半认证 token，实际: {data}"
|
||
assert data.get("user_id") == "new_bind_001"
|
||
assert data.get("name") == "新坐席001"
|
||
assert "role" in data
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_new_agent_login_message(self, client, db_session):
|
||
"""验证 require_otp_bind 响应包含引导消息。"""
|
||
agent = create_test_agent(user_id="new_bind_002", name="新坐席002")
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
data = await _login_and_get_token(client, "new_bind_002", "新坐席002")
|
||
|
||
assert data.get("require_otp_bind") is True
|
||
assert "message" in data
|
||
assert "绑定" in data.get("message", "")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_brand_new_agent_auto_register_and_require_bind(self, client, db_session):
|
||
"""全新坐席（DB 无记录）首次登录自动注册后也返回 require_otp_bind + token。"""
|
||
# 不预先创建 agent，login 会自动注册
|
||
data = await _login_and_get_token(client, "brand_new_001", "全新坐席")
|
||
|
||
assert data.get("require_otp_bind") is True, \
|
||
f"全新坐席首次登录应引导 OTP 绑定，实际: {data}"
|
||
assert "token" in data, \
|
||
f"BUG-001 修复: 应携带半认证 token 供后续 otp-bind 调用"
|
||
|
||
|
||
class TestLoginRequireOtpRegression:
|
||
"""A2: mfa_enabled=True 无 OTP → require_otp: true（回归）"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bound_agent_login_without_otp_returns_require_otp(self, client, db_session):
|
||
"""已绑定 agent 登录不传 otp_code → require_otp: true（回归）。"""
|
||
secret = pyotp.random_base32()
|
||
agent = create_test_agent(user_id="bound_001", name="已绑定坐席")
|
||
agent.mfa_secret = secret
|
||
agent.mfa_enabled = True
|
||
agent.mfa_bound_at = datetime.now()
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
data = await _login_and_get_token(client, "bound_001", "已绑定坐席")
|
||
|
||
assert data.get("require_otp") is True, \
|
||
f"已绑定坐席无 OTP 应返回 require_otp=true，实际: {data}"
|
||
assert "token" not in data, \
|
||
f"require_otp 不应签发 token，实际: {data}"
|
||
assert data.get("message") == "请输入OTP动态码"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bound_agent_login_with_correct_otp_returns_token(self, client, db_session):
|
||
"""已绑定 agent 传正确 OTP → 签发 token（A3 回归）。"""
|
||
secret = pyotp.random_base32()
|
||
agent = create_test_agent(user_id="bound_002", name="已绑定坐席2")
|
||
agent.mfa_secret = secret
|
||
agent.mfa_enabled = True
|
||
agent.mfa_bound_at = datetime.now()
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
otp_code = pyotp.TOTP(secret).now()
|
||
data = await _login_and_get_token(client, "bound_002", "已绑定坐席2", otp_code)
|
||
|
||
assert "token" in data, \
|
||
f"正确 OTP 应签发 token，实际: {data}"
|
||
assert data.get("user_id") == "bound_002"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bound_agent_login_with_wrong_otp_returns_error(self, client, db_session):
|
||
"""已绑定 agent 传错误 OTP → 报错。"""
|
||
secret = pyotp.random_base32()
|
||
agent = create_test_agent(user_id="bound_003", name="已绑定坐席3")
|
||
agent.mfa_secret = secret
|
||
agent.mfa_enabled = True
|
||
agent.mfa_bound_at = datetime.now()
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
response = await client.post("/agents/login", json={
|
||
"user_id": "bound_003",
|
||
"name": "已绑定坐席3",
|
||
"otp_code": "000000",
|
||
})
|
||
|
||
body = response.json()
|
||
# 错误码 1006: OTP验证码错误
|
||
assert body.get("code") == 1006, \
|
||
f"错误 OTP 应返回 1006，实际: {body}"
|
||
|
||
|
||
# =============================================================================
|
||
# Part B: OTP 首次绑定验证 (verify_otp)
|
||
# =============================================================================
|
||
|
||
class TestVerifyOtpFirstBind:
|
||
"""B1-B2: 首次绑定场景 verify_otp 行为"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_first_bind_verify_correct_code_returns_token(self, otp_client, db_session, mock_redis):
|
||
"""首次绑定 + 正确 OTP → verified=true + token + is_first_bind。
|
||
|
||
模拟流程：agent 已调用 otp-bind 获取 secret（mfa_enabled=False, mfa_secret 有值），
|
||
此时输入正确 OTP 验证码完成绑定。
|
||
"""
|
||
secret = pyotp.random_base32()
|
||
agent = create_test_agent(user_id="first_bind_001", name="首次绑定坐席")
|
||
agent.mfa_secret = secret
|
||
agent.mfa_enabled = False # 尚未启用
|
||
agent.mfa_bound_at = None
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
# 为 agent 创建 token（绕过登录的 require_otp_bind 限制）
|
||
token = await _create_token_in_redis(mock_redis, "first_bind_001", "首次绑定坐席")
|
||
|
||
otp_code = pyotp.TOTP(secret).now()
|
||
response = await otp_client.post(
|
||
"/auth/otp-verify",
|
||
headers=_bearer(token),
|
||
json={"otp_code": otp_code},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
assert body["code"] == 0, f"业务码非 0: {body}"
|
||
data = body["data"]
|
||
|
||
# 验证核心字段
|
||
assert data["verified"] is True, f"校验应通过，实际: {data}"
|
||
assert data.get("is_first_bind") is True, \
|
||
f"首次绑定应返回 is_first_bind=true，实际: {data}"
|
||
assert "token" in data, \
|
||
f"首次绑定应签发 token，实际: {data}"
|
||
assert data.get("user_id") == "first_bind_001"
|
||
assert data.get("expires_in") == MFA_VERIFIED_TTL_SECONDS
|
||
|
||
# 验证 DB 状态已更新
|
||
stmt = select(Agent).where(Agent.user_id == "first_bind_001")
|
||
db_agent = (await db_session.execute(stmt)).scalars().first()
|
||
assert db_agent.mfa_enabled is True, \
|
||
f"DB mfa_enabled 应为 True，实际: {db_agent.mfa_enabled}"
|
||
assert db_agent.mfa_bound_at is not None, \
|
||
f"DB mfa_bound_at 应为非空，实际: {db_agent.mfa_bound_at}"
|
||
assert db_agent.mfa_last_verified_at is not None
|
||
|
||
# 验证 Redis 验证标记已写入
|
||
verified_key = f"mfa:verified:first_bind_001"
|
||
assert await mock_redis.exists(verified_key), \
|
||
"Redis 中应有 mfa:verified 标记"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_first_bind_verify_wrong_code_returns_not_verified(self, client, db_session, mock_redis):
|
||
"""首次绑定 + 错误 OTP → verified=false，不改变 DB 状态。"""
|
||
secret = pyotp.random_base32()
|
||
agent = create_test_agent(user_id="first_bind_002", name="首次绑定坐席2")
|
||
agent.mfa_secret = secret
|
||
agent.mfa_enabled = False
|
||
agent.mfa_bound_at = None
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
token = await _create_token_in_redis(mock_redis, "first_bind_002", "首次绑定坐席2")
|
||
|
||
response = await client.post(
|
||
"/auth/otp-verify",
|
||
headers=_bearer(token),
|
||
json={"otp_code": "000000"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
assert body["code"] == 0
|
||
data = body["data"]
|
||
assert data["verified"] is False, f"错误码应返回 false，实际: {data}"
|
||
assert data["expires_in"] == 0
|
||
assert "token" not in data, \
|
||
f"BUG-002 修复: 验证失败 token 应被 exclude，实际: {data}"
|
||
|
||
# DB 状态不应改变
|
||
stmt = select(Agent).where(Agent.user_id == "first_bind_002")
|
||
db_agent = (await db_session.execute(stmt)).scalars().first()
|
||
assert db_agent.mfa_enabled is False, \
|
||
f"验证失败不应改变 mfa_enabled，实际: {db_agent.mfa_enabled}"
|
||
|
||
|
||
class TestVerifyOtpAlreadyBoundRegression:
|
||
"""B3-B4: 已绑定场景 verify_otp 行为（回归）"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bound_agent_verify_correct_code_no_token(self, client, db_session, mock_redis):
|
||
"""已绑定 agent 验证正确 OTP → verified=true，不含 token（回归）。"""
|
||
secret = pyotp.random_base32()
|
||
agent = create_test_agent(user_id="bound_vfy_001", name="已绑定验证坐席")
|
||
agent.mfa_secret = secret
|
||
agent.mfa_enabled = True
|
||
agent.mfa_bound_at = datetime.now()
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
token = await _create_token_in_redis(mock_redis, "bound_vfy_001", "已绑定验证坐席")
|
||
|
||
otp_code = pyotp.TOTP(secret).now()
|
||
response = await client.post(
|
||
"/auth/otp-verify",
|
||
headers=_bearer(token),
|
||
json={"otp_code": otp_code},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
assert body["code"] == 0
|
||
data = body["data"]
|
||
assert data["verified"] is True
|
||
assert data.get("is_first_bind") is not True, \
|
||
f"已绑定场景不应返回 is_first_bind，实际: {data}"
|
||
assert "token" not in data, \
|
||
f"BUG-002 修复: 已绑定场景 token 应被 exclude，实际: {data}"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_no_secret_verify_returns_false(self, client, db_session, mock_redis):
|
||
"""无 secret 的 agent 调用 verify → verified=false。"""
|
||
agent = create_test_agent(user_id="no_secret_001", name="无密钥坐席")
|
||
# mfa_secret 为 None（从未调用 otp-bind）
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
token = await _create_token_in_redis(mock_redis, "no_secret_001", "无密钥坐席")
|
||
|
||
response = await client.post(
|
||
"/auth/otp-verify",
|
||
headers=_bearer(token),
|
||
json={"otp_code": "123456"},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
assert body["code"] == 0
|
||
assert body["data"]["verified"] is False
|
||
|
||
|
||
class TestVerifyOtpFirstBindTokenValidity:
|
||
"""验证首次绑定签发的 token 可用于后续认证请求"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_first_bind_token_can_be_used_for_auth(self, otp_client, db_session, mock_redis):
|
||
"""首次绑定签发的 token 应能用于调用 /agents/me 等需认证端点。"""
|
||
secret = pyotp.random_base32()
|
||
agent = create_test_agent(user_id="token_test_001", name="Token验证坐席")
|
||
agent.mfa_secret = secret
|
||
agent.mfa_enabled = False
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
pre_token = await _create_token_in_redis(mock_redis, "token_test_001", "Token验证坐席")
|
||
|
||
otp_code = pyotp.TOTP(secret).now()
|
||
resp = await otp_client.post(
|
||
"/auth/otp-verify",
|
||
headers=_bearer(pre_token),
|
||
json={"otp_code": otp_code},
|
||
)
|
||
bind_token = resp.json()["data"]["token"]
|
||
|
||
# 用绑定签发的 token 调用 /agents/me
|
||
me_resp = await otp_client.get("/agents/me", headers=_bearer(bind_token))
|
||
assert me_resp.status_code == 200
|
||
me_data = me_resp.json()
|
||
assert me_data["code"] == 0
|
||
assert me_data["data"]["user_id"] == "token_test_001"
|
||
|
||
|
||
# =============================================================================
|
||
# Part C: 管理后台端点
|
||
# =============================================================================
|
||
|
||
class TestAdminOtpUsersEndpoint:
|
||
"""C1-C2: GET /auth/otp-admin-users"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_admin_list_users_includes_mfa_fields(self, client, db_session, mock_redis):
|
||
"""管理员查询列表 → 返回字段含 mfa_enabled / mfa_bound_at / mfa_last_verified_at。"""
|
||
# 创建测试坐席
|
||
agent1 = create_test_agent(user_id="adm_u_001", name="坐席A")
|
||
agent2 = create_test_agent(user_id="adm_u_002", name="坐席B")
|
||
agent2.mfa_enabled = True
|
||
agent2.mfa_secret = pyotp.random_base32()
|
||
agent2.mfa_bound_at = datetime.now()
|
||
db_session.add_all([agent1, agent2])
|
||
|
||
# 管理员
|
||
admin = create_test_agent(user_id="admin_otp_001", name="管理员")
|
||
db_session.add(admin)
|
||
await db_session.flush()
|
||
await _seed_admin_role(db_session, "admin_otp_001")
|
||
|
||
admin_token = await _create_token_in_redis(
|
||
mock_redis, "admin_otp_001", "管理员", ["admin"]
|
||
)
|
||
|
||
response = await client.get(
|
||
"/auth/otp-admin-users",
|
||
headers=_bearer(admin_token),
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
assert body["code"] == 0
|
||
users = body["data"]
|
||
assert isinstance(users, list)
|
||
assert len(users) >= 2
|
||
|
||
# 验证字段完整性
|
||
for user in users:
|
||
assert "employee_id" in user
|
||
assert "name" in user
|
||
assert "mfa_enabled" in user
|
||
assert "mfa_bound_at" in user # 可为 null
|
||
assert "mfa_last_verified_at" in user # 可为 null
|
||
|
||
# 验证坐席A (mfa_enabled=False)
|
||
user_a = next(u for u in users if u["employee_id"] == "adm_u_001")
|
||
assert user_a["mfa_enabled"] is False
|
||
assert user_a["mfa_bound_at"] is None
|
||
|
||
# 验证坐席B (mfa_enabled=True)
|
||
user_b = next(u for u in users if u["employee_id"] == "adm_u_002")
|
||
assert user_b["mfa_enabled"] is True
|
||
assert user_b["mfa_bound_at"] is not None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_non_admin_access_returns_403(self, client, db_session, mock_redis):
|
||
"""非 admin 角色访问 → 403。"""
|
||
agent = create_test_agent(user_id="normal_agent_001", name="普通坐席")
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
token = await _create_token_in_redis(
|
||
mock_redis, "normal_agent_001", "普通坐席", ["agent"]
|
||
)
|
||
|
||
response = await client.get(
|
||
"/auth/otp-admin-users",
|
||
headers=_bearer(token),
|
||
)
|
||
|
||
assert response.status_code == 403, \
|
||
f"非 admin 应返回 403，实际: {response.status_code}"
|
||
|
||
|
||
class TestAdminResetOtpEndpoint:
|
||
"""C3: POST /auth/otp-admin-reset/{employee_id}"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_admin_reset_clears_binding(self, otp_client, db_session, mock_redis):
|
||
"""管理员清除绑定 → DB 清空 mfa_* 字段。"""
|
||
secret = pyotp.random_base32()
|
||
target = create_test_agent(user_id="reset_target_001", name="被重置坐席")
|
||
target.mfa_secret = secret
|
||
target.mfa_enabled = True
|
||
target.mfa_bound_at = datetime.now()
|
||
target.mfa_last_verified_at = datetime.now()
|
||
db_session.add(target)
|
||
|
||
admin = create_test_agent(user_id="admin_reset_001", name="重置管理员")
|
||
db_session.add(admin)
|
||
await db_session.flush()
|
||
await _seed_admin_role(db_session, "admin_reset_001")
|
||
|
||
admin_token = await _create_token_in_redis(
|
||
mock_redis, "admin_reset_001", "重置管理员", ["admin"]
|
||
)
|
||
|
||
# 先写 Redis 验证标记
|
||
await mock_redis.setex("mfa:verified:reset_target_001", 1800, "1")
|
||
|
||
response = await otp_client.post(
|
||
"/auth/otp-admin-reset/reset_target_001",
|
||
headers=_bearer(admin_token),
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
body = response.json()
|
||
assert body["code"] == 0
|
||
assert body["data"]["success"] is True
|
||
|
||
# DB 验证
|
||
stmt = select(Agent).where(Agent.user_id == "reset_target_001")
|
||
db_target = (await db_session.execute(stmt)).scalars().first()
|
||
assert db_target.mfa_secret is None, \
|
||
f"secret 应为 None，实际: {db_target.mfa_secret}"
|
||
assert db_target.mfa_enabled is False, \
|
||
f"enabled 应为 False，实际: {db_target.mfa_enabled}"
|
||
assert db_target.mfa_bound_at is None, \
|
||
f"bound_at 应为 None，实际: {db_target.mfa_bound_at}"
|
||
|
||
# Redis 验证标记应被清除
|
||
assert not await mock_redis.exists("mfa:verified:reset_target_001"), \
|
||
"Redis 验证标记应被清除"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_admin_reset_nonexistent_agent_returns_error(self, client, db_session, mock_redis):
|
||
"""清除不存在的坐席 → 错误。"""
|
||
admin = create_test_agent(user_id="admin_reset_002", name="重置管理员2")
|
||
db_session.add(admin)
|
||
await db_session.flush()
|
||
await _seed_admin_role(db_session, "admin_reset_002")
|
||
|
||
admin_token = await _create_token_in_redis(
|
||
mock_redis, "admin_reset_002", "重置管理员2", ["admin"]
|
||
)
|
||
|
||
response = await client.post(
|
||
"/auth/otp-admin-reset/nonexistent_999",
|
||
headers=_bearer(admin_token),
|
||
)
|
||
|
||
body = response.json()
|
||
assert body["code"] != 0, \
|
||
f"不存在的坐席应返回错误，实际: {body}"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_non_admin_cannot_reset(self, client, db_session, mock_redis):
|
||
"""非 admin 角色调用 reset → 403。"""
|
||
agent = create_test_agent(user_id="normal_reset_001", name="普通坐席")
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
token = await _create_token_in_redis(
|
||
mock_redis, "normal_reset_001", "普通坐席", ["agent"]
|
||
)
|
||
|
||
response = await client.post(
|
||
"/auth/otp-admin-reset/some_target",
|
||
headers=_bearer(token),
|
||
)
|
||
|
||
assert response.status_code == 403, \
|
||
f"非 admin 应返回 403，实际: {response.status_code}"
|
||
|
||
|
||
# =============================================================================
|
||
# Part D: 身份验证缺口探查（Auth Gap Detection）
|
||
# =============================================================================
|
||
|
||
class TestAuthGapDetection:
|
||
"""探查 require_otp_bind 后的认证缺口。
|
||
|
||
当前设计：agent_login 对 mfa_enabled=False 返回 require_otp_bind 但不签发 token，
|
||
但后续 otp-bind / otp-verify 端点需要 Bearer token 认证。
|
||
"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_otp_bind_requires_auth(self, client, db_session, mock_redis):
|
||
"""无 token 调用 otp-bind → 401/403。
|
||
|
||
这验证了认证缺口：require_otp_bind 之后前端无 token 可用来调用 otp-bind。
|
||
"""
|
||
response = await client.post("/auth/otp-bind") # 无 Authorization 头
|
||
|
||
# 期望 401（未认证）或 403（禁止访问）
|
||
assert response.status_code in (401, 403), \
|
||
f"无 token 调用 otp-bind 应返回 401/403，实际: {response.status_code}"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_otp_verify_requires_auth(self, client, db_session, mock_redis):
|
||
"""无 token 调用 otp-verify → 401/403。
|
||
|
||
验证：require_otp_bind 之后无法直接调 otp-verify。
|
||
"""
|
||
response = await client.post(
|
||
"/auth/otp-verify",
|
||
json={"otp_code": "123456"},
|
||
) # 无 Authorization 头
|
||
|
||
assert response.status_code in (401, 403), \
|
||
f"无 token 调用 otp-verify 应返回 401/403，实际: {response.status_code}"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_full_first_bind_flow_auth_gap(self, otp_client, db_session, mock_redis):
|
||
"""BUG-001 修复后：完整首次绑定流程无需 token 注入即可完成。
|
||
|
||
修复前: login → require_otp_bind（无 token）→ otp-bind → 401
|
||
修复后: login → require_otp_bind + token → otp-bind → otp-verify → 成功
|
||
"""
|
||
agent = create_test_agent(user_id="gap_test_001", name="缺口测试坐席")
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
# Step 1: 登录 → 得到 require_otp_bind + 半认证 token
|
||
data = await _login_and_get_token(otp_client, "gap_test_001", "缺口测试坐席")
|
||
assert data.get("require_otp_bind") is True
|
||
assert "token" in data, \
|
||
f"BUG-001 修复: require_otp_bind 应携带 token，实际: {data}"
|
||
bind_token = data["token"]
|
||
|
||
# Step 2: 用半认证 token 调 otp-bind → 应成功（不再 401）
|
||
bind_resp = await otp_client.post(
|
||
"/auth/otp-bind", headers=_bearer(bind_token)
|
||
)
|
||
assert bind_resp.status_code == 200, \
|
||
f"BUG-001 修复: 半认证 token 应能访问 otp-bind，" \
|
||
f"实际: {bind_resp.status_code}, body: {bind_resp.json()}"
|
||
bind_data = bind_resp.json()["data"]
|
||
secret = bind_data["secret"]
|
||
|
||
# Step 3: 用半认证 token + 正确 OTP 调 otp-verify → 成功
|
||
otp_code = pyotp.TOTP(secret).now()
|
||
verify_resp = await otp_client.post(
|
||
"/auth/otp-verify",
|
||
headers=_bearer(bind_token),
|
||
json={"otp_code": otp_code},
|
||
)
|
||
assert verify_resp.status_code == 200
|
||
verify_data = verify_resp.json()["data"]
|
||
assert verify_data["verified"] is True
|
||
assert verify_data.get("is_first_bind") is True
|
||
assert "token" in verify_data, \
|
||
f"首次绑定应返回完整 token，实际: {verify_data}"
|
||
|
||
# Step 4: 验证返回的完整 token 可用于认证
|
||
full_token = verify_data["token"]
|
||
me_resp = await otp_client.get("/agents/me", headers=_bearer(full_token))
|
||
assert me_resp.status_code == 200
|
||
assert me_resp.json()["data"]["user_id"] == "gap_test_001"
|
||
|
||
|
||
# =============================================================================
|
||
# Part E: 全链路端到端流程（with workaround token injection）
|
||
# =============================================================================
|
||
|
||
class TestEndToEndFirstBindFlow:
|
||
"""端到端首次绑定流程（使用 token 注入绕过认证缺口验证业务逻辑正确性）"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_e2e_first_bind_flow(self, otp_client, db_session, mock_redis):
|
||
"""全链路：登录→获取 secret→验证→获得 token→认证可用。
|
||
|
||
通过 token 注入来绕过当前认证缺口，验证业务逻辑链路的正确性。
|
||
"""
|
||
# 1. 创建坐席并登录（mfa_enabled=False）
|
||
agent = create_test_agent(user_id="e2e_test_001", name="E2E测试坐席")
|
||
db_session.add(agent)
|
||
await db_session.flush()
|
||
|
||
login_data = await _login_and_get_token(otp_client, "e2e_test_001", "E2E测试坐席")
|
||
assert login_data.get("require_otp_bind") is True
|
||
|
||
# 2. 注入 token（模拟认证状态）
|
||
token = await _create_token_in_redis(mock_redis, "e2e_test_001", "E2E测试坐席")
|
||
|
||
# 3. 调用 otp-bind 获取 secret 和二维码
|
||
bind_resp = await otp_client.post("/auth/otp-bind", headers=_bearer(token))
|
||
assert bind_resp.status_code == 200
|
||
bind_data = bind_resp.json()["data"]
|
||
assert "secret" in bind_data
|
||
assert "otpauth_url" in bind_data
|
||
assert "qr_code_base64" in bind_data
|
||
secret = bind_data["secret"]
|
||
|
||
# 4. 用生成的 secret 计算 OTP 并验证
|
||
otp_code = pyotp.TOTP(secret).now()
|
||
verify_resp = await otp_client.post(
|
||
"/auth/otp-verify",
|
||
headers=_bearer(token),
|
||
json={"otp_code": otp_code},
|
||
)
|
||
assert verify_resp.status_code == 200
|
||
verify_data = verify_resp.json()["data"]
|
||
assert verify_data["verified"] is True
|
||
assert verify_data.get("is_first_bind") is True
|
||
assert "token" in verify_data, "首次绑定应返回登录 token"
|
||
|
||
# 5. 用绑定返回的 token 调用认证端点
|
||
bind_token = verify_data["token"]
|
||
me_resp = await otp_client.get("/agents/me", headers=_bearer(bind_token))
|
||
assert me_resp.status_code == 200
|
||
me_data = me_resp.json()
|
||
assert me_data["data"]["user_id"] == "e2e_test_001"
|
||
|
||
|
||
# =============================================================================
|
||
# Part F: Reset 后重新绑定流程
|
||
# =============================================================================
|
||
|
||
class TestResetThenRebind:
|
||
"""管理员清除绑定后坐席重新走首次绑定流程"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_reset_then_login_returns_require_otp_bind(self, client, db_session, mock_redis):
|
||
"""管理员清除后，坐席登录应再次返回 require_otp_bind。"""
|
||
# 1. 创建已绑定的坐席
|
||
secret = pyotp.random_base32()
|
||
agent = create_test_agent(user_id="reset_rebind_001", name="重置重绑坐席")
|
||
agent.mfa_secret = secret
|
||
agent.mfa_enabled = True
|
||
agent.mfa_bound_at = datetime.now()
|
||
db_session.add(agent)
|
||
|
||
# 2. 管理员
|
||
admin = create_test_agent(user_id="admin_rebind_001", name="重绑管理员")
|
||
db_session.add(admin)
|
||
await db_session.flush()
|
||
await _seed_admin_role(db_session, "admin_rebind_001")
|
||
|
||
# 3. 管理员清除绑定
|
||
admin_token = await _create_token_in_redis(
|
||
mock_redis, "admin_rebind_001", "重绑管理员", ["admin"]
|
||
)
|
||
reset_resp = await client.post(
|
||
"/auth/otp-admin-reset/reset_rebind_001",
|
||
headers=_bearer(admin_token),
|
||
)
|
||
assert reset_resp.json()["data"]["success"] is True
|
||
|
||
# 4. 坐席重新登录 → 应返回 require_otp_bind
|
||
data = await _login_and_get_token(client, "reset_rebind_001", "重置重绑坐席")
|
||
assert data.get("require_otp_bind") is True, \
|
||
f"清除后登录应返回 require_otp_bind，实际: {data}"
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"])
|