Client¶

client ¶

OpenShell gRPC client wrapper used throughout ShoreGuard.

Provides :class:ShoreGuardClient, the single object every service layer uses to talk to an OpenShell gateway. Manages the gRPC channel (plaintext or mTLS), builds the stubs once at construction time, and exposes four submanagers — :class:ApprovalManager, :class:PolicyManager, :class:ProviderManager, :class:SandboxManager — each of which maps a logical surface onto a subset of the underlying RPCs.

The client is intentionally synchronous: gRPC itself is sync, and all the service layers wrap calls in asyncio.to_thread when they need to run inside a FastAPI async route. Keeping the client sync avoids the double-wrapping that an async gRPC client would require without any benefit at this scale.

ShoreGuardClient ¶

ShoreGuardClient(
    endpoint: str,
    *,
    ca_path: Path | None = None,
    cert_path: Path | None = None,
    key_path: Path | None = None,
    timeout: float = 30.0,
    retry_policy: RetryPolicy | None = None,
    retry_deadline: float | None = None,
    require_mtls: bool = False,
)

Unified gRPC client bound to a single OpenShell gateway.

Holds the channel, the protobuf stubs, and the four submanagers (approvals, policies, providers, sandboxes) that expose OpenShell's RPC surface. One instance maps to one gateway endpoint — do not reuse across gateways; instead, construct a new client per (endpoint, credentials) pair.

The channel is opened lazily on first use so construction is cheap and a bad cert bundle only fails the first call rather than the registry read that created the client.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	gRPC endpoint address as `host:port`.	required
`ca_path`	`Path \| None`	Path to the CA certificate for TLS. `None` for plaintext channels (permitted only in local mode).	`None`
`cert_path`	`Path \| None`	Path to the client certificate for mTLS. `None` for server-only TLS.	`None`
`key_path`	`Path \| None`	Path to the client private key for mTLS. `None` for server-only TLS.	`None`
`timeout`	`float`	Default gRPC call timeout in seconds.	`30.0`
`retry_policy`	`RetryPolicy \| None`	Optional retry policy; defaults to :func:`_default_retry_policy`.	`None`
`retry_deadline`	`float \| None`	Optional total retry budget in seconds; defaults to :func:`_default_retry_deadline`.	`None`
`require_mtls`	`bool`	Reject plaintext channels when `True`. The direct constructor defaults to `False` for local dev; the `from_credentials` factory defaults to the configured `GatewaySettings.require_mtls`.	`False`

Attributes:

Name	Type	Description
`cert_info`	`CertInfo \| None`	Parsed metadata from the most recent bundle validation, or `None` for plaintext channels.

Raises:

Type	Description
`GatewayNotConnectedError`	If `require_mtls` is set but no client bundle is provided, or if eager bundle validation fails.

cert_info `property` ¶

cert_info: CertInfo | None

Return the parsed client-cert metadata for the current channel.

Returns:

Type	Description
`CertInfo \| None`	CertInfo \| None: Metadata from the most recent successful bundle validation, or `None` for plaintext channels.

from_credentials `classmethod` ¶

from_credentials(
    endpoint: str,
    *,
    ca_cert: bytes | None = None,
    client_cert: bytes | None = None,
    client_key: bytes | None = None,
    timeout: float = 30.0,
    retry_policy: RetryPolicy | None = None,
    retry_deadline: float | None = None,
    require_mtls: bool | None = None,
) -> ShoreGuardClient

Connect using raw certificate bytes (from DB or registry).

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	gRPC endpoint address (host:port).	required
`ca_cert`	`bytes \| None`	CA certificate bytes for TLS.	`None`
`client_cert`	`bytes \| None`	Client certificate bytes for mTLS.	`None`
`client_key`	`bytes \| None`	Client private key bytes for mTLS.	`None`
`timeout`	`float`	Default gRPC call timeout in seconds.	`30.0`
`retry_policy`	`RetryPolicy \| None`	Optional retry policy; defaults to :func:`_default_retry_policy`.	`None`
`retry_deadline`	`float \| None`	Optional total retry budget in seconds; defaults to :func:`_default_retry_deadline`.	`None`
`require_mtls`	`bool \| None`	Enforce mTLS. `None` (default) reads the setting from :class:`GatewaySettings`.	`None`

Returns:

Name	Type	Description
`ShoreGuardClient`	`ShoreGuardClient`	Connected client instance.

