140 lines
5.4 KiB
Python
140 lines
5.4 KiB
Python
"""批次 0 安全修复回归测试（TECH_DEBT D5 等）。
|
||
|
||
覆盖：
|
||
- /api/status/{task_id} 鉴权与任务归属校验（无 token 401 / 他人任务 403 / 不存在 404 / 所有者 200）
|
||
- bcrypt 72 字节上限：超长密码显式拒绝而非静默截断
|
||
- SECRET_KEY 惰性校验：未配置时给出明确错误
|
||
"""
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from httpx import AsyncClient, ASGITransport
|
||
from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
|
||
|
||
from moldinsight.api.task_router import router as task_router
|
||
from shared.database.database import get_db_session
|
||
from shared.models.database import User, STPFile, ProcessingTask
|
||
from shared.services.auth_service import (
|
||
get_current_active_user,
|
||
get_password_hash,
|
||
verify_password,
|
||
create_access_token,
|
||
)
|
||
from shared.config.settings import settings
|
||
|
||
|
||
@pytest.fixture(scope="function")
|
||
async def status_client(async_engine, seeded_db):
|
||
"""带 task_router 的测试应用：播种两个任务（owner=user 1 / user 999）。"""
|
||
session_factory = async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)
|
||
|
||
async with session_factory() as session:
|
||
stp_own = STPFile(
|
||
id=9001, user_id=1, object_key="test/own.step", storage_bucket="moldinsight",
|
||
original_filename="own.step", file_size=128, status="completed",
|
||
)
|
||
task_own = ProcessingTask(
|
||
task_id="task-owned", stp_file_id=9001,
|
||
task_type="stp_parsing", status="completed", parameters={},
|
||
)
|
||
stp_other = STPFile(
|
||
id=9002, user_id=999, object_key="test/other.step", storage_bucket="moldinsight",
|
||
original_filename="other.step", file_size=128, status="completed",
|
||
)
|
||
task_other = ProcessingTask(
|
||
task_id="task-foreign", stp_file_id=9002,
|
||
task_type="stp_parsing", status="completed", parameters={},
|
||
)
|
||
session.add_all([stp_own, task_own, stp_other, task_other])
|
||
await session.commit()
|
||
|
||
test_app = FastAPI()
|
||
# 生产环境中 /api 前缀由入口层挂载时添加，测试中需显式指定才能对齐真实路径
|
||
test_app.include_router(task_router, prefix="/api")
|
||
|
||
async def override_get_db_session():
|
||
async with session_factory() as session:
|
||
yield session
|
||
|
||
test_app.dependency_overrides[get_db_session] = override_get_db_session
|
||
|
||
transport = ASGITransport(app=test_app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
||
yield ac, test_app
|
||
|
||
test_app.dependency_overrides.clear()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_status_without_token_is_401(status_client):
|
||
"""未携带 token 访问 /api/status 必须拒绝（此前该端点完全未鉴权）。"""
|
||
ac, _ = status_client
|
||
resp = await ac.post("/api/status/task-owned")
|
||
assert resp.status_code == 401
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_status_owner_can_view(status_client):
|
||
ac, app = status_client
|
||
app.dependency_overrides[get_current_active_user] = lambda: User(id=1, username="tester")
|
||
|
||
resp = await ac.post("/api/status/task-owned")
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["task_id"] == "task-owned"
|
||
assert body["status"] == "completed"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_status_non_owner_is_403(status_client):
|
||
"""他人任务（含无主任务）必须 403，不允许凭任务号枚举。"""
|
||
ac, app = status_client
|
||
app.dependency_overrides[get_current_active_user] = lambda: User(id=2, username="intruder")
|
||
|
||
resp = await ac.post("/api/status/task-owned")
|
||
assert resp.status_code == 403
|
||
|
||
resp = await ac.post("/api/status/task-foreign")
|
||
assert resp.status_code == 403
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_status_unknown_task_is_404(status_client):
|
||
ac, app = status_client
|
||
app.dependency_overrides[get_current_active_user] = lambda: User(id=1, username="tester")
|
||
|
||
resp = await ac.post("/api/status/task-missing")
|
||
assert resp.status_code == 404
|
||
|
||
|
||
def test_password_hash_rejects_over_72_bytes():
|
||
"""bcrypt 72 字节上限：必须显式报错，不能静默截断。"""
|
||
with pytest.raises(ValueError, match="72"):
|
||
get_password_hash("a" * 73)
|
||
# 多字节字符按字节数计：25 个汉字 = 75 字节
|
||
with pytest.raises(ValueError, match="72"):
|
||
get_password_hash("模" * 25)
|
||
|
||
|
||
def test_password_hash_roundtrip_at_limit():
|
||
password = "a" * 72
|
||
hashed = get_password_hash(password)
|
||
assert verify_password(password, hashed)
|
||
# 历史口令按 bcrypt 语义截断比较：73 字节输入截断后与 72 字节口令匹配（兼容旧数据），
|
||
# 但新口令在 get_password_hash 处已被显式拒绝，不会再产生这类哈希
|
||
assert verify_password(password + "x", hashed)
|
||
assert not verify_password("b" * 72, hashed)
|
||
|
||
|
||
def test_verify_password_over_72_bytes_returns_false_not_raise():
|
||
"""超长密码登录不得抛 ValueError（否则登录接口 500），应返回 False 走正常失败路径。"""
|
||
hashed = get_password_hash("short-password")
|
||
assert not verify_password("长" * 40, hashed) # 120 字节
|
||
assert not verify_password("a" * 73, hashed)
|
||
|
||
|
||
def test_create_token_without_secret_key_is_explicit_error(monkeypatch):
|
||
"""SECRET_KEY 未配置时给出可读错误，而不是 jwt.encode 的晦涩 TypeError。"""
|
||
monkeypatch.setattr(settings, "SECRET_KEY", None)
|
||
with pytest.raises(RuntimeError, match="SECRET_KEY"):
|
||
create_access_token({"sub": "tester"})
|