HF certification: enforce trusted dedicated isolation

docs/source/guides/runtime-providers.md

@@ -164,7 +164,8 @@ provider = DockerSwarmProvider()
 
 Runs a registry image in a Hugging Face Sandbox. Pooled mode is the default and
 is appropriate for trusted same-user rollout fan-out. Dedicated mode allocates
-one VM per sandbox and is suited to untrusted or GPU workloads.
+one VM per sandbox and is suited to untrusted or GPU workloads. Official
+validation requires dedicated mode and rejects pooled execution.
 
 ```python
 from openenv.core.containers.runtime.hf_sandbox_provider import HFSandboxProvider
@@ -178,7 +179,8 @@ provider.wait_for_ready(base_url)
 ```
 
 Plain `env_vars` are visible as Job environment metadata. Use the dedicated
-mode's `secrets=` argument for encrypted runtime secrets.
+mode's `secrets=` argument for encrypted runtime secrets. Official validation
+does not forward coordinator secrets through either channel.
 
 ### KubernetesProvider
 

src/openenv/core/containers/runtime/hf_sandbox_provider.py

@@ -10,6 +10,7 @@
 
 import asyncio
 import hashlib
+import re
 import socket
 import threading
 import time
@@ -39,6 +40,28 @@
 _SERVER_COMMAND = _POOLED_SERVER_COMMAND
 _MAX_WS_MESSAGE_SIZE = 100 * 1024 * 1024
 _MAX_HTTP_RESPONSE_SIZE = 100 * 1024 * 1024