Raises:

Type	Description
`GatewayNotConnectedError`	If mTLS is required but the bundle is missing or fails eager validation.

from_active_cluster `classmethod` ¶

from_active_cluster(
    *, cluster: str | None = None, timeout: float = 30.0
) -> ShoreGuardClient

Connect to the active OpenShell gateway using mTLS credentials.

Parameters:

Name	Type	Description	Default
`cluster`	`str \| None`	Cluster name override. Defaults to the active gateway.	`None`
`timeout`	`float`	Default gRPC call timeout in seconds.	`30.0`

Returns:

Name	Type	Description
`ShoreGuardClient`	`ShoreGuardClient`	Connected client instance.

Raises:

Type	Description
`GatewayNotConnectedError`	If gateway metadata is missing or invalid.

health ¶

health() -> dict

Check gateway health.

Returns:

Name	Type	Description
`dict`	`dict`	Status and version of the gateway.

get_inference_bundle ¶

get_inference_bundle() -> dict

Get the resolved inference bundle (routes after policy overlay).

API keys are redacted: each route exposes only has_api_key (bool), never the secret value.

Returns:

Name	Type	Description
`dict`	`dict`	`{revision, generated_at_ms, routes: [...]}` where each route contains name, base_url, protocols, model_id, provider_type, timeout_secs, has_api_key.

get_cluster_inference ¶

get_cluster_inference(*, route_name: str = '') -> dict

Get current cluster inference configuration.

Parameters:

Name	Type	Description	Default
`route_name`	`str`	Optional route name to filter by.	`''`

Returns:

Name	Type	Description
`dict`	`dict`	Inference configuration with provider, model, and route.

get_gateway_config ¶

get_gateway_config() -> dict

Get the global gateway configuration (settings and revision).

Returns:

Name	Type	Description
`dict`	`dict`	Settings map and settings revision number.

update_gateway_setting ¶

update_gateway_setting(
    *,
    key: str,
    value: str | bool | int | None = None,
    delete: bool = False,
) -> dict

Update (or delete) a single global gateway setting.

Parameters:

Name	Type	Description	Default
`key`	`str`	Setting key name.	required
`value`	`str \| bool \| int \| None`	New value. Type determines the `SettingValue` oneof field (`bool` before `int` since `bool` is a subclass of `int`). Ignored when `delete` is True.	`None`
`delete`	`bool`	If True, remove the setting instead of updating it.	`False`

Returns:

Name	Type	Description
`dict`	`dict`	`{"settings_revision": int, "deleted": bool}`.

Raises:

Type	Description
`TypeError`	If `value` has an unsupported Python type.

set_cluster_inference ¶

set_cluster_inference(
    *,
    provider_name: str,
    model_id: str,
    verify: bool = True,
    route_name: str = "",
    timeout_secs: int = 0,
) -> dict

Set cluster inference configuration.

Parameters:

Name	Type	Description	Default
`provider_name`	`str`	Name of the inference provider.	required
`model_id`	`str`	Model identifier to use.	required
`verify`	`bool`	Whether to validate endpoints before saving.	`True`
`route_name`	`str`	Optional route name to configure.	`''`
`timeout_secs`	`int`	Per-route request timeout in seconds (0 = default 60s).	`0`

Returns:

Name	Type	Description
`dict`	`dict`	Updated inference configuration with validation results.

reload_credentials ¶

reload_credentials(
    *, ca_cert: bytes, client_cert: bytes, client_key: bytes
) -> None

Rotate the mTLS bundle by rebuilding the channel and sub-managers.

Validates the new bundle eagerly (via :func:validate_bundle, which raises :class:GatewayNotConnectedError on failure), closes the existing channel, and rebuilds stubs plus every manager so subsequent calls run over the fresh credentials. Callers must serialize invocations: in-flight streams held by other callers will observe a ChannelClosed error and should reconnect on their own iteration.

Parameters:

Name	Type	Description	Default
`ca_cert`	`bytes`	New CA certificate bytes.	required
`client_cert`	`bytes`	New client certificate bytes.	required
`client_key`	`bytes`	New client private key bytes.	required

close ¶

close() -> None

Close the underlying gRPC channel.

enter ¶

__enter__() -> ShoreGuardClient

Support usage as a context manager.

Returns:

Name	Type	Description
`ShoreGuardClient`	`ShoreGuardClient`	This client instance.

