feat(simulation): add portable gaussian provider connector

2026-08-26 00:24:53 +03:00
parent 3ff3b9b587
commit 73fbfddf6c
4 changed files with 749 additions and 0 deletions
@@ -0,0 +1,414 @@
+"""Portable connector for the external DC Gaussian Pipeline service."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import os
+import re
+from collections.abc import Mapping
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Final
+from urllib.parse import quote, unquote, urljoin, urlparse
+
+import httpx
+
+TUS_VERSION: Final = "1.0.0"
+BUILD_REQUEST_SCHEMA: Final = "gaussian-pipeline.build-request/v1"
+JOB_SCHEMA: Final = "gaussian-pipeline.job/v1"
+RESULT_SCHEMA: Final = "gaussian-pipeline.build-result/v1"
+CAPABILITIES_SCHEMA: Final = "gaussian-pipeline.capabilities/v1"
+SAFE_UPLOAD_ID: Final = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
+SHA256_PATTERN: Final = re.compile(r"^[a-f0-9]{64}$")
+DEFAULT_CHUNK_BYTES: Final = 8 * 1024 * 1024
+MAX_JSON_RESPONSE_BYTES: Final = 32 * 1024 * 1024
+MAX_RETRIES: Final = 3
+
+
+class GaussianPipelineGatewayError(RuntimeError):
+    """The configured Gaussian provider violated or rejected its connector contract."""
+
+
+class GaussianPipelineConfigurationError(GaussianPipelineGatewayError):
+    """The provider connector configuration is incomplete or unsafe."""
+
+
+class GaussianPipelineUnavailableError(GaussianPipelineGatewayError):
+    """The provider cannot currently be reached."""
+
+
+class GaussianPipelineIntegrityError(GaussianPipelineGatewayError):
+    """Transferred bytes do not match their immutable descriptor."""
+
+
+@dataclass(frozen=True, slots=True)
+class GaussianSourceUpload:
+    upload_id: str
+    filename: str
+    format: str
+    sha256: str
+    byte_length: int
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "upload_id": self.upload_id,
+            "filename": self.filename,
+            "format": self.format,
+            "sha256": self.sha256,
+            "byte_length": self.byte_length,
+        }
+
+
+class GaussianPipelineGateway:
+    """TUS and JSON client with bounded responses and independent digest checks."""
+
+    def __init__(
+        self,
+        endpoint: str,
+        token_file: Path,
+        *,
+        timeout_seconds: float = 30.0,
+        chunk_bytes: int = DEFAULT_CHUNK_BYTES,
+        transport: httpx.BaseTransport | None = None,
+    ) -> None:
+        self.endpoint = _endpoint(endpoint)
+        self.token_file = token_file.expanduser().absolute()
+        self.token = _read_token(self.token_file)
+        if timeout_seconds <= 0:
+            raise GaussianPipelineConfigurationError("provider timeout must be positive")
+        if chunk_bytes <= 0 or chunk_bytes > 64 * 1024 * 1024:
+            raise GaussianPipelineConfigurationError("provider upload chunk size is invalid")
+        self.chunk_bytes = chunk_bytes
+        self._client = httpx.Client(
+            base_url=self.endpoint,
+            headers={"Authorization": f"Bearer {self.token}"},
+            timeout=httpx.Timeout(timeout_seconds),
+            transport=transport,
+            follow_redirects=False,
+        )
+
+    def close(self) -> None:
+        self._client.close()
+
+    def __enter__(self) -> GaussianPipelineGateway:
+        return self
+
+    def __exit__(self, *_: object) -> None:
+        self.close()
+
+    def capabilities(self) -> dict[str, Any]:
+        document = self._json("GET", "/v1/capabilities")
+        if (
+            document.get("schema_version") != CAPABILITIES_SCHEMA
+            or document.get("service") != "ndc-gaussian-pipeline"
+        ):
+            raise GaussianPipelineGatewayError("Gaussian provider capabilities do not match v1")
+        return document
+
+    def upload_source(
+        self,
+        source_path: Path,
+        *,
+        filename: str,
+        source_format: str,
+        sha256: str,
+    ) -> GaussianSourceUpload:
+        source_candidate = source_path.expanduser().absolute()
+        if source_candidate.is_symlink() or not source_candidate.is_file():
+            raise GaussianPipelineIntegrityError("Gaussian source must be one regular file")
+        source = source_candidate.resolve()
+        if source_format not in {"lcc", "lcc2"}:
+            raise GaussianPipelineIntegrityError("Gaussian source format must be lcc or lcc2")
+        if Path(filename).name != filename or not filename.lower().endswith(f".{source_format}"):
+            raise GaussianPipelineIntegrityError("Gaussian source filename is unsafe")
+        if SHA256_PATTERN.fullmatch(sha256) is None:
+            raise GaussianPipelineIntegrityError("Gaussian source digest is invalid")
+        byte_length = source.stat().st_size
+        if byte_length <= 0:
+            raise GaussianPipelineIntegrityError("Gaussian source is empty")
+        if _sha256(source) != sha256:
+            raise GaussianPipelineIntegrityError("Gaussian source digest does not match")
+
+        metadata = _tus_metadata(
+            {"filename": filename, "format": source_format, "sha256": sha256}
+        )
+        try:
+            response = self._client.post(
+                "/v1/uploads",
+                headers={
+                    "Tus-Resumable": TUS_VERSION,
+                    "Upload-Length": str(byte_length),
+                    "Upload-Metadata": metadata,
+                },
+                content=b"",
+            )
+            response.raise_for_status()
+        except httpx.HTTPError as exc:
+            raise _unavailable("Gaussian upload could not be created", exc) from exc
+        location = response.headers.get("Location")
+        if location is None:
+            raise GaussianPipelineGatewayError("Gaussian upload response omitted Location")
+        upload_url = urljoin(str(response.request.url), location)
+        if not _is_provider_upload_url(self.endpoint, upload_url):
+            raise GaussianPipelineGatewayError("Gaussian upload Location escaped the provider")
+        upload_id = unquote(urlparse(upload_url).path.rsplit("/", 1)[-1])
+        if SAFE_UPLOAD_ID.fullmatch(upload_id) is None:
+            raise GaussianPipelineGatewayError("Gaussian upload id is invalid")
+        self._send_file(source, upload_url, byte_length)
+        return GaussianSourceUpload(
+            upload_id=upload_id,
+            filename=filename,
+            format=source_format,
+            sha256=sha256,
+            byte_length=byte_length,
+        )
+
+    def submit_build(self, document: Mapping[str, object]) -> dict[str, Any]:
+        if document.get("schema_version") != BUILD_REQUEST_SCHEMA:
+            raise GaussianPipelineGatewayError("Gaussian build request schema is invalid")
+        result = self._json("POST", "/v1/jobs", document=document)
+        if result.get("schema_version") != JOB_SCHEMA:
+            raise GaussianPipelineGatewayError("Gaussian job response does not match v1")
+        return result
+
+    def get_job(self, job_id: str) -> dict[str, Any]:
+        _safe_id(job_id, "job id")
+        result = self._json("GET", f"/v1/jobs/{quote(job_id, safe='')}")
+        if result.get("schema_version") != JOB_SCHEMA or result.get("job_id") != job_id:
+            raise GaussianPipelineGatewayError("Gaussian job identity does not match")
+        return result
+
+    def get_result(self, job_id: str) -> dict[str, Any]:
+        _safe_id(job_id, "job id")
+        result = self._json("GET", f"/v1/jobs/{quote(job_id, safe='')}/result")
+        if result.get("schema_version") != RESULT_SCHEMA or result.get("job_id") != job_id:
+            raise GaussianPipelineGatewayError("Gaussian result identity does not match")
+        return result
+
+    def download_artifact(
+        self,
+        job_id: str,
+        descriptor: Mapping[str, object],
+        destination: Path,
+    ) -> Path:
+        _safe_id(job_id, "job id")
+        logical_path = descriptor.get("logical_path")
+        expected_sha256 = descriptor.get("sha256")
+        expected_bytes = descriptor.get("byte_length")
+        if (
+            not isinstance(logical_path, str)
+            or not logical_path
+            or logical_path.startswith("/")
+            or "\\" in logical_path
+            or any(part in {"", ".", ".."} for part in logical_path.split("/"))
+        ):
+            raise GaussianPipelineGatewayError("Gaussian artifact logical path is invalid")
+        if (
+            not isinstance(expected_sha256, str)
+            or SHA256_PATTERN.fullmatch(expected_sha256) is None
+        ):
+            raise GaussianPipelineGatewayError("Gaussian artifact digest is invalid")
+        if (
+            not isinstance(expected_bytes, int)
+            or isinstance(expected_bytes, bool)
+            or expected_bytes < 0
+        ):
+            raise GaussianPipelineGatewayError("Gaussian artifact byte length is invalid")
+        target = destination.expanduser().resolve()
+        target.parent.mkdir(parents=True, exist_ok=True)
+        temporary = target.with_name(f".{target.name}.partial-{os.getpid()}")
+        digest = hashlib.sha256()
+        byte_length = 0
+        temporary_created = False
+        try:
+            with self._client.stream(
+                "GET",
+                f"/v1/jobs/{quote(job_id, safe='')}/artifacts/{quote(logical_path, safe='/')}",
+            ) as response:
+                response.raise_for_status()
+                with temporary.open("xb") as output:
+                    temporary_created = True
+                    for chunk in response.iter_bytes(chunk_size=1024 * 1024):
+                        output.write(chunk)
+                        digest.update(chunk)
+                        byte_length += len(chunk)
+            if byte_length != expected_bytes or digest.hexdigest() != expected_sha256:
+                raise GaussianPipelineIntegrityError(
+                    "Gaussian artifact bytes do not match the result manifest"
+                )
+            temporary.replace(target)
+            return target
+        except httpx.HTTPError as exc:
+            raise _unavailable("Gaussian artifact download failed", exc) from exc
+        finally:
+            if temporary_created:
+                temporary.unlink(missing_ok=True)
+
+    def _send_file(self, source: Path, upload_url: str, byte_length: int) -> None:
+        offset = self._upload_offset(upload_url, byte_length)
+        retries = 0
+        with source.open("rb") as input_file:
+            while offset < byte_length:
+                input_file.seek(offset)
+                chunk = input_file.read(min(self.chunk_bytes, byte_length - offset))
+                if not chunk:
+                    raise GaussianPipelineIntegrityError(
+                        "Gaussian source ended before upload length"
+                    )
+                try:
+                    response = self._client.patch(
+                        upload_url,
+                        headers={
+                            "Tus-Resumable": TUS_VERSION,
+                            "Upload-Offset": str(offset),
+                            "Content-Type": "application/offset+octet-stream",
+                        },
+                        content=chunk,
+                    )
+                    response.raise_for_status()
+                    next_offset = _offset(response.headers.get("Upload-Offset"), byte_length)
+                    if next_offset != offset + len(chunk):
+                        raise GaussianPipelineGatewayError(
+                            "Gaussian upload returned a non-contiguous offset"
+                        )
+                    offset = next_offset
+                    retries = 0
+                except httpx.TransportError as exc:
+                    retries += 1
+                    if retries > MAX_RETRIES:
+                        raise _unavailable(
+                            "Gaussian upload failed after resume attempts", exc
+                        ) from exc
+                    offset = self._upload_offset(upload_url, byte_length)
+                except httpx.HTTPStatusError as exc:
+                    if exc.response.status_code == 409 and retries < MAX_RETRIES:
+                        retries += 1
+                        offset = self._upload_offset(upload_url, byte_length)
+                        continue
+                    raise _unavailable("Gaussian upload was rejected", exc) from exc
+
+    def _upload_offset(self, upload_url: str, byte_length: int) -> int:
+        try:
+            response = self._client.head(
+                upload_url,
+                headers={"Tus-Resumable": TUS_VERSION},
+            )
+            response.raise_for_status()
+        except httpx.HTTPError as exc:
+            raise _unavailable("Gaussian upload offset is unavailable", exc) from exc
+        return _offset(response.headers.get("Upload-Offset"), byte_length)
+
+    def _json(
+        self,
+        method: str,
+        path: str,
+        *,
+        document: Mapping[str, object] | None = None,
+    ) -> dict[str, Any]:
+        try:
+            response = self._client.request(method, path, json=document)
+            response.raise_for_status()
+        except httpx.HTTPError as exc:
+            raise _unavailable("Gaussian provider request failed", exc) from exc
+        if len(response.content) > MAX_JSON_RESPONSE_BYTES:
+            raise GaussianPipelineGatewayError("Gaussian provider JSON response is too large")
+        try:
+            value = response.json()
+        except json.JSONDecodeError as exc:
+            raise GaussianPipelineGatewayError("Gaussian provider response is not JSON") from exc
+        if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
+            raise GaussianPipelineGatewayError("Gaussian provider response must be an object")
+        return value
+
+
+def configured_gaussian_pipeline_gateway() -> GaussianPipelineGateway | None:
+    endpoint = os.environ.get("MISSIONCORE_GAUSSIAN_ENDPOINT")
+    token_file = os.environ.get("MISSIONCORE_GAUSSIAN_TOKEN_FILE")
+    if endpoint is None and token_file is None:
+        return None
+    if not endpoint or not token_file:
+        raise GaussianPipelineConfigurationError(
+            "MISSIONCORE_GAUSSIAN_ENDPOINT and MISSIONCORE_GAUSSIAN_TOKEN_FILE must be set together"
+        )
+    return GaussianPipelineGateway(endpoint, Path(token_file))
+
+
+def _endpoint(value: str) -> str:
+    parsed = urlparse(value)
+    if (
+        parsed.scheme not in {"http", "https"}
+        or not parsed.hostname
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.path not in {"", "/"}
+        or parsed.query
+        or parsed.fragment
+    ):
+        raise GaussianPipelineConfigurationError("Gaussian provider endpoint is invalid")
+    return value.rstrip("/")
+
+
+def _is_provider_upload_url(endpoint: str, upload_url: str) -> bool:
+    provider = urlparse(endpoint)
+    upload = urlparse(upload_url)
+    return (
+        upload.scheme == provider.scheme
+        and upload.hostname == provider.hostname
+        and upload.port == provider.port
+        and upload.username is None
+        and upload.password is None
+        and upload.path.startswith("/v1/uploads/")
+        and not upload.params
+        and not upload.query
+        and not upload.fragment
+    )
+
+
+def _read_token(token_file: Path) -> str:
+    try:
+        metadata = token_file.lstat()
+        if token_file.is_symlink() or not token_file.is_file():
+            raise GaussianPipelineConfigurationError("Gaussian token must be one regular file")
+        token = token_file.read_text(encoding="utf-8").strip()
+    except OSError as exc:
+        raise GaussianPipelineConfigurationError("Gaussian token file is unavailable") from exc
+    if metadata.st_size > 4096 or not 32 <= len(token) <= 4096:
+        raise GaussianPipelineConfigurationError("Gaussian token length is invalid")
+    return token
+
+
+def _tus_metadata(values: Mapping[str, str]) -> str:
+    return ",".join(
+        f"{key} {base64.b64encode(value.encode('utf-8')).decode('ascii')}"
+        for key, value in values.items()
+    )
+
+
+def _sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _offset(value: str | None, byte_length: int) -> int:
+    if value is None or not value.isdigit():
+        raise GaussianPipelineGatewayError("Gaussian upload offset is invalid")
+    offset = int(value)
+    if not 0 <= offset <= byte_length:
+        raise GaussianPipelineGatewayError("Gaussian upload offset is outside the source")
+    return offset
+
+
+def _safe_id(value: str, label: str) -> None:
+    if SAFE_UPLOAD_ID.fullmatch(value) is None:
+        raise GaussianPipelineGatewayError(f"Gaussian {label} is invalid")
+
+
+def _unavailable(message: str, error: httpx.HTTPError) -> GaussianPipelineGatewayError:
+    if isinstance(error, httpx.TransportError):
+        return GaussianPipelineUnavailableError(message)
+    return GaussianPipelineGatewayError(message)
@@ -0,0 +1,66 @@
+"""Read-only Mission Core projection of the external Gaussian build provider."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any, Literal
+
+from fastapi import APIRouter
+from pydantic import BaseModel, ConfigDict
+
+from k1link.simulation.gaussian_pipeline_gateway import (
+    GaussianPipelineGateway,
+    GaussianPipelineGatewayError,
+    configured_gaussian_pipeline_gateway,
+)
+
+ProviderFactory = Callable[[], GaussianPipelineGateway | None]
+
+
+class SimulationWorldProviderStatus(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    schema_version: Literal["missioncore.simulation-world-provider-status/v1"] = (
+        "missioncore.simulation-world-provider-status/v1"
+    )
+    provider_id: Literal["gaussian-pipeline"] = "gaussian-pipeline"
+    configured: bool
+    state: Literal["ready", "unavailable"]
+    capabilities: dict[str, Any] | None
+
+
+def build_simulation_world_provider_router(
+    provider_factory: ProviderFactory = configured_gaussian_pipeline_gateway,
+) -> APIRouter:
+    router = APIRouter()
+
+    @router.get(
+        "/api/v1/simulation-worlds/provider",
+        response_model=SimulationWorldProviderStatus,
+    )
+    def provider_status() -> SimulationWorldProviderStatus:
+        provider: GaussianPipelineGateway | None = None
+        try:
+            provider = provider_factory()
+            if provider is None:
+                return SimulationWorldProviderStatus(
+                    configured=False,
+                    state="unavailable",
+                    capabilities=None,
+                )
+            return SimulationWorldProviderStatus(
+                configured=True,
+                state="ready",
+                capabilities=provider.capabilities(),
+            )
+        except GaussianPipelineGatewayError:
+            return SimulationWorldProviderStatus(
+                configured=True,
+                state="unavailable",
+                capabilities=None,
+            )
+        finally:
+            if provider is not None:
+                provider.close()
+
+    return router