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

from __future__ import annotations

import json
from datetime import UTC, datetime
from pathlib import Path
from types import SimpleNamespace

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 (
    PhaseAdvanceResult,
    ScriptMissionStartResult,
)
from script_build_host.domain.artifacts import (
    ArtifactKind,
    ArtifactState,
    ArtifactVersion,
    EvidenceRecordV1,
)
from script_build_host.domain.errors import ArtifactNotFound
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
        self.advanced = False

    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=(),
        )

    async def advance_to_phase_two(
        self, script_build_id: int, principal: Principal
    ) -> PhaseAdvanceResult:
        assert principal.subject == "owner"
        self.advanced = True
        return PhaseAdvanceResult(script_build_id, BuildStatus.RUNNING, "root-7", "11")


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:
    def __init__(self) -> None:
        criterion = SimpleNamespace(criterion_id="criterion-1", description="safe", hard=True)
        current_spec = SimpleNamespace(
            version=1,
            objective="bounded task",
            acceptance_criteria=(criterion,),
            context_refs=("script-build://task-kinds/paragraph",),
        )
        self.task = SimpleNamespace(
            task_id="task-7",
            parent_task_id="root-task",
            display_path="Root/task-7",
            status="completed",
            current_spec=current_spec,
            child_task_ids=(),
            attempt_ids=(),
            validation_ids=(),
            decision_ids=(),
            blocked_reason=None,
            superseded_by=None,
            created_at="2026-07-19T00:00:00Z",
            updated_at="2026-07-19T00:00:00Z",
        )

    async def load(self, root_trace_id: str) -> object:
        del root_trace_id
        return SimpleNamespace(
            tasks={"task-7": self.task},
            to_dict=lambda: {
                "root_trace_id": "root-7",
                "protected_context": {"token": "must-not-leak"},
                "command_records": [{"arguments": "secret"}],
            },
        )


def _app(
    principal: Principal | None,
    *,
    trace_store: object | None = None,
    uploaded_topics: object | None = None,
    business_artifacts: 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=business_artifacts or 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.prompt_requests) >= 8
        assert mission.command.runtime_prompt_manifest == ()

        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
        assert allowed.json() == {"root_trace_id": "root-7"}
        assert "must-not-leak" not in allowed.text
        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


@pytest.mark.asyncio
async def test_phase_two_advance_auth_and_hidden_not_found() -> 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/7/phase-two/advance")
        assert response.status_code == 401

    forbidden, forbidden_mission = _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/7/phase-two/advance")
        assert response.status_code == 404
        assert response.json()["detail"]["error_code"] == "BUILD_NOT_FOUND"
        assert forbidden_mission.advanced is False

    allowed, mission = _app(Principal("owner"))
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=allowed), base_url="http://test"
    ) as client:
        response = await client.post("/api/pattern/script_builds/7/phase-two/advance")
        assert response.status_code == 200
        assert response.json() == {
            "success": True,
            "script_build_id": 7,
            "status": "running",
            "root_trace_id": "root-7",
            "input_snapshot_id": "11",
        }
        assert mission.advanced is True


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:
    def __init__(self, trace_id: str, root_trace_id: str | None = None) -> None:
        self.trace_id = trace_id
        self.context = {"root_trace_id": root_trace_id} if root_trace_id is not None else {}

    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:
        values = {
            "root-7": _Trace("root-7"),
            "worker-7": _Trace("worker-7", "root-7"),
            "worker-8": _Trace("worker-8", "root-8"),
        }
        return values.get(trace_id)

    async def get_goal_tree(self, trace_id: str) -> None:
        del trace_id
        return None

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


class _BusinessArtifacts:
    def __init__(self) -> None:
        now = datetime.now(UTC)
        evidence = EvidenceRecordV1(
            evidence_id="evidence-api-1",
            source_type="decode",
            tool_name="retrieve_decode",
            query={"query": "safe query"},
            source_refs=("decode:item:1",),
            raw_artifact_ref=None,
            summary="authorized phase-two artifact content",
            supports=("scope:opening",),
            confidence="high",
            limitations=(),
            content_sha256="sha256:" + "a" * 64,
            created_at=now,
        )
        self.value = ArtifactVersion(
            artifact_version_id=71,
            script_build_id=7,
            task_id="task-7",
            attempt_id="attempt-7",
            spec_version=1,
            artifact_type=ArtifactKind.EVIDENCE,
            canonical_sha256="sha256:" + "b" * 64,
            state=ArtifactState.FROZEN,
            artifact=evidence,
            created_at=now,
            frozen_at=now,
        )

    async def get_by_id(self, artifact_version_id: int, *, script_build_id: int) -> ArtifactVersion:
        if artifact_version_id != 71 or script_build_id != 7:
            raise ArtifactNotFound()
        return self.value


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"}

        for foreign_trace_id in ("worker-8", "root-8"):
            with pytest.raises(WebSocketDisconnect) as caught:
                with client.websocket_connect(
                    f"/api/pattern/script_builds/7/traces/{foreign_trace_id}/watch",
                    headers={"origin": "https://ui.example"},
                ):
                    pass
            assert caught.value.code == 4404


@pytest.mark.asyncio
async def test_phase_two_observation_hides_foreign_resources_and_returns_owned_artifact() -> None:
    app, _ = _app(
        Principal("owner"),
        trace_store=_TraceStore(),
        business_artifacts=_BusinessArtifacts(),
    )
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as client:
        owned_task = await client.get("/api/pattern/script_builds/7/tasks/task-7")
        assert owned_task.status_code == 200
        assert owned_task.json()["task_id"] == "task-7"

        foreign_task = await client.get("/api/pattern/script_builds/7/tasks/task-8")
        assert foreign_task.status_code == 404
        assert foreign_task.json()["detail"]["error_code"] == "TASK_NOT_FOUND"

        owned_artifact = await client.get("/api/pattern/script_builds/7/artifacts/71")
        assert owned_artifact.status_code == 200
        assert (
            owned_artifact.json()["artifact"]["summary"] == "authorized phase-two artifact content"
        )

        foreign_artifact = await client.get("/api/pattern/script_builds/7/artifacts/81")
        assert foreign_artifact.status_code == 404
        assert foreign_artifact.json()["error_code"] == "ARTIFACT_NOT_FOUND"

        owned_trace = await client.get("/api/pattern/script_builds/7/traces/worker-7")
        assert owned_trace.status_code == 200
        assert owned_trace.json()["trace"]["trace_id"] == "worker-7"

        foreign_trace = await client.get("/api/pattern/script_builds/7/traces/worker-8")
        assert foreign_trace.status_code == 404
        assert foreign_trace.json()["detail"]["error_code"] == "TRACE_NOT_FOUND"

        forged_root = await client.get("/api/v2/script-builds/7/roots/root-8/snapshot")
        assert forged_root.status_code == 404
        assert forged_root.json()["detail"]["error_code"] == "ROOT_NOT_FOUND"


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