exit ¶

__exit__(*args: object) -> None

Close the channel on context exit.

Parameters:

Name	Type	Description	Default
`*args`	`object`	Exception info (exc_type, exc_val, exc_tb).	`()`

sandboxes ¶

gRPC wrapper for OpenShell's sandbox lifecycle and exec RPCs.

Covers create / list / get / delete plus ExecSandbox — both regular and TTY variants — and the SSH session open / revoke pair. Sandbox metadata (labels, description) kept on the ShoreGuard side is not handled here; callers merge that in at the service layer.

ExecSandbox is also the delegation target for the post-create boot hooks flow: the hook service reuses this manager to run warm-up commands inside a sandbox once it has been created.

SandboxManager ¶

SandboxManager(
    stub: OpenShellStub,
    *,
    timeout: float = 30.0,
    retry_policy: RetryPolicy | None = None,
    retry_deadline: float | None = 60.0,
)

Sandbox CRUD, execution, and lifecycle operations.

Parameters:

Name	Type	Description	Default
`stub`	`OpenShellStub`	OpenShell gRPC stub.	required
`timeout`	`float`	gRPC call timeout in seconds.	`30.0`
`retry_policy`	`RetryPolicy \| None`	Retry/backoff policy applied to every unary RPC and to stream-open calls. Defaults to :data:`DEFAULT_POLICY`.	`None`
`retry_deadline`	`float \| None`	Total wall-clock budget in seconds including retries. `None` lets `max_attempts` be the only limiter.	`60.0`

list ¶

list(
    *, limit: int = 100, offset: int = 0
) -> list[dict[str, Any]]

List all sandboxes.

Parameters:

Name	Type	Description	Default
`limit`	`int`	Maximum number of sandboxes to return.	`100`
`offset`	`int`	Pagination offset.	`0`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: List of sandbox dicts.

get ¶

get(name: str) -> dict[str, Any]

Get a sandbox by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Sandbox name.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Sandbox data dict.

create ¶

create(
    *,
    name: str = "",
    image: str = "",
    policy: dict | None = None,
    providers: list[str] | None = None,
    gpu: bool = False,
    environment: dict[str, str] | None = None,
    log_level: str = "",
) -> dict[str, Any]

Create a new sandbox.

Parameters:

Name	Type	Description	Default
`name`	`str`	Sandbox name.	`''`
`image`	`str`	Container image to use.	`''`
`policy`	`dict \| None`	Optional sandbox policy dict.	`None`
`providers`	`list[str] \| None`	Optional list of provider names.	`None`
`gpu`	`bool`	Whether to request GPU resources.	`False`
`environment`	`dict[str, str] \| None`	Optional environment variable key-value pairs.	`None`
`log_level`	`str`	Sandbox-supervisor log verbosity. Empty string (default) means the gateway's default level. Accepted values upstream: `debug`, `info`, `warn`, `error`.	`''`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Created sandbox data dict.

delete ¶

delete(name: str) -> bool

Delete a sandbox by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Sandbox name.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if the sandbox was deleted.

list_providers ¶

list_providers(sandbox_name: str) -> list[dict[str, Any]]

List provider records attached to a sandbox.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name (canonical lookup key).	required

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: Attached provider records, each as the flat dict returned by :func:`_provider_to_dict`.

attach_provider ¶

attach_provider(
    sandbox_name: str, provider_name: str
) -> dict[str, Any]

Attach a provider record to a sandbox.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name (canonical lookup key).	required
`provider_name`	`str`	Provider name to attach.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: `{sandbox: <sandbox dict>, attached: bool}` where `attached` is False if the provider was already attached.

detach_provider ¶

detach_provider(
    sandbox_name: str, provider_name: str
) -> dict[str, Any]

Detach a provider record from a sandbox.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name (canonical lookup key).	required
`provider_name`	`str`	Provider name to detach.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: `{sandbox: <sandbox dict>, detached: bool}` where `detached` is False if the provider was not attached.

wait_ready ¶

wait_ready(
    name: str, *, timeout_seconds: float = 300.0
) -> dict[str, Any]

Block until a sandbox reaches READY phase.

Parameters:

Name	Type	Description	Default
`name`	`str`	Sandbox name.	required
`timeout_seconds`	`float`	Maximum time to wait in seconds.	`300.0`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Sandbox data dict once ready.