+_CREDENTIAL_ENV_NAME = re.compile(
+    r"(?i)(?:^|[_-])(?:auth(?:orization)?|credentials?|password|passwd|"
+    r"private[_-]?key|secret|token|api[_-]?key|access[_-]?key(?:[_-]?id)?)"
+    r"(?:$|[_-])|(?:^|[_-])(?:jwt|pat|askpass)(?:$|[_-])"
+)
+_CREDENTIAL_ENV_ALIASES = frozenset(
+    {"PGPASSWORD", "PGPASSFILE", "MYSQL_PWD", "GIT_ASKPASS", "NETRC"}
+)
+_CREDENTIAL_ENV_VALUE = (
+    re.compile(r"(?i)\bbearer\s+\S+"),
+    re.compile(r"(?i)\b(?:hf_|sk-|github_pat_|gh[pousr]_)[A-Za-z0-9_-]{8,}\b"),
+    re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b"),
+    re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
+    re.compile(r"(?i)\b[a-z][a-z0-9+.-]*://[^/\s@]*:[^/\s@]+@"),
+    re.compile(
+        r"(?i)\b[a-z][a-z0-9+.-]*://[^\s#]*[?&]"
+        r"(?:auth(?:orization)?|credentials?|password|passwd|token|"
+        r"api[_-]?key|access[_-]?key|key|sig(?:nature)?|awsaccesskeyid|"
+        r"googleaccessid|x[-_]amz[-_](?:credential|security[-_]token|signature)|"
+        r"x[-_]goog[-_](?:credential|signature))=[^&#\s]+"
+    ),
+)
 _HOP_BY_HOP_HEADERS = {
     "connection",
     "content-encoding",
@@ -56,6 +79,21 @@
 _POOL_LOCK = threading.Lock()
 
 
+def credential_environment_names(env_vars: dict[str, str] | None) -> list[str]:
+    """Return names whose name or value appears to carry credentials."""
+    flagged: list[str] = []
+    for raw_name, raw_value in (env_vars or {}).items():
+        name = str(raw_name)
+        value = str(raw_value)
+        if (
+            name.upper() in _CREDENTIAL_ENV_ALIASES
+            or _CREDENTIAL_ENV_NAME.search(name)
+            or any(pattern.search(value) for pattern in _CREDENTIAL_ENV_VALUE)
+        ):
+            flagged.append(name)
+    return sorted(flagged)
+
+
 class _NoRedirectWebSocketConnect(ws_connect):
     def process_redirect(self, exc: Exception) -> Exception | str:
         return exc
@@ -300,6 +338,10 @@ def __init__(
         self._secrets = dict(secrets) if secrets is not None else None
         self.flavor = flavor
         self._mode = mode
+        self._official_certification = False
+        self._official_image: str | None = None
+        self._official_flavor: str | None = None
+        self._official_env_vars: dict[str, str] | None = None
         self._sandbox: Any = None
         self._server_process: Any = None
         self._proxy: _LocalAuthProxy | None = None
@@ -326,6 +368,34 @@ def isolation_mode(self) -> Literal["pooled", "dedicated", "unknown"]:
             return "unknown"
         return "dedicated" if host_id is None else "pooled"
 
+    def _lock_for_official_certification(self) -> None:
+        """Freeze the settings that cross the official runner trust boundary."""
+        if self._sandbox is not None:
+            raise RuntimeError(
+                "Official certification must validate and lock the provider "
+                "before a Hugging Face sandbox is started."
+            )
+        self._official_certification = True
+        self._official_image = self.image
+        self._official_flavor = self.flavor
+        self._official_env_vars = (
+            dict(self._env_vars) if self._env_vars is not None else None
+        )
+
+    def _verify_official_runtime_isolation(self) -> None:
+        try:
+            host_id = self._sandbox.host_id
+        except (AttributeError, RuntimeError) as exc:
+            raise RuntimeError(
+                "Official certification could not verify dedicated isolation "
+                "for the created Hugging Face sandbox."
+            ) from exc
+        if host_id is not None:
+            raise RuntimeError(
+                "Official certification requires a dedicated Hugging Face sandbox; "
+                "the created sandbox is attached to a shared host."
+            )
+
     def _create_sandbox(self, *, image: str, env_vars: dict[str, str] | None) -> Any:
         if self.mode == "dedicated":
             Sandbox = _get_sandbox_cls()
@@ -360,12 +430,37 @@ def start_container(
                 f"HFSandboxProvider only supports port {_DEFAULT_PORT} (got {port})."
             )
 
+        if self._official_certification:
+            if (
+                self.image != self._official_image
+                or self.flavor != self._official_flavor
+                or self.mode != "dedicated"
+                or self._env_vars != self._official_env_vars
+                or (image is not None and image != self._official_image)
+            ):
+                raise RuntimeError(
+                    "Official certification does not allow execution-setting changes "
+                    "after the provider trust check."
+                )
+            if env_vars is not None:
+                raise RuntimeError(
+                    "Official certification does not allow environment overrides "
+                    "after the provider trust check."
+                )
+            credential_names = credential_environment_names(self._env_vars)
+            if credential_names or self._secrets:
+                raise RuntimeError(
+                    "Official subject sandboxes cannot receive coordinator credentials."
+                )
+
         effective_image = self.image if image is None else image
         effective_env = self._env_vars if env_vars is None else env_vars
         self._sandbox = self._create_sandbox(
             image=effective_image, env_vars=effective_env
         )
         try:
+            if self._official_certification:
+                self._verify_official_runtime_isolation()
             if not hasattr(self._sandbox, "proxy_url_for") or not hasattr(
                 self._sandbox, "proxy_headers"
             ):

src/openenv/validation/__init__.py

@@ -24,6 +24,7 @@
     OPENENV_VALIDATION_POLICY,
     VALIDATION_POLICY_VERSION,
 )
+from .security import ensure_official_hf_sandbox
 
 __all__ = [
     "CheckOutcome",
@@ -42,6 +43,7 @@
     "ValidationSeverity",
     "ValidationStatus",
     "build_validation_plan",
+    "ensure_official_hf_sandbox",
     "execute_validation_plan",
     "format_shared_validation_report",
     "run_local_validation",

src/openenv/validation/security.py

@@ -0,0 +1,38 @@
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Fail-closed trust-boundary guards used by remote certification."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from openenv.core.containers.runtime.hf_sandbox_provider import (
+    credential_environment_names,
+)
+
+
+def ensure_official_hf_sandbox(provider: Any) -> None:
+    """Reject shared HF pools for official certification workloads."""
+    if getattr(provider, "isolation_mode", None) != "dedicated":
+        raise RuntimeError(
+            "Official certification requires a dedicated Hugging Face sandbox; "
+            "pooled execution is a same-user trust boundary."
+        )
+    if getattr(provider, "secrets", None):
+        raise RuntimeError(
+            "Official subject sandboxes cannot receive coordinator secrets."
+        )
+    env_vars = getattr(provider, "env_vars", None) or {}
+    credential_names = credential_environment_names(env_vars)
+    if credential_names:
+        raise RuntimeError(
+            "Official subject sandboxes cannot receive credential-like "
+            f"environment variables: {', '.join(credential_names)}"
+        )
+    lock = getattr(provider, "_lock_for_official_certification", None)
+    if not callable(lock):
+        raise RuntimeError(
+            "Official certification requires a provider that can lock its "
+            "validated execution settings."
+        )
+    lock()

tests/test_validation/test_security.py

@@ -0,0 +1,134 @@
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Tests for validation certification trust-boundary enforcement."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from openenv.core.containers.runtime import hf_sandbox_provider as provider_module
+from openenv.core.containers.runtime.hf_sandbox_provider import HFSandboxProvider
+from openenv.validation.security import ensure_official_hf_sandbox
+
+
+def test_official_certification_rejects_pooled_hf_sandbox() -> None:
+    pooled = HFSandboxProvider(image="example")
+    dedicated = HFSandboxProvider(image="example", mode="dedicated")
+
+    with pytest.raises(RuntimeError, match="dedicated"):
+        ensure_official_hf_sandbox(pooled)
+
+    ensure_official_hf_sandbox(dedicated)
+
+
+@pytest.mark.parametrize(
+    "env_vars",
+    [
+        {"HF_TOKEN": "secret"},
+        {"HF_AUTH": "hf_1234567890abcdef"},
+        {"AUTH": "Bearer coordinator-token"},
+        {"AWS_ACCESS_KEY_ID": "AKIA1234567890ABCDEF"},
+        {"MODE": "ASIA1234567890ABCDEF"},
+        {"DATABASE_URL": "https://alice:hunter2@example.com/database"},
+        {"DATABASE_URL": "postgresql://alice:hunter2@db/app"},
+        {"CACHE_URL": "redis://:hunter2@cache/0"},
+        {"PGPASSWORD": "secret"},
+        {"MYSQL_PWD": "secret"},
+        {"CI_JOB_JWT": "secret"},
+        {"GIT_ASKPASS": "/tmp/helper"},
+        {"GOOGLE_APPLICATION_CREDENTIALS": "/tmp/google.json"},
+        {"NETRC": "/tmp/netrc"},
+        {"MODE": "postgresql://db/app?password=hunter2"},
+        {"MODE": "redis://cache/0?token=hunter2"},
+        {"MODE": "https://service.example/path?api_key=hunter2"},
+        {
+            "DOWNLOAD_URL": (
+                "https://service.example/object?X-Amz-Credential=scope&"
+                "X-Amz-Signature=deadbeef"
+            )
+        },
+        {
+            "DOWNLOAD_URL": (
+                "https://service.example/object?X-Goog-Credential=scope&"
+                "X-Goog-Signature=deadbeef"
+            )
+        },
+        {"MODE": "hf_1234567890abcdef"},
+    ],
+)
+def test_official_certification_rejects_forwarded_credentials(
+    env_vars: dict[str, str],
+) -> None:
+    provider = HFSandboxProvider(
+        image="example",
+        mode="dedicated",
+        env_vars=env_vars,
+    )
+
+    with pytest.raises(RuntimeError, match="credential-like"):
+        ensure_official_hf_sandbox(provider)
+
+
+def test_official_certification_locks_start_environment() -> None:
+    provider = HFSandboxProvider(
+        image="example",
+        mode="dedicated",
+        env_vars={"MODE": "certify"},
+    )
+    ensure_official_hf_sandbox(provider)
+
+    with pytest.raises(RuntimeError, match="environment overrides"):
+        provider.start_container(env_vars={"HF_TOKEN": "secret"})
+
+
+def test_official_certification_detects_post_lock_environment_mutation() -> None:
+    provider = HFSandboxProvider(
+        image="example",
+        mode="dedicated",
+        env_vars={"MODE": "certify"},
+    )
+    ensure_official_hf_sandbox(provider)
+    assert provider._env_vars is not None
+    provider._env_vars["MODE"] = "changed"
+
+    with pytest.raises(RuntimeError, match="execution-setting changes"):
+        provider.start_container()
+
+
+def test_official_start_rejects_unverifiable_runtime_isolation() -> None:
+    class SandboxWithoutHostIdentity:
+        proxy_headers: dict[str, str] = {}
+
+        def run(self, *args: object, **kwargs: object) -> MagicMock:
+            return MagicMock()
+
+        def proxy_url_for(self, port: int, path: str) -> str:
+            return "https://sandbox.example/proxy"
+
+        def kill(self) -> None:
+            self._killed = True
+
+    sandbox = SandboxWithoutHostIdentity()
+    sandbox_cls = MagicMock()
+    sandbox_cls.create.return_value = sandbox
+    provider = HFSandboxProvider(image="example", mode="dedicated")
+    ensure_official_hf_sandbox(provider)
+
+    with patch.object(provider_module, "_get_sandbox_cls", return_value=sandbox_cls):
+        with pytest.raises(RuntimeError, match="verify dedicated isolation"):
+            provider.start_container()
+
+    assert provider._sandbox is None
+
+
+def test_official_certification_cannot_lock_active_sandbox() -> None:
+    provider = HFSandboxProvider(image="example", mode="dedicated")
+    sandbox = MagicMock()
+    sandbox.host_id = None
+    provider._sandbox = sandbox
+
+    with pytest.raises(RuntimeError, match="before.*started"):
+        ensure_official_hf_sandbox(provider)
+
+    assert provider._official_certification is False