script-build-and-framework-tmp/script_build_host/tests/test_api_security.py

from __future__ import annotations

import json
from pathlib import Path
from types import SimpleNamespace
from typing import ClassVar

import httpx
import pytest
from fastapi.testclient import TestClient
from starlette.websockets import WebSocketDisconnect

from script_build_host.api import ApiSecurity, create_app
from script_build_host.application.mission_service import ScriptMissionStartResult
from script_build_host.domain.records import BuildStatus, Principal
from script_build_host.infrastructure.outbound import OutboundPolicy


async def _public_resolver(_host: str, _port: int) -> tuple[str, ...]:
    return ("93.184.216.34",)


class _PrincipalProvider:
    def __init__(self, principal: Principal | None) -> None:
        self.value = principal

    async def current(self, request_context: object | None = None) -> Principal:
        assert request_context is not None
        if self.value is None:
            raise PermissionError("missing")
        return self.value


class _Authorizer:
    async def require_source_access(self, principal: Principal, **_source: int) -> None:
        if principal.subject != "owner":
            raise PermissionError("forbidden")

    async def require_access(self, principal: Principal, script_build_id: int) -> None:
        if principal.subject != "owner" or script_build_id != 7:
            raise PermissionError("forbidden")


class _MissionService:
    def __init__(self) -> None:
        self.command = None

    async def start(self, command: object) -> ScriptMissionStartResult:
        self.command = command
        return ScriptMissionStartResult(7, BuildStatus.RUNNING, "root-7", "11")

    async def stop(self, script_build_id: int, principal: Principal) -> object:
        del principal
        return SimpleNamespace(
            script_build_id=script_build_id,
            status=BuildStatus.STOPPED,
            stopped_operation_ids=(),
        )


class _Bindings:
    async def get_by_build(self, script_build_id: int) -> object:
        return SimpleNamespace(script_build_id=script_build_id, root_trace_id="root-7")


class _TaskStore:
    async def load(self, root_trace_id: str) -> object:
        del root_trace_id
        return SimpleNamespace(to_dict=lambda: {"root_trace_id": "root-7"})


def _app(
    principal: Principal | None,
    *,
    trace_store: object | None = None,
    uploaded_topics: object | None = None,
) -> tuple[object, _MissionService]:
    mission = _MissionService()
    app = create_app(
        mission_service=mission,
        security=ApiSecurity(
            _PrincipalProvider(principal),
            _Authorizer(),
            websocket_allowed_origins=("https://ui.example",),
        ),
        bindings=_Bindings(),
        business_artifacts=SimpleNamespace(),
        publications=SimpleNamespace(),
        coordinator=SimpleNamespace(task_store=_TaskStore()),
        trace_store=trace_store or SimpleNamespace(),
        uploaded_topics=uploaded_topics,
        outbound_policy=OutboundPolicy(
            frozenset({"data.example"}),
            resolver=_public_resolver,
        ),
    )
    return app, mission


@pytest.mark.asyncio
async def test_start_requires_auth_and_preserves_old_response_fields() -> None:
    unauthenticated, _ = _app(None)
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=unauthenticated), base_url="http://test"
    ) as client:
        response = await client.post(
            "/api/pattern/script_builds",
            json={"execution_id": 1, "topic_build_id": 2, "topic_id": 3},
        )
        assert response.status_code == 401

    forbidden, _ = _app(Principal("intruder"))
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=forbidden), base_url="http://test"
    ) as client:
        response = await client.post(
            "/api/pattern/script_builds",
            json={"execution_id": 1, "topic_build_id": 2, "topic_id": 3},
        )
        assert response.status_code == 404

    authenticated, mission = _app(Principal("owner"))
    legacy_request = json.loads(
        (Path(__file__).parent / "fixtures" / "legacy_script_build_request.json").read_text(
            encoding="utf-8"
        )
    )
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=authenticated), base_url="http://test"
    ) as client:
        response = await client.post(
            "/api/pattern/script_builds",
            json=legacy_request,
        )
        assert response.status_code == 200, response.text
        body = response.json()
        assert body["success"] is True
        assert body["script_build_id"] == 7
        assert body["status"] == "running"
        assert mission.command.strategies_always_on == (13,)
        assert len(mission.command.runtime_prompt_manifest) == 8

        secret_config = dict(legacy_request)
        secret_config["agent_config"] = {
            **legacy_request["agent_config"],
            "api_key": "must-not-persist",
        }
        secret_response = await client.post(
            "/api/pattern/script_builds",
            json=secret_config,
        )
        assert secret_response.status_code == 200
        assert "must-not-persist" not in str(mission.command.agent_config)

        safe = await client.post(
            "/api/pattern/script_builds",
            json={
                "execution_id": 1,
                "topic_build_id": 2,
                "topic_id": 3,
                "data_source_url": "https://data.example/input",
            },
        )
        assert safe.status_code == 200
        assert mission.command.data_source_url == "https://data.example/input"
        assert mission.command.datasource_manifest == {
            "data_source_host": "data.example",
            "scheme": "https",
        }


