trending/OpenShell
1
0
Fork 0
mirror of https://github.com/NVIDIA/OpenShell.git synced 2026-10-16 05:20:22 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
deb-pkg
OpenShell/e2e/python/test_sandbox_providers.py
T
Derek Carr 5e28ea3a4b feat(server): add object meta convention to top-level objects (#919) 
- adds filterable label selectors on resources

Closes #864

Signed-off-by: Derek Carr <decarr@redhat.com>
2026-04-27 07:16:07 -07:00

413 lines
14 KiB
Python
Raw Permalink Blame History

# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
"""E2E tests for supervisor-managed provider placeholders in sandboxes.
Provider credentials are fetched at runtime by the sandbox supervisor via the
GetSandboxProviderEnvironment gRPC call. Sandboxed child processes should see
placeholder values (not raw secrets). Credentials must never be present in the
persisted sandbox spec environment map.
"""
from __future__ import annotations
from contextlib import contextmanager
from typing import TYPE_CHECKING
import grpc
import pytest
from openshell._proto import datamodel_pb2, openshell_pb2, sandbox_pb2
if TYPE_CHECKING:
from collections.abc import Callable, Iterator
from openshell import Sandbox, SandboxClient
# ---------------------------------------------------------------------------
# Policy helpers
# ---------------------------------------------------------------------------
def _default_policy() -> sandbox_pb2.SandboxPolicy:
"""Build a sandbox policy with standard filesystem/process/landlock settings."""
return sandbox_pb2.SandboxPolicy(
version=1,
filesystem=sandbox_pb2.FilesystemPolicy(
include_workdir=True,
read_only=["/usr", "/lib", "/etc", "/app", "/dev/urandom"],
read_write=["/sandbox", "/tmp"],
),
landlock=sandbox_pb2.LandlockPolicy(compatibility="best_effort"),
process=sandbox_pb2.ProcessPolicy(
run_as_user="sandbox", run_as_group="sandbox"
),
)
# ---------------------------------------------------------------------------
# Provider lifecycle helper
# ---------------------------------------------------------------------------
@contextmanager
def provider(
stub: object,
*,
name: str,
provider_type: str,
credentials: dict[str, str],
) -> Iterator[str]:
"""Create a provider for the duration of the block, then delete it."""
_delete_provider(stub, name)
stub.CreateProvider(
openshell_pb2.CreateProviderRequest(
provider=datamodel_pb2.Provider(
metadata=datamodel_pb2.ObjectMeta(name=name),
type=provider_type,
credentials=credentials,
)
)
)
try:
yield name
finally:
_delete_provider(stub, name)
def _delete_provider(stub: object, name: str) -> None:
"""Delete a provider, ignoring not-found errors."""
try:
stub.DeleteProvider(openshell_pb2.DeleteProviderRequest(name=name))
except grpc.RpcError as exc:
if hasattr(exc, "code") and exc.code() == grpc.StatusCode.NOT_FOUND:
pass
else:
raise
# ===========================================================================
# Tests: placeholder visibility
# ===========================================================================
def test_provider_credentials_available_as_env_vars(
sandbox: Callable[..., Sandbox],
sandbox_client: SandboxClient,
) -> None:
"""Sandbox child processes see provider env vars as placeholders."""
with provider(
sandbox_client._stub,
name="e2e-test-provider-env",
provider_type="claude",
credentials={"ANTHROPIC_API_KEY": "sk-e2e-test-key-12345"},
) as provider_name:
spec = datamodel_pb2.SandboxSpec(
policy=_default_policy(),
providers=[provider_name],
)
def read_env_var() -> str:
import os
return os.environ.get("ANTHROPIC_API_KEY", "NOT_SET")
with sandbox(spec=spec, delete_on_exit=True) as sb:
result = sb.exec_python(read_env_var)
assert result.exit_code == 0, result.stderr
value = result.stdout.strip()
assert value == "openshell:resolve:env:ANTHROPIC_API_KEY"
assert value != "sk-e2e-test-key-12345"
def test_generic_provider_credentials_available_as_env_vars(
sandbox: Callable[..., Sandbox],
sandbox_client: SandboxClient,
) -> None:
"""Generic provider env vars are placeholders, not raw secrets."""
with provider(
sandbox_client._stub,
name="e2e-test-generic-provider-env",
provider_type="generic",
credentials={
"CUSTOM_SERVICE_TOKEN": "token-generic-123",
"CUSTOM_SERVICE_URL": "https://internal.example.test/api",
},
) as provider_name:
spec = datamodel_pb2.SandboxSpec(
policy=_default_policy(),
providers=[provider_name],
)
def read_generic_env_vars() -> str:
import os
token = os.environ.get("CUSTOM_SERVICE_TOKEN", "NOT_SET")
url = os.environ.get("CUSTOM_SERVICE_URL", "NOT_SET")
return f"{token}|{url}"
with sandbox(spec=spec, delete_on_exit=True) as sb:
result = sb.exec_python(read_generic_env_vars)
assert result.exit_code == 0, result.stderr
assert (
result.stdout.strip()
== "openshell:resolve:env:CUSTOM_SERVICE_TOKEN|openshell:resolve:env:CUSTOM_SERVICE_URL"
)
def test_nvidia_provider_injects_nvidia_api_key_env_var(
sandbox: Callable[..., Sandbox],
sandbox_client: SandboxClient,
) -> None:
"""NVIDIA provider projects a placeholder env value into child processes."""
with provider(
sandbox_client._stub,
name="e2e-test-nvidia-provider-env",
provider_type="nvidia",
credentials={"NVIDIA_API_KEY": "nvapi-e2e-test-key"},
) as provider_name:
spec = datamodel_pb2.SandboxSpec(
policy=_default_policy(),
providers=[provider_name],
)
def read_nvidia_key() -> str:
import os
return os.environ.get("NVIDIA_API_KEY", "NOT_SET")
with sandbox(spec=spec, delete_on_exit=True) as sb:
result = sb.exec_python(read_nvidia_key)
assert result.exit_code == 0, result.stderr
assert result.stdout.strip() == "openshell:resolve:env:NVIDIA_API_KEY"
# ===========================================================================
# Tests: security & edge cases
# ===========================================================================
def test_ssh_handshake_secret_not_visible_in_exec_environment(
sandbox: Callable[..., Sandbox],
) -> None:
def read_handshake_secret() -> str:
import os
return os.environ.get("OPENSHELL_SSH_HANDSHAKE_SECRET", "NOT_SET")
with sandbox(delete_on_exit=True) as sb:
result = sb.exec_python(read_handshake_secret)
assert result.exit_code == 0, result.stderr
assert result.stdout.strip() == "NOT_SET"
def test_create_sandbox_rejects_unknown_provider(
sandbox_client: SandboxClient,
) -> None:
"""CreateSandbox fails fast when a provider name does not exist."""
spec = datamodel_pb2.SandboxSpec(
policy=_default_policy(),
providers=["nonexistent-provider-xyz"],
)
with pytest.raises(grpc.RpcError) as exc_info:
sandbox_client.create(spec=spec)
assert exc_info.value.code() == grpc.StatusCode.FAILED_PRECONDITION
assert "nonexistent-provider-xyz" in (exc_info.value.details() or "")
def test_credentials_not_in_persisted_spec_environment(
sandbox: Callable[..., Sandbox],
sandbox_client: SandboxClient,
) -> None:
"""Provider credentials should NOT appear in the sandbox spec's environment map."""
with provider(
sandbox_client._stub,
name="e2e-test-no-persist",
provider_type="claude",
credentials={"ANTHROPIC_API_KEY": "sk-should-not-persist"},
) as provider_name:
spec = datamodel_pb2.SandboxSpec(
policy=_default_policy(),
providers=[provider_name],
)
with sandbox(spec=spec, delete_on_exit=True) as sb:
fetched = sandbox_client._stub.GetSandbox(
openshell_pb2.GetSandboxRequest(name=sb.sandbox.name)
)
persisted_env = dict(fetched.sandbox.spec.environment)
assert "ANTHROPIC_API_KEY" not in persisted_env, (
"credentials should not be persisted in sandbox spec environment"
)
# ===========================================================================
# Tests: provider update merge semantics
# ===========================================================================
def test_update_provider_preserves_unset_credentials_and_config(
sandbox_client: SandboxClient,
) -> None:
"""Updating one credential must not clobber other credentials or config."""
stub = sandbox_client._stub
name = "merge-test-preserve"
_delete_provider(stub, name)
try:
stub.CreateProvider(
openshell_pb2.CreateProviderRequest(
provider=datamodel_pb2.Provider(
metadata=datamodel_pb2.ObjectMeta(name=name),
type="generic",
credentials={"KEY_A": "val-a", "KEY_B": "val-b"},
config={"BASE_URL": "https://example.com"},
)
)
)
stub.UpdateProvider(
openshell_pb2.UpdateProviderRequest(
provider=datamodel_pb2.Provider(
metadata=datamodel_pb2.ObjectMeta(name=name),
type="",
credentials={"KEY_A": "rotated-a"},
)
)
)
got = stub.GetProvider(openshell_pb2.GetProviderRequest(name=name))
p = got.provider
# Credential keys are preserved but values are redacted.
assert len(p.credentials) > 0, "credential keys should be preserved"
for key, val in p.credentials.items():
assert val == "REDACTED", (
f"credential '{key}' should be REDACTED, got '{val}'"
)
assert p.config["BASE_URL"] == "https://example.com", (
"config should be preserved"
)
finally:
_delete_provider(stub, name)
def test_update_provider_empty_maps_preserves_all(
sandbox_client: SandboxClient,
) -> None:
"""Sending empty credential and config maps should be a no-op."""
stub = sandbox_client._stub
name = "merge-test-noop"
_delete_provider(stub, name)
try:
stub.CreateProvider(
openshell_pb2.CreateProviderRequest(
provider=datamodel_pb2.Provider(
metadata=datamodel_pb2.ObjectMeta(name=name),
type="generic",
credentials={"TOKEN": "secret"},
config={"URL": "https://api.example.com"},
)
)
)
stub.UpdateProvider(
openshell_pb2.UpdateProviderRequest(
provider=datamodel_pb2.Provider(
metadata=datamodel_pb2.ObjectMeta(name=name),
type="",
)
)
)
got = stub.GetProvider(openshell_pb2.GetProviderRequest(name=name))
p = got.provider
# Credential keys are preserved but values are redacted.
assert len(p.credentials) > 0, "credential keys should be preserved"
for key, val in p.credentials.items():
assert val == "REDACTED", (
f"credential '{key}' should be REDACTED, got '{val}'"
)
assert p.config["URL"] == "https://api.example.com"
finally:
_delete_provider(stub, name)
def test_update_provider_merges_config_preserves_credentials(
sandbox_client: SandboxClient,
) -> None:
"""Updating only config should not touch credentials."""
stub = sandbox_client._stub
name = "merge-test-config-only"
_delete_provider(stub, name)
try:
stub.CreateProvider(
openshell_pb2.CreateProviderRequest(
provider=datamodel_pb2.Provider(
metadata=datamodel_pb2.ObjectMeta(name=name),
type="generic",
credentials={"API_KEY": "original-key"},
config={"ENDPOINT": "https://old.example.com"},
)
)
)
stub.UpdateProvider(
openshell_pb2.UpdateProviderRequest(
provider=datamodel_pb2.Provider(
metadata=datamodel_pb2.ObjectMeta(name=name),
type="",
config={"ENDPOINT": "https://new.example.com"},
)
)
)
got = stub.GetProvider(openshell_pb2.GetProviderRequest(name=name))
p = got.provider
# Credential keys are preserved but values are redacted.
assert len(p.credentials) > 0, "credential keys should be preserved"
for key, val in p.credentials.items():
assert val == "REDACTED", (
f"credential '{key}' should be REDACTED, got '{val}'"
)
assert p.config["ENDPOINT"] == "https://new.example.com"
finally:
_delete_provider(stub, name)
def test_update_provider_rejects_type_change(
sandbox_client: SandboxClient,
) -> None:
"""Attempting to change a provider's type must be rejected."""
stub = sandbox_client._stub
name = "merge-test-type-reject"
_delete_provider(stub, name)
try:
stub.CreateProvider(
openshell_pb2.CreateProviderRequest(
provider=datamodel_pb2.Provider(
metadata=datamodel_pb2.ObjectMeta(name=name),
type="generic",
credentials={"KEY": "val"},
)
)
)
with pytest.raises(grpc.RpcError) as exc_info:
stub.UpdateProvider(
openshell_pb2.UpdateProviderRequest(
provider=datamodel_pb2.Provider(
metadata=datamodel_pb2.ObjectMeta(name=name),
type="nvidia",
)
)
)
assert exc_info.value.code() == grpc.StatusCode.INVALID_ARGUMENT
assert "type cannot be changed" in exc_info.value.details()
finally:
_delete_provider(stub, name)
Reference in New Issue View Git Blame Copy Permalink