From 2a7cccd598728866ccdff2de6c87ce6078e7aa71 Mon Sep 17 00:00:00 2001 From: m4 Date: Wed, 12 Aug 2026 20:07:59 +0800 Subject: [PATCH] feat(model-registry): add admin config export endpoints with plaintext secrets --- EvoScientist/model_registry/http_api.py | 97 +++++++++++++ tests/test_config_export.py | 177 ++++++++++++++++++++++++ 2 files changed, 274 insertions(+) create mode 100644 tests/test_config_export.py diff --git a/EvoScientist/model_registry/http_api.py b/EvoScientist/model_registry/http_api.py index 61f376c..beb9ddd 100644 --- a/EvoScientist/model_registry/http_api.py +++ b/EvoScientist/model_registry/http_api.py @@ -30,6 +30,7 @@ import sqlite3 import uuid from collections.abc import Callable from dataclasses import dataclass +from datetime import datetime, timezone from typing import Any, Literal from pydantic import ( @@ -133,6 +134,36 @@ class PutModelRegistryRequest(BaseModel): credential_writes: list[CredentialWrite] = Field(default_factory=list) +class RegistryExportCredential(BaseModel): + """One plaintext credential in the registry export envelope.""" + + credential_id: CredentialId + secret_value: str + + +class RegistryExportPayload(BaseModel): + registry: RegistryV4 + credentials: list[RegistryExportCredential] + + +class ModelRegistryExportResponse(BaseModel): + """Admin-only whole-registry export; contains plaintext secrets.""" + + kind: Literal["evoscientist.model-registry"] = "evoscientist.model-registry" + format_version: Literal[1] = 1 + exported_at: str + payload: RegistryExportPayload + + +class ImageGenerationExportResponse(BaseModel): + """Admin-only image-settings export; contains plaintext api keys.""" + + kind: Literal["evoscientist.image-generation"] = "evoscientist.image-generation" + format_version: Literal[1] = 1 + exported_at: str + payload: ImageGenerationSettings + + class CredentialReplaceRequest(BaseModel): """The section 9.3 credential rotation request.""" @@ -530,6 +561,11 @@ class ModelRegistryHttpApi: self.test_provider, methods=["POST"], ), + Route( + "/api/model-registry/export", + self.export_model_registry, + methods=["GET"], + ), Route("/api/models", self.get_selectable_models, methods=["GET"]), Route( "/api/image-generation", @@ -546,6 +582,11 @@ class ModelRegistryHttpApi: self.test_image_generation, methods=["POST"], ), + Route( + "/api/image-generation/export", + self.export_image_generation, + methods=["GET"], + ), Route("/api/runtime-snapshots", self.create_snapshot, methods=["POST"]), Route( "/api/runtime-snapshots/{snapshot_id}/bind", @@ -728,6 +769,28 @@ class ModelRegistryHttpApi: return _validation_error_response(exc, request_id) return JSONResponse(response) + async def export_model_registry(self, request: Request) -> Response: + request_id = uuid.uuid4().hex + try: + services, _actor = await self._authenticate( + request, required_scope="config:export", require_thread_id=False + ) + response = await asyncio.to_thread(self._registry_export, services) + except ModelRegistryError as exc: + return _error_response(exc, request_id) + return JSONResponse(response.model_dump(mode="json")) + + async def export_image_generation(self, request: Request) -> Response: + request_id = uuid.uuid4().hex + try: + await self._authenticate( + request, required_scope="config:export", require_thread_id=False + ) + response = await asyncio.to_thread(self._image_generation_export) + except ModelRegistryError as exc: + return _error_response(exc, request_id) + return JSONResponse(response.model_dump(mode="json")) + # --- snapshot API ---------------------------------------------------------- async def create_snapshot(self, request: Request) -> Response: @@ -938,6 +1001,40 @@ class ModelRegistryHttpApi: "models": models, } + @staticmethod + def _registry_export(services: ApiServices) -> ModelRegistryExportResponse: + store = services.store + registry = store.load_registry() + referenced = sorted( + { + provider.auth.credential_id + for provider in registry.providers + if provider.auth.credential_id is not None + } + ) + revisions = store.list_credential_revisions() + credentials = [ + RegistryExportCredential( + credential_id=credential_id, + secret_value=store.resolve_credential( + credential_id, revisions[credential_id] + ), + ) + for credential_id in referenced + if credential_id in revisions + ] + return ModelRegistryExportResponse( + exported_at=datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), + payload=RegistryExportPayload(registry=registry, credentials=credentials), + ) + + @staticmethod + def _image_generation_export() -> ImageGenerationExportResponse: + return ImageGenerationExportResponse( + exported_at=datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), + payload=load_image_generation_settings(), + ) + @staticmethod def _save_image_generation(body: Any) -> dict[str, Any]: request = PutImageGenerationRequest.model_validate(body) diff --git a/tests/test_config_export.py b/tests/test_config_export.py new file mode 100644 index 0000000..37ca3e4 --- /dev/null +++ b/tests/test_config_export.py @@ -0,0 +1,177 @@ +"""Contract tests for the admin config-export endpoints. + +These two GET routes deliberately break the "APIs never return secrets" +invariant behind the admin-only ``config:export`` scope (config +import/export design doc, 2026-08-12). +""" + +from __future__ import annotations + +import time +import uuid + +import jwt +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec +from starlette.applications import Starlette +from starlette.testclient import TestClient + +from EvoScientist.image_gen import config as image_config +from EvoScientist.image_gen.config import ImageGenerationSettings, ImageModelEntry +from EvoScientist.model_registry import http_api +from EvoScientist.model_registry.auth import BffAuthenticator +from EvoScientist.model_registry.endpoint_policy import EndpointPolicy +from EvoScientist.model_registry.http_api import ApiServices, model_registry_routes +from EvoScientist.model_registry.platform import DelegationPublicKey +from EvoScientist.model_registry.resolver import ModelRegistryResolver +from EvoScientist.model_registry.snapshots import SnapshotService +from EvoScientist.model_registry.store import ModelRuntimeStore +from tests.registry_fixtures import ZHIPU_SECRET, activate_store + +SERVICE_TOKEN = "bff-service-token" + +_PRIVATE_KEY = ec.generate_private_key(ec.SECP256R1()) +_PRIVATE_PEM = _PRIVATE_KEY.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), +) +_PUBLIC_PEM = _PRIVATE_KEY.public_key().public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, +) + + +@pytest.fixture +def config_path(tmp_path): + path = tmp_path / "config.yaml" + path.write_text("other_section: {keep: true}\n", encoding="utf-8") + return path + + +@pytest.fixture +def services(tmp_path, config_path, monkeypatch): + store = ModelRuntimeStore(config_dir=tmp_path / "runtime") + monkeypatch.setattr( + http_api, + "load_image_generation_settings", + lambda: image_config.load_image_generation_settings(config_path=config_path), + ) + resolver = ModelRegistryResolver(store) + return ApiServices( + store=store, + resolver=resolver, + snapshot_service=SnapshotService(store, resolver), + endpoint_policy=EndpointPolicy([]), + authenticator=BffAuthenticator( + service_token=SERVICE_TOKEN, + service_token_hash=None, + delegation_keys=( + DelegationPublicKey(deployment_id="webui-1", public_key=_PUBLIC_PEM), + ), + jti_store=store, + ), + ) + + +@pytest.fixture +def client(services): + app = Starlette(routes=model_registry_routes(lambda: services)) + return TestClient(app) + + +def _headers(scopes): + now = int(time.time()) + claims = { + "iss": "WebUI", + "aud": "EvoScientist", + "sub": "admin-1", + "scopes": list(scopes), + "deployment_id": "webui-1", + "iat": now, + "exp": now + 30, + "jti": uuid.uuid4().hex, + } + return { + "Authorization": f"Bearer {SERVICE_TOKEN}", + "X-Evo-Actor": jwt.encode(claims, _PRIVATE_PEM, algorithm="ES256"), + } + + +def _export_headers(): + return _headers(["config:export"]) + + +def test_registry_export_returns_registry_with_plaintext_credentials(client, services): + activate_store(services.store) + response = client.get("/api/model-registry/export", headers=_export_headers()) + assert response.status_code == 200 + body = response.json() + assert body["kind"] == "evoscientist.model-registry" + assert body["format_version"] == 1 + assert isinstance(body["exported_at"], str) and body["exported_at"] + provider_ids = {p["id"] for p in body["payload"]["registry"]["providers"]} + assert provider_ids == {"zhipu-glm", "local-ollama"} + assert body["payload"]["credentials"] == [ + {"credential_id": "zhipu-primary", "secret_value": ZHIPU_SECRET} + ] + + +def test_registry_export_empty_store_has_empty_credentials(client): + response = client.get("/api/model-registry/export", headers=_export_headers()) + assert response.status_code == 200 + assert response.json()["payload"]["credentials"] == [] + + +def test_registry_export_requires_export_scope(client, services): + activate_store(services.store) + response = client.get( + "/api/model-registry/export", + headers=_headers(["model_config:read", "model_config:write"]), + ) + assert response.status_code == 403 + assert response.json()["code"] == "FORBIDDEN" + + +def test_registry_export_requires_delegation(client): + response = client.get( + "/api/model-registry/export", + headers={"Authorization": f"Bearer {SERVICE_TOKEN}"}, + ) + assert response.status_code == 401 + + +def test_image_export_returns_plaintext_api_keys(client, config_path): + settings = ImageGenerationSettings( + default_model="gpt-image-2", + timeout_seconds=60.0, + models=[ + ImageModelEntry( + id="gpt-image-2", + name="GPT Image", + provider="openai", + api_key="sk-image-secret", + base_url="https://api.example.com/v1", + ) + ], + ) + image_config.save_image_generation_settings(settings, config_path=config_path) + response = client.get("/api/image-generation/export", headers=_export_headers()) + assert response.status_code == 200 + body = response.json() + assert body["kind"] == "evoscientist.image-generation" + assert body["format_version"] == 1 + payload = body["payload"] + assert payload["default_model"] == "gpt-image-2" + assert payload["timeout_seconds"] == 60.0 + assert payload["models"][0]["api_key"] == "sk-image-secret" + assert "api_key_configured" not in payload["models"][0] + + +def test_image_export_requires_export_scope(client): + response = client.get( + "/api/image-generation/export", + headers=_headers(["model_config:read", "model_config:write"]), + ) + assert response.status_code == 403