Raises:

Type	Description
`SandboxError`	If the sandbox enters an error phase.
`TimeoutError`	If the sandbox is not ready within the timeout.

get_config ¶

get_config(sandbox_id: str) -> dict[str, Any]

Fetch the effective sandbox config (policy + settings + revision).

Parameters:

Name	Type	Description	Default
`sandbox_id`	`str`	Sandbox identifier (proto field accepts the name).	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: `{policy, version, policy_hash, settings, config_revision, policy_source, global_policy_version}`. `settings` is a flat `{key: {value, scope}}` map.

get_provider_environment ¶

get_provider_environment(sandbox_id: str) -> dict[str, str]

Fetch the resolved provider environment variables for a sandbox.

Parameters:

Name	Type	Description	Default
`sandbox_id`	`str`	Sandbox identifier (proto field accepts the name).	required

Returns:

Type	Description
`dict[str, str]`	dict[str, str]: Environment variables map.

exec ¶

exec(
    sandbox_id: str,
    command: list[str],
    *,
    workdir: str = "",
    env: dict[str, str] | None = None,
    timeout_seconds: int = 0,
    tty: bool = False,
) -> dict[str, Any]

Execute a command in a sandbox and return the result.

Parameters:

Name	Type	Description	Default
`sandbox_id`	`str`	Sandbox identifier.	required
`command`	`list[str]`	Command and arguments to execute.	required
`workdir`	`str`	Working directory inside the sandbox.	`''`
`env`	`dict[str, str] \| None`	Optional environment variables for the command.	`None`
`timeout_seconds`	`int`	Command timeout in seconds (0 for default).	`0`
`tty`	`bool`	Allocate a TTY for the command (for interactive programs that detect `isatty()`). Added in OpenShell v0.0.23.	`False`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Execution result with exit_code, stdout, and stderr.

exec_interactive ¶

exec_interactive(
    sandbox_id: str,
    command: list[str],
    *,
    inbound: Queue[dict[str, Any] | None],
    on_call: Callable[[Any], None] | None = None,
    workdir: str = "",
    env: dict[str, str] | None = None,
    cols: int = 80,
    rows: int = 24,
    timeout: float | None = None,
) -> Iterator[dict[str, Any]]

Run a command with bidirectional streaming (interactive TTY session).

The first wire message carries the exec request; subsequent messages are drained from inbound — {"type": "stdin", "data": bytes} or {"type": "resize", "cols": int, "rows": int} — until a None sentinel half-closes the request stream. Output events are yielded as dicts. ExecSandboxInteractive always allocates a TTY.

Parameters:

Name	Type	Description	Default
`sandbox_id`	`str`	Sandbox identifier.	required
`command`	`list[str]`	Command and arguments to execute.	required
`inbound`	`Queue[dict[str, Any] \| None]`	Thread-safe queue of stdin/resize frames; `None` ends input.	required
`on_call`	`Callable[[Any], None] \| None`	Optional callback handed the live gRPC call so the caller can cancel it on teardown.	`None`
`workdir`	`str`	Working directory inside the sandbox.	`''`
`env`	`dict[str, str] \| None`	Optional environment variables for the command.	`None`
`cols`	`int`	Initial terminal columns.	`80`
`rows`	`int`	Initial terminal rows.	`24`
`timeout`	`float \| None`	Optional gRPC deadline in seconds; `None` for no deadline.	`None`

Yields:

Type	Description
`dict[str, Any]`	dict[str, Any]: `{"type": "stdout"\|"stderr", "data": bytes}` or `{"type": "exit", "exit_code": int}`.

forward_tcp ¶

forward_tcp(
    *,
    init: dict[str, Any],
    inbound: Queue[bytes | None],
    on_call: Callable[[Any], None] | None = None,
    timeout: float | None = None,
) -> Iterator[dict[str, Any]]

Relay a raw TCP/SSH tunnel to a sandbox via bidirectional streaming.

The first wire frame carries the relay target (ssh or tcp); subsequent frames carry raw bytes from inbound until a None sentinel. Inbound gateway bytes are yielded as {"type": "data", "data": bytes}.

Parameters:

