Steven
2b493bd9f8
CI / backend (push) Canceled after 0s
跑通迁移与测试后修掉的真问题： - alembic 0001：asyncpg 走预备语句，一次 execute 只能一条 SQL。 CREATE FUNCTION / CREATE POLICY / GRANT 全部拆成单语句逐条执行， 否则报 "cannot insert multiple commands into a prepared statement"。 - 创建工作区：ORM flush 只按 relationship 排序，裸外键不算数， member 可能先于 workspace 落地撞 FK。改为先 flush workspace 再插 member。 - datetime.UTC 是 3.11+ 才有的，本机 3.10 直接 ImportError。改用 timezone.utc。 - 幂等中间件：原来会缓存 4xx。校验失败被缓存后，用户改对参数仍拿到同一个 422。 改为只缓存 <400。 - 发码节流窗口改为可配置 SMS_THROTTLE_SECONDS（测试置 0）， 节流用例自己显式打开，避免"整库关掉节流"导致节流从未被验证。 - 租户隔离用例：pytest.raises(Exception) 改为精确的 ProgrammingError， 否则任何异常都会让"应用角色不能建表"这条断言假通过。 验收：pytest 33 passed；ruff check / format 通过；真机跑 uvicorn， /health、/version、登录、建工作区、列表、/auth/me 全通。
116 lines
4.6 KiB
Python
116 lines
4.6 KiB
Python
"""跨租户隔离 —— 本仓库**最重要的测试**（AGENTS.md 2.5）。
|
|
|
|
三层验证，任何一层失效都算事故：
|
|
1. 接口层：A 租户的令牌打不到 B 租户的资源。
|
|
2. 数据层：应用角色连库、绑定 A 的上下文，物理上看不到 B 的行（RLS 生效）。
|
|
3. 配置层：RLS 确实 ENABLE，且应用角色不是 superuser / BYPASSRLS（否则 RLS 被跳过）。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import ProgrammingError
|
|
|
|
from app.db import SessionFactory, bind_tenant, current_app_role
|
|
from app.models import TENANT_TABLES
|
|
|
|
from .conftest import create_workspace, login, unwrap
|
|
|
|
|
|
async def _two_tenants(client: AsyncClient):
|
|
alice = await login(client, "13900000001")
|
|
bob = await login(client, "13900000002")
|
|
ws_alice = await create_workspace(client, alice, "Alice 的团队")
|
|
ws_bob = await create_workspace(client, bob, "Bob 的团队")
|
|
return alice, bob, ws_alice, ws_bob
|
|
|
|
|
|
async def test_cannot_read_other_workspace(client: AsyncClient) -> None:
|
|
alice, _bob, _ws_a, ws_b = await _two_tenants(client)
|
|
resp = await client.get(f"/api/v1/workspaces/{ws_b['id']}", headers=alice["headers"])
|
|
assert resp.status_code == 403
|
|
assert resp.json()["code"] == 3003
|
|
|
|
|
|
async def test_cannot_list_other_workspace_members(client: AsyncClient) -> None:
|
|
alice, _bob, _ws_a, ws_b = await _two_tenants(client)
|
|
resp = await client.get(f"/api/v1/workspaces/{ws_b['id']}/members", headers=alice["headers"])
|
|
assert resp.status_code == 403
|
|
|
|
|
|
async def test_cannot_invite_into_other_workspace(client: AsyncClient) -> None:
|
|
alice, _bob, _ws_a, ws_b = await _two_tenants(client)
|
|
resp = await client.post(
|
|
f"/api/v1/workspaces/{ws_b['id']}/invitations",
|
|
json={"role": "member"},
|
|
headers=alice["headers"],
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
async def test_my_workspace_list_is_scoped(client: AsyncClient) -> None:
|
|
alice, _bob, ws_a, ws_b = await _two_tenants(client)
|
|
mine = unwrap(await client.get("/api/v1/workspaces", headers=alice["headers"]))
|
|
ids = [w["id"] for w in mine["items"]]
|
|
assert ids == [ws_a["id"]]
|
|
assert ws_b["id"] not in ids
|
|
|
|
|
|
async def test_rls_hides_rows_at_database_level(client: AsyncClient) -> None:
|
|
_alice, _bob, ws_a, ws_b = await _two_tenants(client)
|
|
ws_a_id = uuid.UUID(ws_a["id"])
|
|
ws_b_id = uuid.UUID(ws_b["id"])
|
|
|
|
async with SessionFactory() as session, session.begin():
|
|
await bind_tenant(session, ws_a_id, None)
|
|
visible = (await session.execute(text("SELECT id FROM workspace"))).scalars().all()
|
|
member_ids = (
|
|
(await session.execute(text("SELECT workspace_id FROM member"))).scalars().all()
|
|
)
|
|
|
|
assert [str(v) for v in visible] == [ws_a["id"]]
|
|
assert set(member_ids) == {ws_a_id}
|
|
assert ws_b_id not in set(member_ids)
|
|
|
|
|
|
async def test_rls_without_context_returns_nothing(client: AsyncClient) -> None:
|
|
await _two_tenants(client)
|
|
async with SessionFactory() as session, session.begin():
|
|
count = (await session.execute(text("SELECT count(*) FROM workspace"))).scalar_one()
|
|
assert count == 0, "未绑定租户上下文时不应看到任何行"
|
|
|
|
|
|
async def test_rls_enabled_on_every_tenant_table(client: AsyncClient) -> None:
|
|
async with SessionFactory() as session:
|
|
rows = (
|
|
await session.execute(
|
|
text(
|
|
"SELECT relname, relrowsecurity FROM pg_class "
|
|
"WHERE relname = ANY(:names) AND relkind = 'r'"
|
|
),
|
|
{"names": list(TENANT_TABLES)},
|
|
)
|
|
).all()
|
|
status = dict(rows)
|
|
assert set(status) == set(TENANT_TABLES), "有租户表没建出来"
|
|
assert all(status.values()), f"未启用 RLS 的表：{[k for k, v in status.items() if not v]}"
|
|
|
|
|
|
async def test_app_role_cannot_bypass_rls(client: AsyncClient) -> None:
|
|
async with SessionFactory() as session, session.begin():
|
|
role = await current_app_role(session)
|
|
assert role["is_super"] is False, "应用角色绝不能是 superuser，否则 RLS 被跳过"
|
|
assert role["bypass_rls"] is False, "应用角色不能带 BYPASSRLS"
|
|
assert role["read_all"] is False
|
|
|
|
|
|
async def test_app_role_has_no_ddl(client: AsyncClient) -> None:
|
|
"""应用角色不能建表——否则一次 SQL 注入就能改结构。"""
|
|
async with SessionFactory() as session, session.begin():
|
|
with pytest.raises(ProgrammingError): # 权限不足，DDL 必须失败
|
|
await session.execute(text("CREATE TABLE should_not_exist (id int)"))
|