@pytest.mark.asyncio
async def test_observation_is_build_scoped_and_control_routes_are_not_mounted() -> None:
    app, _ = _app(Principal("owner"))
    paths = set(app.openapi()["paths"])
    assert not any(path.endswith("/operations") for path in paths)
    assert not any(path.startswith("/api/traces") for path in paths)
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as client:
        allowed = await client.get("/api/pattern/script_builds/7/mission")
        assert allowed.status_code == 200
        denied = await client.get("/api/pattern/script_builds/8/mission")
        assert denied.status_code == 404
        unsafe = await client.post(
            "/api/pattern/script_builds",
            json={
                "execution_id": 1,
                "topic_build_id": 2,
                "topic_id": 3,
                "data_source_url": "http://127.0.0.1/private",
            },
        )
        assert unsafe.status_code == 400
        secret_query = await client.post(
            "/api/pattern/script_builds",
            json={
                "execution_id": 1,
                "topic_build_id": 2,
                "topic_id": 3,
                "data_source_url": "https://data.example/private?token=secret",
            },
        )
        assert secret_query.status_code == 400


def test_websocket_rejects_unauthenticated_subscription_before_accept() -> None:
    app, _ = _app(None)
    with TestClient(app) as client:
        with pytest.raises(WebSocketDisconnect) as caught:
            with client.websocket_connect("/api/pattern/script_builds/7/traces/root-7/watch"):
                pass
    assert caught.value.code == 4404


class _Trace:
    trace_id = "root-7"
    context: ClassVar[dict[str, str]] = {}

    def to_dict(self) -> dict[str, str]:
        return {"trace_id": self.trace_id}


class _TraceStore:
    async def get_trace(self, trace_id: str) -> _Trace | None:
        return _Trace() if trace_id == "root-7" else None

    async def get_events(self, trace_id: str, since: int) -> list[dict[str, object]]:
        assert trace_id == "root-7"
        del since
        return []


def test_websocket_requires_origin_and_accepts_authorized_build_trace() -> None:
    app, _ = _app(Principal("owner"), trace_store=_TraceStore())
    with TestClient(app) as client:
        with pytest.raises(WebSocketDisconnect) as caught:
            with client.websocket_connect(
                "/api/pattern/script_builds/7/traces/root-7/watch",
                headers={"origin": "https://evil.example"},
            ):
                pass
        assert caught.value.code == 4404

        with client.websocket_connect(
            "/api/pattern/script_builds/7/traces/root-7/watch",
            headers={"origin": "https://ui.example"},
        ) as websocket:
            websocket.send_text("ping")
            assert websocket.receive_json() == {"event": "pong"}


class _UploadedTopics:
    async def parse(self, _value: object) -> dict[str, object]:
        return {
            "account_name": "acct",
            "topic_fusion": "topic",
            "points": [{"point_type": "关键点"}],
            "item_count": 1,
        }

    async def create(
        self, _parsed: dict[str, object], *, account_name: str | None
    ) -> dict[str, object]:
        return {
            "topic_build_id": 22,
            "topic_id": 33,
            "account_name": account_name or "acct",
            "item_count": 1,
            "point_count": 1,
        }


@pytest.mark.asyncio
async def test_upload_parse_and_start_routes_keep_legacy_response_contract() -> None:
    app, mission = _app(Principal("owner"), uploaded_topics=_UploadedTopics())
    payload = {"选题融合": "topic"}
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as client:
        parsed = await client.post(
            "/api/pattern/script_builds/parse_topic_json",
            json={"json_data": payload},
        )
        assert parsed.status_code == 200
        assert parsed.json()["point_type_stats"] == {"关键点": 1}
        started = await client.post(
            "/api/pattern/script_builds/from_topic_json",
            json={"json_data": payload, "account_name": "acct"},
        )
        assert started.status_code == 200
        assert started.json()["topic_build_id"] == 22
        assert started.json()["topic_id"] == 33
        assert mission.command.execution_id == 0