Name	Type	Description	Default
`init`	`dict[str, Any]`	Relay-init mapping (see :func:`_build_tcp_forward_init`).	required
`inbound`	`Queue[bytes \| None]`	Thread-safe queue of outbound byte chunks; `None` ends it.	required
`on_call`	`Callable[[Any], None] \| None`	Optional callback handed the live gRPC call for cancellation.	`None`
`timeout`	`float \| None`	Optional gRPC deadline in seconds; `None` for no deadline.	`None`

Yields:

Type	Description
`dict[str, Any]`	dict[str, Any]: `{"type": "data", "data": bytes}` per inbound chunk.

issue_token ¶

issue_token() -> dict[str, Any]

Mint a gateway JWT bound to the calling mTLS identity.

Note

The request is empty — the gateway binds the token to the caller's identity. Called from ShoreGuard this yields a token for ShoreGuard's own gateway identity (a diagnostic), not one scoped to a sandbox.

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: `{"token": str, "expires_at_ms": int}` (0 = non-expiring).

refresh_token ¶

refresh_token() -> dict[str, Any]

Mint a fresh gateway JWT for the calling identity.

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: `{"token": str, "expires_at_ms": int}` (0 = non-expiring).

create_ssh_session ¶

create_ssh_session(sandbox_id: str) -> dict[str, Any]

Create a temporary SSH session for shell access to a sandbox.

