ouyangbo/EvoScientist
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 3 Wiki Activity

feat(model-registry): add admin config export endpoints with plaintext secrets

Browse Source
This commit is contained in:
m4
2026-08-12 20:07:59 +08:00
parent aae8d0a379
commit 2a7cccd598
2 changed files with 274 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
EvoScientist/model_registry/http_api.py
+97
View File
@@ -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)
tests/test_config_export.py
+177
View File
@@ -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