ZY-Agent/backend/tests/test_api_key_control_plane.py

"""API key control-plane default-deny tests (Stage 1 收口).

service principal (API key) 只能访问数据平面 (threads/runs/assistants);
控制平面 (models/mcp/memory/skills/channels/agents 与管理/auth) 一律 403。
真人 cookie 路径不受影响。设计见 spec
docs/superpowers/specs/2026-06-28-api-key-control-plane-default-deny-design.md。
"""

from __future__ import annotations

import pytest
from fastapi import Request
from starlette.testclient import TestClient

from app.gateway.auth_middleware import _is_dataplane_path
from deerflow.auth.tokens import generate_api_key

pytestmark = pytest.mark.anyio


@pytest.mark.parametrize(
    "path",
    [
        "/api/threads",
        "/api/threads/abc",
        "/api/v1/threads",
        "/api/v1/threads/abc/runs/xyz/feedback",
        "/api/runs",
        "/api/runs/stream",
        "/api/v1/runs/stream",
        "/api/assistants",
        "/api/assistants/search",
    ],
)
def test_dataplane_paths_allowed(path):
    assert _is_dataplane_path(path) is True


@pytest.mark.parametrize(
    "path",
    [
        "/api/models",
        "/api/v1/models",
        "/api/mcp/config",
        "/api/v1/mcp/config",
        "/api/v1/memory",
        "/api/v1/skills/install",
        "/api/v1/channels/restart",
        "/api/v1/agents",
        "/api/v1/service-accounts",
        "/api/v1/api-keys",
        "/api/v1/auth/me",
        "/api/v1/assistants",  # assistants 是 LangGraph 兼容 shim,无 /api/v1 孪生:只放行 /api/assistants,缺 v1 变体是有意为之
        "/api/langgraph/threads",  # nginx 死代码:中间件本看不到,真混进来也应 deny
        "/api/threads-export",  # boundary guard — prefix must end at a path segment
        "/api/runsX",  # boundary guard — prefix must end at a path segment
    ],
)
def test_control_plane_paths_denied(path):
    assert _is_dataplane_path(path) is False


# ---------------------------------------------------------------------------
# Integration tests: AuthMiddleware bearer default-deny (Task 3)
# ---------------------------------------------------------------------------


@pytest.fixture
def anyio_backend() -> str:
    return "asyncio"


async def _seed_key(tmp_path, *, scopes="threads:read"):
    from deerflow.persistence.api_key import ApiKeyRepository
    from deerflow.persistence.engine import get_session_factory, init_engine
    from deerflow.persistence.service_account.model import ServiceAccountRow
    from deerflow.persistence.user.model import UserRow
    from deerflow.persistence.workspace.model import WorkspaceRow

    url = f"sqlite+aiosqlite:///{tmp_path / 'test.db'}"
    await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
    sf = get_session_factory()
    async with sf() as session:
        session.add(UserRow(id="u-alice", email="alice@example.com"))
        await session.commit()
    async with sf() as session:
        session.add(WorkspaceRow(id="w-1", name="WS", slug="ws", owner_id="u-alice"))
        await session.commit()
    async with sf() as session:
        session.add(ServiceAccountRow(id="sa-1", workspace_id="w-1", name="bot", role="member", identity_mode="collapsed", status="active", created_by="u-alice"))
        await session.commit()
    repo = ApiKeyRepository(sf)
    gen = generate_api_key("live")
    await repo.create(service_account_id="sa-1", key_prefix=gen.prefix, key_hash=gen.key_hash, name="k", scopes=scopes)
    return gen


async def _cleanup():
    from deerflow.persistence.engine import close_engine

    await close_engine()


def _make_app():
    from fastapi import FastAPI

    from app.gateway.auth_middleware import AuthMiddleware
    from deerflow.runtime.user_context import get_effective_user_id

    app = FastAPI()
    app.add_middleware(AuthMiddleware)

    # NOTE: Request must NOT be imported locally here. With `from __future__ import
    # annotations` active, local imports are invisible to get_type_hints, causing
    # FastAPI to treat `request: Request` as a query param → 422. Module-level
    # import (above) makes it resolvable. See test_auth_middleware_api_key.py docstring.
    @app.get("/api/v1/threads/_probe")
    async def threads_probe(request: Request):
        return {"user_id": get_effective_user_id()}

    @app.get("/api/assistants/search")
    async def assistants_probe():
        return {"ok": True}

    return app


async def test_sa_allowed_on_dataplane(tmp_path):
    gen = await _seed_key(tmp_path)
    try:
        client = TestClient(_make_app())
        r = client.get("/api/v1/threads/_probe", headers={"Authorization": f"Bearer {gen.plaintext}"})
        assert r.status_code == 200
        assert r.json() == {"user_id": "sa-1"}
    finally:
        await _cleanup()


async def test_sa_allowed_on_assistants_init(tmp_path):
    gen = await _seed_key(tmp_path)
    try:
        client = TestClient(_make_app())
        r = client.get("/api/assistants/search", headers={"Authorization": f"Bearer {gen.plaintext}"})
        assert r.status_code == 200
    finally:
        await _cleanup()


@pytest.mark.parametrize(
    "path",
    [
        "/api/v1/mcp/config",
        "/api/mcp/config",
        "/api/v1/models",
        "/api/v1/skills/install",
        "/api/v1/channels/restart",
        "/api/v1/agents",
        "/api/v1/memory",
        "/api/v1/service-accounts",
    ],
)
async def test_sa_denied_on_control_plane(tmp_path, path):
    gen = await _seed_key(tmp_path)
    try:
        client = TestClient(_make_app())
        r = client.get(path, headers={"Authorization": f"Bearer {gen.plaintext}"})
        assert r.status_code == 403
        assert r.json()["detail"]["code"] == "insufficient_scope"
    finally:
        await _cleanup()


async def test_invalid_key_still_401_not_403(tmp_path):
    # 无效 key 命中控制平面路径,应是 401 (TOKEN_INVALID),不是 403 ——
    # deny 检查在 None 校验之后。
    await _seed_key(tmp_path)
    try:
        client = TestClient(_make_app())
        r = client.get("/api/v1/mcp/config", headers={"Authorization": "Bearer dfk_live_bogus00000000000000000"})
        assert r.status_code == 401
    finally:
        await _cleanup()


async def test_cookie_path_unaffected_by_deny(tmp_path):
    # 非 bearer-dfk 请求不进 bearer 分支:控制平面路径走 cookie 路径,
    # 无 cookie → 401 not_authenticated,绝不会拿到 403 insufficient_scope。
    await _seed_key(tmp_path)
    try:
        client = TestClient(_make_app())
        r = client.get("/api/v1/mcp/config")
        assert r.status_code == 401
        assert r.json()["detail"]["code"] != "insufficient_scope"
    finally:
        await _cleanup()