The upstream gateway holds a documented charset contract on every response field (see CreateSshSessionResponse in openshell.proto, tightened in NVIDIA/OpenShell#876 <https://github.com/NVIDIA/OpenShell/pull/876>_). The fields flow into an SSH ProxyCommand string executed through /bin/sh -c on the caller's workstation, so anything outside the contract is a ProxyCommand-injection vector. ShoreGuard enforces the same contract client-side as defence-in-depth: a compromised or misconfigured gateway cannot push shell metacharacters into ShoreGuard's REST response.

Parameters:

Name	Type	Description	Default
`sandbox_id`	`str`	Sandbox identifier.	required

:func:_validate_ssh_session_response raises :class:SandboxError before the response can escape if any field is outside the documented charset or numeric range.

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: SSH session details including token and gateway connection info.

revoke_ssh_session ¶

revoke_ssh_session(token: str) -> bool

Revoke an active SSH session.

Parameters:

Name	Type	Description	Default
`token`	`str`	Session token to revoke.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if the session was revoked.

get_logs ¶

get_logs(
    sandbox_id: str,
    *,
    lines: int = 200,
    since_ms: int = 0,
    sources: list[str] | None = None,
    min_level: str = "",
) -> list[dict[str, Any]]

Fetch recent sandbox logs.

Parameters:

Name	Type	Description	Default
`sandbox_id`	`str`	Sandbox identifier.	required
`lines`	`int`	Maximum number of log lines to return.	`200`
`since_ms`	`int`	Only return logs after this timestamp (ms).	`0`
`sources`	`list[str] \| None`	Optional list of log sources to filter by.	`None`
`min_level`	`str`	Minimum log level to include.	`''`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: List of log entry dicts.

watch ¶

watch(
    sandbox_id: str,
    *,
    follow_status: bool = True,
    follow_logs: bool = True,
    follow_events: bool = True,
    log_tail_lines: int = 50,
) -> Iterator[dict[str, Any]]

Stream live sandbox events (status, logs, platform events, draft updates).

Parameters:

Name	Type	Description	Default
`sandbox_id`	`str`	Sandbox identifier.	required
`follow_status`	`bool`	Subscribe to status changes.	`True`
`follow_logs`	`bool`	Subscribe to log output.	`True`
`follow_events`	`bool`	Subscribe to platform events.	`True`
`log_tail_lines`	`int`	Number of existing log lines to replay.	`50`

Yields:

Type	Description
`dict[str, Any]`	dict[str, Any]: Event dict with `type` and `data` keys.

policies ¶

gRPC wrapper for OpenShell's sandbox policy RPCs.

Exposes read and write operations against a sandbox's policy via GetSandbox / UpdateConfig / ListPolicyRevisions / DiffPolicyRevisions. Converts protobuf PolicyChunk / NetworkRule / FilesystemPath / ProcessPolicy messages to plain dicts through the shared converters so every caller sees the same JSON-ready shape.

This manager is deliberately stateless: it holds the stub and nothing else. Atomic single-rule CRUD (read-modify-write over whole-policy updates) lives in :class:~shoreguard.services.policy.PolicyService, not here, because the read-modify-write loop needs to be aware of pinning, audit logging, and denial-context capture.

MergeOperationError ¶

Bases: ValueError

Invalid merge operation dict.

Raised when a merge-operation dict cannot be converted to its protobuf form (unknown type, missing required field).

PolicyManager ¶

PolicyManager(
    stub: OpenShellStub, *, timeout: float = 30.0
)

Policy read/write operations against OpenShell gateway.

Parameters:

Name	Type	Description	Default
`stub`	`OpenShellStub`	OpenShell gRPC stub.	required
`timeout`	`float`	gRPC call timeout in seconds.	`30.0`

get ¶

get(sandbox_name: str) -> dict[str, Any]

Get the current active policy for a sandbox.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Active policy status with revision details.

get_version ¶

get_version(
    sandbox_name: str, version: int
) -> dict[str, Any]

Get a specific policy revision by version number.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required
`version`	`int`	Policy version number to retrieve.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Policy status with revision details.

list_revisions ¶

list_revisions(
    sandbox_name: str, *, limit: int = 20, offset: int = 0
) -> list[dict[str, Any]]

List policy revision history for a sandbox.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required
`limit`	`int`	Maximum number of revisions to return.	`20`
`offset`	`int`	Pagination offset.	`0`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: List of policy revision summary dicts.

update ¶

update(
    sandbox_name: str,
    policy: SandboxPolicy,
    *,
    global_scope: bool = False,
) -> dict[str, Any]

Push a new policy version to a sandbox (or globally).

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required
`policy`	`SandboxPolicy`	SandboxPolicy protobuf message.	required
`global_scope`	`bool`	If True, apply policy globally.	`False`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Version and policy hash of the new revision.

apply_merge_operations ¶

apply_merge_operations(
    sandbox_name: str, operations: list[dict[str, Any]]
) -> dict[str, Any]

Apply incremental policy merge operations against a sandbox.

Maps to UpdateConfigRequest.merge_operations (upstream PR #860, NVIDIA/OpenShell#860 <https://github.com/NVIDIA/OpenShell/pull/860>_). Each dict in operations carries a type discriminator and op-specific fields; see :func:_dict_to_merge_operation for the accepted shapes.

Operations are applied in order by the gateway. Callers constructing deltas from a Git source of truth (M23 GitOps) should emit remove_* operations before add_* operations so a partial failure cannot leave the gateway in a state that is inconsistent with either the source or the target revision.

This path is sandbox-scoped only — the upstream proto does not support a global merge. Callers that need global replacement must use :meth:update instead.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox whose policy is being mutated.	required
`operations`	`list[dict[str, Any]]`	Ordered list of merge-op dicts. Empty list is a no-op (the gateway still assigns a new revision).	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: `version` and `policy_hash` of the new revision, shaped identically to :meth:`update` so downstream code can treat the two as interchangeable.

submit_analysis ¶

submit_analysis(
    sandbox_name: str,
    *,
    summaries: list[dict[str, Any]],
    proposed_chunks: list[dict[str, Any]],
    analysis_mode: str = "",
) -> dict[str, Any]

Submit denial-analysis results and proposed chunks to the gateway.

The gateway merges accepted chunks into the draft policy and rejects the rest with a per-chunk reason. Used by external analyzers (LLM-backed or rule-based) that observe sandbox denials and propose policy fixes.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Target sandbox name — goes into `request.name`.	required
`summaries`	`list[dict[str, Any]]`	`DenialSummary` dicts. Unknown keys raise `TypeError` from the proto constructor — the caller is responsible for sending only fields that the currently pinned OpenShell proto defines.	required
`proposed_chunks`	`list[dict[str, Any]]`	`PolicyChunk` dicts with the rules that would fix the denials described in summaries.	required
`analysis_mode`	`str`	Optional mode tag forwarded verbatim, e.g. `"auto"` or `"manual"`.	`''`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: ``{"accepted_chunks": int, "rejected_chunks": int,
`dict[str, Any]`	"rejection_reasons": list[str]}``.

approvals ¶

gRPC wrapper for OpenShell's draft policy approval RPCs.

Owns the thin mapping from ShoreGuardClient.approvals method calls to the underlying SubmitPolicyAnalysis / GetDraftPolicy / ApproveChunk / RejectChunk gRPC methods, translating protobuf messages into plain dicts via the shared converters. No business logic lives here — quorum, pinning, webhook fan-out, and audit logging all happen at the service layer above.

ApprovalManager ¶

ApprovalManager(
    stub: OpenShellStub, *, timeout: float = 30.0
)

Draft policy approval flow: review, approve, reject blocked requests.

Parameters:

Name	Type	Description	Default
`stub`	`OpenShellStub`	OpenShell gRPC stub.	required
`timeout`	`float`	gRPC call timeout in seconds.	`30.0`

get_draft ¶

get_draft(
    sandbox_name: str, *, status_filter: str = ""
) -> dict[str, Any]

Get draft policy recommendations for a sandbox.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required
`status_filter`	`str`	Optional status to filter chunks by.	`''`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Draft policy with chunks, summary, and version.

get_pending ¶

get_pending(sandbox_name: str) -> list[dict[str, Any]]

Get only pending (unapproved) draft chunks.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: List of pending chunk dicts.

approve ¶

approve(sandbox_name: str, chunk_id: str) -> dict[str, Any]

Approve a single draft policy chunk (merges into active policy).

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required
`chunk_id`	`str`	Chunk identifier to approve.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: New policy version and hash.

reject ¶

reject(
    sandbox_name: str, chunk_id: str, *, reason: str = ""
) -> None

Reject a single draft policy chunk.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required
`chunk_id`	`str`	Chunk identifier to reject.	required
`reason`	`str`	Optional rejection reason.	`''`

approve_all ¶

approve_all(
    sandbox_name: str,
    *,
    include_security_flagged: bool = False,
) -> dict[str, Any]

Approve all pending draft chunks.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required
`include_security_flagged`	`bool`	Whether to also approve security-flagged chunks.	`False`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Policy version, hash, and approval counts.

edit ¶

edit(
    sandbox_name: str, chunk_id: str, proposed_rule: dict
) -> None

Edit a pending draft chunk in-place.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required
`chunk_id`	`str`	Chunk identifier to edit.	required
`proposed_rule`	`dict`	Network rule dict to replace the existing rule.	required

undo ¶

undo(sandbox_name: str, chunk_id: str) -> dict[str, Any]

Reverse an approval (remove merged rule from active policy).

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required
`chunk_id`	`str`	Chunk identifier to undo.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated policy version and hash.

clear ¶

clear(sandbox_name: str) -> dict[str, int]

Clear all pending draft chunks for a sandbox.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required

Returns:

Type	Description
`dict[str, int]`	dict[str, int]: Number of chunks cleared.

get_history ¶

get_history(sandbox_name: str) -> list[dict[str, Any]]

Get decision history for a sandbox's draft policy.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Sandbox name.	required

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: List of history entry dicts.

providers ¶

gRPC wrapper for OpenShell's provider RPCs.

Exposes list / get / create / update / delete against the upstream provider surface plus the env-projection call used to render a redacted view of a provider's environment variables. Credentials flow only into this manager via create and update; reads never return raw secrets — the wrapping service layer relies on that invariant when it renders the [REDACTED] env projection.

ProviderManager ¶

ProviderManager(
    stub: OpenShellStub, *, timeout: float = 30.0
)

Provider CRUD operations against OpenShell gateway.

Parameters:

Name	Type	Description	Default
`stub`	`OpenShellStub`	OpenShell gRPC stub.	required
`timeout`	`float`	gRPC call timeout in seconds.	`30.0`

list ¶

list(
    *, limit: int = 100, offset: int = 0
) -> list[dict[str, Any]]

List all providers.

Parameters:

Name	Type	Description	Default
`limit`	`int`	Maximum number of providers to return.	`100`
`offset`	`int`	Pagination offset.	`0`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: List of provider dicts.

get ¶

get(name: str) -> dict[str, Any]

Get a provider by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Provider name.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Provider data dict.

create ¶

create(
    *,
    name: str,
    provider_type: str,
    credentials: dict[str, str] | None = None,
    config: dict[str, str] | None = None,
    labels: dict[str, str] | None = None,
) -> dict[str, Any]

Create a new provider.

Parameters:

Name	Type	Description	Default
`name`	`str`	Provider name.	required
`provider_type`	`str`	Provider type identifier.	required
`credentials`	`dict[str, str] \| None`	Provider credential key-value pairs.	`None`
`config`	`dict[str, str] \| None`	Provider configuration key-value pairs.	`None`
`labels`	`dict[str, str] \| None`	Optional Kubernetes-style labels.	`None`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Created provider data dict.

update ¶

update(
    *,
    name: str,
    provider_type: str = "",
    credentials: dict[str, str] | None = None,
    config: dict[str, str] | None = None,
    labels: dict[str, str] | None = None,
) -> dict[str, Any]

Update an existing provider.

Parameters:

Name	Type	Description	Default
`name`	`str`	Provider name.	required
`provider_type`	`str`	Provider type identifier.	`''`
`credentials`	`dict[str, str] \| None`	Provider credential key-value pairs.	`None`
`config`	`dict[str, str] \| None`	Provider configuration key-value pairs.	`None`
`labels`	`dict[str, str] \| None`	Optional Kubernetes-style labels.	`None`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated provider data dict.

delete ¶

delete(name: str) -> bool

Delete a provider by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Provider name.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if the provider was deleted.

configure_refresh ¶

configure_refresh(
    *,
    provider: str,
    credential_key: str,
    strategy: str,
    material: dict[str, str] | None = None,
    secret_material_keys: list[str] | None = None,
    expires_at_ms: int | None = None,
) -> dict[str, Any]

Configure automatic refresh for a provider credential.

Parameters:

Name	Type	Description	Default
`provider`	`str`	Provider name.	required
`credential_key`	`str`	Credential key within the provider to refresh.	required
`strategy`	`str`	Refresh strategy (e.g. `static`, `oauth2_refresh_token`).	required
`material`	`dict[str, str] \| None`	Strategy-specific material (token URLs, client ids, etc.).	`None`
`secret_material_keys`	`list[str] \| None`	Keys within material that hold secret values.	`None`
`expires_at_ms`	`int \| None`	Optional absolute expiry of the current credential.	`None`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: The resulting refresh status.

get_refresh_status ¶

get_refresh_status(
    provider: str, *, credential_key: str = ""
) -> list[dict[str, Any]]

List credential-refresh status entries for a provider.

Parameters:

Name	Type	Description	Default
`provider`	`str`	Provider name.	required
`credential_key`	`str`	Optional credential key; empty returns all entries.	`''`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: One status dict per configured credential.

rotate_credential ¶

rotate_credential(
    *, provider: str, credential_key: str
) -> dict[str, Any]

Rotate a provider credential immediately.

Parameters:

Name	Type	Description	Default
`provider`	`str`	Provider name.	required
`credential_key`	`str`	Credential key to rotate.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: The refresh status after rotation.

delete_refresh ¶

delete_refresh(
    *, provider: str, credential_key: str
) -> bool

Delete a credential-refresh configuration.

Parameters:

Name	Type	Description	Default
`provider`	`str`	Provider name.	required
`credential_key`	`str`	Credential key whose refresh config is removed.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if a configuration existed and was deleted.

Client¶

client ¶

ShoreGuardClient ¶

cert_info property ¶

from_credentials classmethod ¶

from_active_cluster classmethod ¶

health ¶

get_inference_bundle ¶

get_cluster_inference ¶

get_gateway_config ¶

update_gateway_setting ¶

set_cluster_inference ¶

reload_credentials ¶

close ¶

__enter__ ¶

__exit__ ¶

sandboxes ¶

SandboxManager ¶

list ¶

get ¶

create ¶

delete ¶

list_providers ¶

attach_provider ¶

detach_provider ¶

wait_ready ¶

get_config ¶

get_provider_environment ¶

exec ¶

exec_interactive ¶

forward_tcp ¶

issue_token ¶

refresh_token ¶

create_ssh_session ¶

revoke_ssh_session ¶

get_logs ¶

watch ¶

policies ¶

MergeOperationError ¶

PolicyManager ¶

get ¶

get_version ¶

list_revisions ¶

update ¶

apply_merge_operations ¶

submit_analysis ¶

approvals ¶

ApprovalManager ¶

get_draft ¶

get_pending ¶

approve ¶

reject ¶

approve_all ¶

edit ¶

undo ¶

clear ¶

get_history ¶

providers ¶

ProviderManager ¶

list ¶

get ¶

create ¶

update ¶

delete ¶

configure_refresh ¶

get_refresh_status ¶

rotate_credential ¶

delete_refresh ¶

cert_info `property` ¶

from_credentials `classmethod` ¶

from_active_cluster `classmethod` ¶

enter ¶

exit ¶