Services¶

registry ¶

Gateway registry: persistent list of every known OpenShell endpoint.

Single source of truth for "which gateways does ShoreGuard know about". Backed by the gateways SQL table. Tracks connection material (endpoint, scheme, auth mode, mTLS cert bundle), per-gateway metadata the UI consumes (labels, free-text description), and the last-known health status maintained by the background health monitor.

The registry intentionally does not own live gRPC clients — those belong to :class:~shoreguard.services.gateway.GatewayService. Keeping CRUD separate from client lifecycle means a registry edit never has to worry about tearing down an in-flight call.

GatewayRegistry ¶

GatewayRegistry(session_factory: sessionmaker[Session])

Persistent CRUD and health tracking for registered gateways.

Handles insert / update / delete of gateway rows, label + metadata merging, and lazy health-status updates from the background health monitor. Does not manage live gRPC channels — those live in :class:~shoreguard.services.gateway.GatewayService — and does not enforce endpoint-format validation, which is applied at the API route layer before any call reaches this service.

Parameters:

Name	Type	Description	Default
`session_factory`	`sessionmaker[Session]`	SQLAlchemy session factory used to open a fresh session per call.	required

register ¶

register(
    name: str,
    endpoint: str,
    scheme: str = "https",
    auth_mode: str | None = "mtls",
    *,
    ca_cert: bytes | None = None,
    client_cert: bytes | None = None,
    client_key: bytes | None = None,
    metadata: dict[str, Any] | None = None,
    description: str | None = None,
    labels: dict[str, str] | None = None,
) -> dict[str, Any]

Register a new gateway.

Parameters:

Name	Type	Description	Default
`name`	`str`	Unique gateway name.	required
`endpoint`	`str`	Gateway endpoint address.	required
`scheme`	`str`	Connection scheme (e.g. "https").	`'https'`
`auth_mode`	`str \| None`	Authentication mode (e.g. "mtls").	`'mtls'`
`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`
`metadata`	`dict[str, Any] \| None`	Optional metadata dict.	`None`
`description`	`str \| None`	Optional free-text description.	`None`
`labels`	`dict[str, str] \| None`	Optional key-value labels for filtering.	`None`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: The registered gateway record.

Raises:

Type	Description
`ConflictError`	If a gateway with the given name already exists.

unregister ¶

unregister(name: str) -> bool

Remove a gateway.

Parameters:

Name	Type	Description	Default
`name`	`str`	Gateway name to unregister.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if the gateway existed and was removed.

Raises:

Type	Description
`SQLAlchemyError`	If the commit fails.

get ¶

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

Return a single gateway or None.

Parameters:

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

Returns:

Type	Description
`dict[str, Any] \| None`	dict[str, Any] \| None: Gateway record, or None if not found.

Raises:

Type	Description
`SQLAlchemyError`	If the query fails.

list_all ¶

list_all(
    *, labels_filter: dict[str, str] | None = None
) -> list[dict[str, Any]]

Return all registered gateways, optionally filtered by labels.

Parameters:

Name	Type	Description	Default
`labels_filter`	`dict[str, str] \| None`	If provided, only return gateways whose labels contain all specified key-value pairs.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: Gateway records ordered by name.

Raises:

Type	Description
`SQLAlchemyError`	If the query fails.

update_health ¶

update_health(
    name: str, status: str, last_seen: datetime
) -> None

Update health status and last-seen timestamp.

Parameters:

Name	Type	Description	Default
`name`	`str`	Gateway name.	required
`status`	`str`	New health status string.	required
`last_seen`	`datetime`	Timestamp of the health check.	required

Raises:

Type	Description
`SQLAlchemyError`	If the commit fails.

update_metadata ¶

update_metadata(
    name: str, metadata: dict[str, Any]
) -> None

Replace the metadata JSON blob.

Parameters:

Name	Type	Description	Default
`name`	`str`	Gateway name.	required
`metadata`	`dict[str, Any]`	New metadata dict to store.	required

Raises:

Type	Description
`SQLAlchemyError`	If the commit fails.

update_gateway_metadata ¶

update_gateway_metadata(
    name: str,
    *,
    description: str | None | object = _UNSET,
    labels: dict[str, str] | None | object = _UNSET,
) -> dict[str, Any] | None

Update description and/or labels for a gateway.

Parameters:

Name	Type	Description	Default
`name`	`str`	Gateway name.	required
`description`	`str \| None \| object`	New description, None to clear, or _UNSET to skip.	`_UNSET`
`labels`	`dict[str, str] \| None \| object`	New labels dict, None to clear, or _UNSET to skip.	`_UNSET`

Returns:

Type	Description
`dict[str, Any] \| None`	dict[str, Any] \| None: Updated gateway dict, or None if not found.

Raises:

Type	Description
`SQLAlchemyError`	If the commit fails.

get_credentials ¶

get_credentials(
    name: str,
) -> dict[str, str | bytes | None] | None

Return raw cert bytes for a gateway (for connection logic only).

Parameters:

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

Returns:

Type	Description
`dict[str, str \| bytes \| None] \| None`	dict[str, str \| bytes \| None] \| None: Credential dict, or None if not found.

Raises:

Type	Description
`SQLAlchemyError`	If the query fails.

gateway ¶

Live gRPC channel management for registered gateways.

Companion to :mod:shoreguard.services.registry: the registry owns what gateways exist, this module owns the actual live :class:ShoreGuardClient instances connected to each one. Provides lazy client lookup, connection health probing, graceful channel shutdown on gateway removal, and an in-memory cache so every API call does not pay the mTLS handshake cost.

Separating client lifecycle from persistent CRUD means registry edits (e.g. rename, label change) never need to tear down an in-flight call, and a channel failure only affects the one gateway rather than wedging the whole service.

GatewayService ¶

GatewayService(registry: GatewayRegistry)

Gateway connection management and registry-backed discovery.

Handles gRPC client connections, backoff, health probing, and gateway registration/unregistration.

Parameters:

Name	Type	Description	Default
`registry`	`GatewayRegistry`	Gateway registry for persistence.	required

Attributes:

Name	Type	Description
`registry`	`GatewayRegistry`	The underlying gateway registry.

registry `property` ¶

registry: GatewayRegistry

The underlying gateway registry.

get_client ¶

get_client(name: str) -> ShoreGuardClient

Return a client for the given gateway, attempting reconnect with backoff.

Parameters:

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

Returns:

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

Raises:

Type	Description
`GatewayNotConnectedError`	If connection fails.

set_client ¶

set_client(
    client: ShoreGuardClient | None, name: str
) -> None

Set or clear a client for the given gateway.

Parameters:

Name	Type	Description	Default
`client`	`ShoreGuardClient \| None`	Client to cache, or None to clear.	required
`name`	`str`	Gateway name.	required

reset_backoff ¶

reset_backoff(name: str) -> None

Reset connection backoff for a gateway.

Parameters:

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

register ¶

register(
    name: str,
    endpoint: str,
    scheme: str = "https",
    auth_mode: str | None = "mtls",
    *,
    ca_cert: bytes | None = None,
    client_cert: bytes | None = None,
    client_key: bytes | None = None,
    metadata: dict[str, Any] | None = None,
    description: str | None = None,
    labels: dict[str, str] | None = None,
) -> dict[str, Any]

Register a gateway and attempt initial connection.

Parameters:

Name	Type	Description	Default
`name`	`str`	Unique gateway name.	required
`endpoint`	`str`	Gateway endpoint address.	required
`scheme`	`str`	Connection scheme (e.g. "https").	`'https'`
`auth_mode`	`str \| None`	Authentication mode (e.g. "mtls").	`'mtls'`
`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`
`metadata`	`dict[str, Any] \| None`	Optional metadata dict.	`None`
`description`	`str \| None`	Optional free-text description.	`None`
`labels`	`dict[str, str] \| None`	Optional key-value labels for filtering.	`None`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Gateway record with connection status.

unregister ¶

unregister(name: str) -> bool

Unregister a gateway and close its connection.

Parameters:

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

Returns:

Name	Type	Description
`bool`	`bool`	True if the gateway existed and was removed.

test_connection ¶

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

Explicitly test connectivity to a registered gateway.

Parameters:

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

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Connection test result.

Raises:

Type	Description
`NotFoundError`	If the gateway is not registered.

update_gateway_metadata ¶

update_gateway_metadata(
    name: str,
    *,
    description: str | None | object = _UNSET,
    labels: dict[str, str] | None | object = _UNSET,
) -> dict[str, Any]

Update description and/or labels for a gateway.

Parameters:

Name	Type	Description	Default
`name`	`str`	Gateway name.	required
`description`	`str \| None \| object`	New description, None to clear, or sentinel to skip.	`_UNSET`
`labels`	`dict[str, str] \| None \| object`	New labels dict, None to clear, or sentinel to skip.	`_UNSET`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated gateway record.

Raises:

Type	Description
`NotFoundError`	If the gateway does not exist.

list_all ¶

list_all(
    *, labels_filter: dict[str, str] | None = None
) -> list[dict[str, Any]]

List all registered gateways with cached connection status.

Uses the cached client state instead of live health probes to avoid N+1 blocking gRPC calls. The background health monitor keeps last_status up-to-date.

Parameters:

Name	Type	Description	Default
`labels_filter`	`dict[str, str] \| None`	If provided, only return gateways matching all specified label key-value pairs.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: Gateway records with connection status.

get_info ¶

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

Get detailed info for a gateway.

Parameters:

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

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Detailed gateway information.

get_config ¶

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

Fetch the gateway configuration via gRPC.

Parameters:

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

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Gateway configuration.

update_setting ¶

update_setting(
    name: str,
    key: str,
    value: str | bool | int | None = None,
    *,
    delete: bool = False,
) -> dict[str, Any]

Update (or delete) a single global gateway setting via gRPC.

Parameters:

Name	Type	Description	Default
`name`	`str`	Gateway name.	required
`key`	`str`	Setting key.	required
`value`	`str \| bool \| int \| None`	New value. Ignored when `delete` is True.	`None`
`delete`	`bool`	If True, remove the setting instead of updating it.	`False`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: `{"settings_revision": int, "deleted": bool}`.

check_all_health ¶

check_all_health() -> None

Probe all registered gateways and update their health in the registry.

get_cached_client ¶

get_cached_client(name: str) -> ShoreGuardClient | None

Return the cached client for a gateway, or None if not connected.

Parameters:

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

Returns:

Type	Description
`ShoreGuardClient \| None`	ShoreGuardClient \| None: Cached client, or None.

sandbox ¶

Sandbox lifecycle operations: create, list, delete, exec, metadata.

Wraps the ShoreGuardClient.sandboxes gRPC surface with behavior the raw client does not provide: merging sandbox metadata from the local sandbox_meta_store into list + detail responses, running pre-create validation hooks before CreateSandbox reaches the gateway, and running post-create warm-up hooks inside the new sandbox once it is up.

The create flow is intentionally centralised here so both the Web UI and the TUI can share the same preset-application + metadata + boot-hook sequence — earlier versions of ShoreGuard duplicated most of it in browser JS.

SandboxService ¶

SandboxService(
    client: ShoreGuardClient,
    meta_store: SandboxMetaStore | None = None,
    *,
    boot_hooks: BootHookService | None = None,
    gateway_name: str | None = None,
)

Gateway-scoped sandbox lifecycle operations.

One instance is constructed per (gateway, operation) invocation so the embedded gRPC client is already bound to the right endpoint. Methods are synchronous because the gRPC client is synchronous; async callers wrap them via asyncio.to_thread.

Metadata merging, preset application, and pre/post-create hook dispatch happen inside this service rather than in the route layer so the same flow is available to the TUI and any future programmatic client without duplicated code.

Parameters:

Name	Type	Description	Default
`client`	`ShoreGuardClient`	OpenShell gRPC client bound to the target gateway.	required
`meta_store`	`SandboxMetaStore \| None`	Optional store for sandbox labels + description kept locally in ShoreGuard (OpenShell does not model them).	`None`
`boot_hooks`	`BootHookService \| None`	Optional hook service invoked around `create()`; when `None`, hook support is disabled for this instance even if hooks exist in the database.	`None`
`gateway_name`	`str \| None`	Gateway name this service is bound to, used for hook lookups when `create()` is called without an explicit `gateway_name` argument.	`None`

list ¶

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

List all sandboxes with merged metadata.

Parameters:

Name	Type	Description	Default
`limit`	`int`	Maximum number of sandboxes to return.	`100`
`offset`	`int`	Number of sandboxes to skip.	`0`
`gateway_name`	`str \| None`	Gateway name for metadata lookup.	`None`
`labels_filter`	`dict[str, str] \| None`	Filter sandboxes by label key-value pairs.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: Sandbox records with metadata.

get ¶

get(
    name: str, *, gateway_name: str | None = None
) -> dict[str, Any]

Get a sandbox by name with merged metadata.

Parameters:

Name	Type	Description	Default
`name`	`str`	Sandbox name.	required
`gateway_name`	`str \| None`	Gateway name for metadata lookup.	`None`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Sandbox record with metadata.

delete ¶

delete(
    name: str, *, gateway_name: str | None = None
) -> bool

Delete a sandbox by name and clean up metadata.

Parameters:

Name	Type	Description	Default
`name`	`str`	Sandbox name.	required
`gateway_name`	`str \| None`	Gateway name for metadata cleanup.	`None`

Returns:

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

exec ¶

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

Execute a command inside a sandbox.

Accepts command as a raw string (parsed with shlex) or list.

Parameters:

Name	Type	Description	Default
`name`	`str`	Sandbox name.	required
`command`	`str \| list[str]`	Command as a string or list of arguments.	required
`workdir`	`str`	Working directory inside the sandbox.	`''`
`env`	`dict[str, str] \| None`	Environment variables to set.	`None`
`timeout_seconds`	`int`	Execution timeout (0 for no timeout).	`0`
`tty`	`bool`	Allocate a TTY for the command (for interactive programs).	`False`

Returns:

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

Raises:

Type	Description
`ValidationError`	If the command string has invalid syntax.

get_logs ¶

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

Fetch recent logs from a sandbox.

Parameters:

Name	Type	Description	Default
`name`	`str`	Sandbox name.	required
`lines`	`int`	Maximum number of log lines to return.	`200`
`since_ms`	`int`	Only return logs after this epoch millisecond timestamp.	`0`
`sources`	`list[str] \| None`	Filter by log source names.	`None`
`min_level`	`str`	Minimum log level filter.	`''`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: Log entries.

create_ssh_session ¶

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

Create an SSH session for a sandbox, resolving name to ID.

Parameters:

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

Returns:

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

get_config ¶

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

Get the stored sandbox configuration from the gateway.

Resolves name to sandbox id, then calls GetSandboxConfig. The returned dict is the gateway-side spec (policy, providers, template, etc.) as the gateway currently has it recorded — the source of truth for "what is this sandbox configured with" in the pinning / GitOps workflow.

Parameters:

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

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Sandbox configuration as returned by the gateway.

get_provider_environment ¶

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

Get the provider environment variables injected into a sandbox.

Resolves name to sandbox id, then calls GetSandboxProviderEnvironment. Returns the env map the gateway will hand to the sandbox's runtime when it launches. Values may be secret — callers that surface this over REST should redact against the shared secret-key pattern.

Parameters:

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

Returns:

Type	Description
`dict[str, str]`	dict[str, str]: Environment key-value pairs.

list_providers ¶

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

List provider records attached to a sandbox.

Parameters:

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

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: Attached provider records.

attach_provider ¶

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

Attach a provider record to a sandbox.

Parameters:

Name	Type	Description	Default
`name`	`str`	Sandbox name.	required
`provider_name`	`str`	Provider name to attach.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: `{sandbox, attached}`.

detach_provider ¶

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

Detach a provider record from a sandbox.

Parameters:

Name	Type	Description	Default
`name`	`str`	Sandbox name.	required
`provider_name`	`str`	Provider name to detach.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: `{sandbox, detached}`.

revoke_ssh_session ¶

revoke_ssh_session(token: str) -> bool

Revoke an active SSH session by token.

Parameters:

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

Returns:

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

update_metadata ¶

update_metadata(
    gateway_name: str,
    name: str,
    *,
    description: str | None | object = _UNSET,
    labels: dict[str, str] | None | object = _UNSET,
) -> dict[str, Any]

Update labels and/or description for a sandbox.

Parameters:

Name	Type	Description	Default
`gateway_name`	`str`	Name of the gateway.	required
`name`	`str`	Sandbox name.	required
`description`	`str \| None \| object`	New description (or _UNSET to skip).	`_UNSET`
`labels`	`dict[str, str] \| None \| object`	New labels (or _UNSET to skip).	`_UNSET`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated sandbox record with metadata.

Raises:

Type	Description
`RuntimeError`	If no meta store is configured.

create ¶

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

Create a sandbox and optionally apply presets.

This replaces the multi-step wizard polling loop that was in the browser JS. The entire workflow happens server-side: 1. Create sandbox 2. Wait for ready state 3. Wait for initial policy 4. Apply presets sequentially

Parameters:

Name	Type	Description	Default
`name`	`str`	Sandbox name (empty for auto-generated).	`''`
`image`	`str`	Container image to use.	`''`
`gpu`	`bool`	Whether to enable GPU support.	`False`
`providers`	`list[str] \| None`	Provider names to attach.	`None`
`environment`	`dict[str, str] \| None`	Environment variables to set.	`None`
`presets`	`list[str] \| None`	Policy presets to apply after creation.	`None`
`gateway_name`	`str \| None`	Gateway name for metadata storage.	`None`
`description`	`str \| None`	Optional sandbox description.	`None`
`labels`	`dict[str, str] \| None`	Optional sandbox labels.	`None`
`skip_hooks`	`bool`	If true, bypass pre/post boot hook execution (admin override for cases where a broken hook would block recreation).	`False`
`log_level`	`str`	Supervisor log verbosity (empty = gateway default).	`''`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Created sandbox record with preset status
`dict[str, Any]`	and (when hooks ran) `boot_hooks` result lists. Pre-create
`dict[str, Any]`	hook failures bubble up as `BootHookError` from the boot
`dict[str, Any]`	hook service.

policy ¶

High-level sandbox policy operations for UI and API callers.

Wraps the gRPC ShoreGuardClient.policies surface with additional behavior the raw client does not offer: atomic add/delete of individual network rules and filesystem paths over a read-modify-write loop, eager preset resolution into the stored policy so the gateway's effective view matches the declared document, and submit-time capture of denial context so the approval detail modal can render binary ancestry and L7 samples without a round-trip.

PolicyService ¶

PolicyService(client: ShoreGuardClient)

Policy management shared by the Web UI, TUI, and REST API.

Thin wrapper over ShoreGuardClient.policies that lifts a few operations up from the raw gRPC surface: atomic single-rule CRUD (against an API that only exposes replace-whole-policy), preset application with server-side merge, and denial-context capture for approvals. Everything is synchronous because the gRPC client is synchronous; callers that need to avoid blocking the event loop wrap individual methods in asyncio.to_thread.

Parameters:

Name	Type	Description	Default
`client`	`ShoreGuardClient`	OpenShell gRPC client scoped to the target gateway.	required

get ¶

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

Get the current active policy for a sandbox.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required

Returns:

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

get_effective ¶

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

Get the effective policy — what the gateway currently enforces.

In the current architecture, presets are merged eagerly into the declared policy at apply time (read-modify-write in :meth:apply_preset), so the gateway's stored policy is already the fully resolved document. This method returns that document under a stable /policy/effective contract, giving the UI and API clients a dedicated endpoint that will keep working if OpenShell ever separates declared from effective server-side.

The return value is the same envelope policies.get returns (active_version, revision, policy) with an added source: "gateway_runtime" marker so callers can distinguish this response from the plain GET /policy route.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Effective policy envelope.

submit_analysis ¶

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

Forward a policy analysis submission to the gateway.

Thin pass-through to :meth:PolicyManager.submit_analysis. The gateway merges accepted chunks into the draft policy; rejected chunks come back with a per-chunk reason. Used by external analyzers (LLM-backed or rule-based) that observe sandbox denials and propose fixes — ShoreGuard itself does not generate analysis results, it only brokers them.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Target sandbox name.	required
`summaries`	`list[dict[str, Any]]`	`DenialSummary` dicts. See the OpenShell proto for the field layout.	required
`proposed_chunks`	`list[dict[str, Any]]`	`PolicyChunk` dicts containing the rules that would fix the denials described in summaries.	required
`analysis_mode`	`str`	Opaque mode tag forwarded verbatim, e.g. `"auto"` or `"manual"`.	`''`

Returns:

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

update ¶

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

Push a new policy version and return the full PolicyResponse.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`policy_dict`	`dict`	Policy content as a dict.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated policy response.

update_merge ¶

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

Apply incremental merge operations and return the fresh policy.

Thin wrapper over :meth:shoreguard.client.policies.PolicyManager.apply_merge_operations that reshapes the response into the same PolicyResponse envelope :meth:update returns, so route handlers can treat the two write paths as interchangeable.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`operations`	`list[dict[str, Any]]`	Ordered merge-operation dicts, as synthesised by :func:`shoreguard.services.policy_merge_ops.compute_merge_operations`.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated policy response.

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`	Name of the sandbox.	required
`version`	`int`	Revision version number.	required

Returns:

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

diff_revisions ¶

diff_revisions(
    sandbox_name: str, version_a: int, version_b: int
) -> dict[str, Any]

Fetch two revisions and return both for client-side diffing.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`version_a`	`int`	First revision version number.	required
`version_b`	`int`	Second revision version number.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Both policy revisions for comparison.

list_revisions ¶

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

List policy revision history.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`limit`	`int`	Maximum number of revisions to return.	`20`
`offset`	`int`	Number of revisions to skip.	`0`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: Revision history entries.

apply_preset ¶

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

Apply a policy preset to a sandbox (merges network_policies).

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`preset_name`	`str`	Name of the preset to apply.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated policy response.

Raises:

Type	Description
`NotFoundError`	If the preset name is not found.

add_network_rule ¶

add_network_rule(
    sandbox_name: str, key: str, rule: dict[str, Any]
) -> dict[str, Any]

Add or update a single network rule (read-modify-write).

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`key`	`str`	Network rule key.	required
`rule`	`dict[str, Any]`	Rule definition dict.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated policy response.

delete_network_rule ¶

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

Delete a single network rule (read-modify-write).

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`key`	`str`	Network rule key to remove.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated policy response.

add_filesystem_path ¶

add_filesystem_path(
    sandbox_name: str, path: str, access: str
) -> dict[str, Any]

Add a filesystem path (read-modify-write).

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`path`	`str`	Filesystem path to add.	required
`access`	`str`	Access mode ("ro" or "rw").	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated policy response.

delete_filesystem_path ¶

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

Delete a filesystem path (read-modify-write).

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`path`	`str`	Filesystem path to remove.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated policy response.

update_process_policy ¶

update_process_policy(
    sandbox_name: str,
    *,
    run_as_user: str | None = None,
    run_as_group: str | None = None,
    landlock_compatibility: str | None = None,
) -> dict[str, Any]

Update process and landlock settings (read-modify-write).

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`run_as_user`	`str \| None`	User to run processes as.	`None`
`run_as_group`	`str \| None`	Group to run processes as.	`None`
`landlock_compatibility`	`str \| None`	Landlock compatibility level.	`None`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Updated policy response.

approvals ¶

Draft-policy approval workflow wrapper over the gRPC approval RPCs.

OpenShell's approval flow works in terms of draft policy chunks: each time a denial fires, the gateway emits a chunk that proposes the rules that would allow the blocked action. This service mediates the operator-facing view of that flow — listing pending chunks, approving / rejecting / editing individual ones, and supporting bulk approve-all and undo operations.

The service is deliberately thin: almost every call forwards directly to ShoreGuardClient.approvals with additional request-level validation and audit logging. Multi-stage quorum is not handled here — that lives in :class:~shoreguard.services.approval_workflow.ApprovalWorkflowService, which the approval route consults before calling through to this service.

ApprovalService ¶

ApprovalService(client: ShoreGuardClient)

Draft policy approval operations shared by Web UI and TUI.

Parameters:

Name	Type	Description	Default
`client`	`ShoreGuardClient`	OpenShell gRPC client instance.	required

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`	Name of the sandbox.	required
`status_filter`	`str`	Optional status to filter by.	`''`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Draft policy data with denial context enrichment.

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`	Name of the sandbox.	required

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: Pending draft chunks with denial context.

approve ¶

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

Approve a single draft policy chunk.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`chunk_id`	`str`	Identifier of the chunk to approve.	required

Returns:

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

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`	Name of the sandbox.	required
`chunk_id`	`str`	Identifier of the chunk to reject.	required
`reason`	`str`	Optional reason for rejection.	`''`

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`	Name of the sandbox.	required
`include_security_flagged`	`bool`	Whether to include security-flagged chunks.	`False`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Summary of approved chunks.

edit ¶

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

Edit a pending draft chunk's proposed rule.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`chunk_id`	`str`	Identifier of the chunk to edit.	required
`proposed_rule`	`dict`	New proposed rule content.	required

undo ¶

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

Reverse an approval decision.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required
`chunk_id`	`str`	Identifier of the chunk to undo.	required

Returns:

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

clear ¶

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

Clear all pending draft chunks for a sandbox.

Parameters:

Name	Type	Description	Default
`sandbox_name`	`str`	Name of the sandbox.	required

Returns:

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

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`	Name of the sandbox.	required

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: Decision history entries.

operations ¶

Persistent state for long-running async operations.

Some REST endpoints — sandbox create, SSH session setup, gateway restart — take long enough that returning 202 with an operation id is better than holding the HTTP connection open. This service backs that pattern: endpoints enqueue an operation, return the id, and clients poll GET /operations/{id} until the status flips from running to a terminal value.

Operations are stored in SQL so a poll survives server restarts and fan out across replicas via the shared database. The worker side of each operation lives in the calling service; this module only owns lifecycle and state transitions.

AsyncOperationService ¶

AsyncOperationService(
    session_factory: async_sessionmaker[AsyncSession],
    running_ttl: float = 600.0,
    retention_days: int = 30,
)

DB-backed operation tracking using async SQLAlchemy.

Provides create/start/complete/fail/cancel plus cleanup helpers, with an in-memory task registry that lets cancel() interrupt a running asyncio task.

Parameters:

Name	Type	Description	Default
`session_factory`	`async_sessionmaker[AsyncSession]`	Async SQLAlchemy session factory.	required
`running_ttl`	`float`	Seconds before a stuck running operation is timed out.	`600.0`
`retention_days`	`int`	Days to keep completed operations before cleanup.	`30`

create `async` ¶

create(
    resource_type: str,
    resource_key: str,
    *,
    actor: str | None = None,
    gateway_name: str | None = None,
    idempotency_key: str | None = None,
) -> OperationRecord

Create a new operation in pending state.

Parameters:

Name	Type	Description	Default
`resource_type`	`str`	Type of resource being operated on.	required
`resource_key`	`str`	Name of the resource.	required
`actor`	`str \| None`	Identity of the user starting the operation.	`None`
`gateway_name`	`str \| None`	Gateway the operation targets.	`None`
`idempotency_key`	`str \| None`	Optional client-provided idempotency key.	`None`

Returns:

Name	Type	Description
`OperationRecord`	`OperationRecord`	The newly created operation record.

create_if_not_running `async` ¶

create_if_not_running(
    resource_type: str,
    resource_key: str,
    *,
    actor: str | None = None,
    gateway_name: str | None = None,
    idempotency_key: str | None = None,
) -> OperationRecord | None

Create a new operation if none is active for this resource.

Parameters:

Name	Type	Description	Default
`resource_type`	`str`	Type of resource being operated on.	required
`resource_key`	`str`	Name of the resource.	required
`actor`	`str \| None`	Identity of the user starting the operation.	`None`
`gateway_name`	`str \| None`	Gateway the operation targets.	`None`
`idempotency_key`	`str \| None`	Optional client-provided idempotency key.	`None`

Returns:

Type	Description
`OperationRecord \| None`	OperationRecord \| None: The new operation, or None if one is already active.

start `async` ¶

start(op_id: str) -> None

Transition pending → running.

Parameters:

Name	Type	Description	Default
`op_id`	`str`	The operation ID to start.	required

complete `async` ¶

complete(op_id: str, result: dict[str, Any]) -> None

Mark an operation as succeeded.

Parameters:

Name	Type	Description	Default
`op_id`	`str`	The operation ID to complete.	required
`result`	`dict[str, Any]`	Result payload to store.	required

fail `async` ¶

fail(
    op_id: str,
    error: str,
    error_code: str = ErrorCode.internal,
) -> None

Mark an operation as failed.

Parameters:

Name	Type	Description	Default
`op_id`	`str`	The operation ID to mark as failed.	required
`error`	`str`	Human-readable error message.	required
`error_code`	`str`	Machine-readable error code.	`internal`

update_progress `async` ¶

update_progress(
    op_id: str, pct: int, message: str | None = None
) -> None

Update progress for an active operation.

Parameters:

Name	Type	Description	Default
`op_id`	`str`	The operation ID to update.	required
`pct`	`int`	Progress percentage (0-100).	required
`message`	`str \| None`	Optional progress message.	`None`

get `async` ¶

get(op_id: str) -> OperationRecord | None

Get an operation by ID.

Parameters:

Name	Type	Description	Default
`op_id`	`str`	The operation ID to look up.	required

Returns:

Type	Description
`OperationRecord \| None`	OperationRecord \| None: The operation record, or None if not found.

get_by_idempotency_key `async` ¶

get_by_idempotency_key(key: str) -> OperationRecord | None

Look up an operation by its idempotency key.

Parameters:

Name	Type	Description	Default
`key`	`str`	The idempotency key to search for.	required

Returns:

Type	Description
`OperationRecord \| None`	OperationRecord \| None: The matching operation record, or None.

list_ops `async` ¶

list_ops(
    *,
    status: str | None = None,
    resource_type: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> tuple[list[OperationRecord], int]

List operations with optional filtering.

Parameters:

Name	Type	Description	Default
`status`	`str \| None`	Filter by status.	`None`
`resource_type`	`str \| None`	Filter by resource type.	`None`
`limit`	`int`	Maximum number of results (max 200).	`50`
`offset`	`int`	Number of results to skip.	`0`

Returns:

Type	Description
`tuple[list[OperationRecord], int]`	tuple[list[OperationRecord], int]: Tuple of (operations, total_count).

is_running `async` ¶

is_running(resource_type: str, resource_key: str) -> bool

Check if there is an active operation for the given resource.

Parameters:

Name	Type	Description	Default
`resource_type`	`str`	Type of resource to check.	required
`resource_key`	`str`	Name of the resource to check.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if an active (pending/running) operation exists.

status_counts `async` ¶

status_counts() -> dict[str, int]

Return counts of operations grouped by status.

Returns:

Type	Description
`dict[str, int]`	dict[str, int]: Mapping of status to count.

register_task ¶

register_task(op_id: str, task: Task[None]) -> None

Register an asyncio task for cancellation support.

Parameters:

Name	Type	Description	Default
`op_id`	`str`	The operation ID.	required
`task`	`Task[None]`	The asyncio task running the operation.	required

cancel `async` ¶

cancel(op_id: str) -> OperationRecord | None

Cancel an active operation.

Parameters:

Name	Type	Description	Default
`op_id`	`str`	The operation ID to cancel.	required

Returns:

Type	Description
`OperationRecord \| None`	OperationRecord \| None: The updated operation record, or None if
`OperationRecord \| None`	not found or not active.

recover_orphans `async` ¶

recover_orphans() -> int

Mark all active operations as failed (startup recovery).

Returns:

Name	Type	Description
`int`	`int`	Number of orphaned operations recovered.

cleanup `async` ¶

cleanup() -> int

Expire stuck active operations and remove old completed ones.

Returns:

Name	Type	Description
`int`	`int`	Number of operations cleaned up.

to_dict `staticmethod` ¶

to_dict(op: OperationRecord) -> dict[str, Any]

Convert an operation record to a JSON-serializable dict.

Parameters:

Name	Type	Description	Default
`op`	`OperationRecord`	The operation record to convert.	required

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: JSON-serializable representation for the API.

providers ¶

LLM provider CRUD with schema metadata from openshell.yaml.

Providers are the named credential containers OpenShell uses to route sandbox inference (Anthropic, OpenAI, Nvidia, ...). This service wraps the gRPC provider RPCs and overlays schema metadata read from the gateway's openshell.yaml so the UI can render a typed form rather than a generic KV editor: which fields each provider type requires, which are secret, which are optional, and what the default env-var projection looks like.

Credentials only flow one way — into the service — and are never returned in responses; the /env projection redacts values to [REDACTED] with a source annotation so operators can debug provider configuration without seeing the secret material.

ProviderService ¶

ProviderService(client: ShoreGuardClient)

Provider management shared by Web UI and TUI.

Parameters:

Name	Type	Description	Default
`client`	`ShoreGuardClient`	OpenShell gRPC client instance.	required

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`	Number of providers to skip.	`0`

Returns:

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

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 record.

create ¶

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

Create a provider with automatic credential key mapping.

Looks up the correct environment variable name from openshell.yaml. Falls back to API_KEY for unknown types.

Parameters:

Name	Type	Description	Default
`name`	`str`	Provider name.	required
`provider_type`	`str`	Provider type identifier.	required
`api_key`	`str`	Primary API key value.	required
`extra_credentials`	`dict[str, str] \| None`	Additional credential key-value pairs.	`None`
`config`	`dict[str, str] \| None`	Optional provider configuration.	`None`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: The created provider record.

update ¶

update(
    *,
    name: str,
    provider_type: str = "",
    credentials: dict[str, str] | None = None,
    config: 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`	New provider type (empty string to keep current).	`''`
`credentials`	`dict[str, str] \| None`	New credential key-value pairs.	`None`
`config`	`dict[str, str] \| None`	New provider configuration.	`None`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: The updated provider record.

delete ¶

delete(name: str) -> bool

Delete a provider.

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.

get_env ¶

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

Get the redacted environment projection for a provider.

Returns the environment variables this provider injects into sandboxes. Secret values are never included — only keys, their source (credential, config, type_default), and a constant [REDACTED] placeholder.

The key set is derived from three places:

Every key in the provider's credentials dict (source=credential).
Every key in the provider's config dict (source=config).
The provider type's cred_key from openshell.yaml (source=type_default), added only if it was not already present in the credentials dict.

Parameters:

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

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Record with `provider`, `type` and `env`.

list_known_types `staticmethod` ¶

list_known_types() -> list[dict[str, str]]

Return metadata about known provider types from openshell.yaml.

Returns:

Type	Description
`list[dict[str, str]]`	list[dict[str, str]]: Provider type metadata records.

list_inference_providers `staticmethod` ¶

list_inference_providers() -> list[dict[str, str]]

Return known inference provider options from openshell.yaml.

Returns:

Type	Description
`list[dict[str, str]]`	list[dict[str, str]]: Inference provider option records.

list_community_sandboxes `staticmethod` ¶

list_community_sandboxes() -> list[dict[str, Any]]

Return community sandbox templates from openshell.yaml.

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: Community sandbox template records.

audit ¶

Persistent audit log for every state-changing operation.

Every API route that mutates state calls audit_log with a short action verb (policy.update, approval.approved, sbom.uploaded, ...), a resource type, a resource id, and optional detail payload. The row is written asynchronously so a slow audit path never blocks the request, and a background task fans it out to the webhook pipeline.

The audit log is the single source of truth for "who did what when" — it is filterable, exportable as CSV or JSON, and deliberately append-only. Retention is up to operators; there is no built-in prune.

AuditIntegrityError ¶

Bases: RuntimeError

Raised when code tries to mutate or delete an AuditEntry illegally.

AuditService ¶

AuditService(
    session_factory: sessionmaker,
    exporter: AuditExporter | None = None,
)

DB-backed audit trail for all state-changing operations.

Parameters:

Name	Type	Description	Default
`session_factory`	`sessionmaker`	SQLAlchemy session factory for database access.	required
`exporter`	`AuditExporter \| None`	Optional :class:`AuditExporter` that fans successful writes out across stdout-JSON, syslog, and webhook lanes.	`None`

log ¶

log(
    *,
    actor: str,
    actor_role: str,
    action: str,
    resource_type: str,
    resource_id: str = "",
    gateway: str | None = None,
    detail: dict[str, Any] | None = None,
    client_ip: str | None = None,
) -> None

Write an audit entry. Never raises -- failures are logged and swallowed.

Parameters:

Name	Type	Description	Default
`actor`	`str`	Identity of the user performing the action.	required
`actor_role`	`str`	Role of the actor (e.g. "admin", "viewer").	required
`action`	`str`	Action being performed (e.g. "create", "delete").	required
`resource_type`	`str`	Type of resource affected.	required
`resource_id`	`str`	Identifier of the affected resource.	`''`
`gateway`	`str \| None`	Optional gateway name for scoping.	`None`
`detail`	`dict[str, Any] \| None`	Optional structured detail payload.	`None`
`client_ip`	`str \| None`	IP address of the client.	`None`

list ¶

list(
    *,
    limit: int = 100,
    offset: int = 0,
    actor: str | None = None,
    action: str | None = None,
    resource_type: str | None = None,
    gateway: str | None = None,
    since: str | None = None,
    until: str | None = None,
) -> list[dict[str, Any]]

Query audit entries with optional filters and pagination.

Parameters:

Name	Type	Description	Default
`limit`	`int`	Maximum number of entries to return.	`100`
`offset`	`int`	Number of entries to skip.	`0`
`actor`	`str \| None`	Filter by actor identity.	`None`
`action`	`str \| None`	Filter by action type.	`None`
`resource_type`	`str \| None`	Filter by resource type.	`None`
`gateway`	`str \| None`	Filter by gateway name (matches the `gateway_name` column).	`None`
`since`	`str \| None`	ISO-format start timestamp filter.	`None`
`until`	`str \| None`	ISO-format end timestamp filter.	`None`

Returns:

Type	Description
`list[dict[str, Any]]`	list[dict[str, Any]]: Matching audit entries.

list_with_count ¶

list_with_count(
    *,
    limit: int = 100,
    offset: int = 0,
    actor: str | None = None,
    action: str | None = None,
    resource_type: str | None = None,
    gateway: str | None = None,
    since: str | None = None,
    until: str | None = None,
) -> tuple[list[dict[str, Any]], int]

Query audit entries with total count for pagination.

Parameters:

Name	Type	Description	Default
`limit`	`int`	Maximum number of entries to return.	`100`
`offset`	`int`	Number of entries to skip.	`0`
`actor`	`str \| None`	Filter by actor identity.	`None`
`action`	`str \| None`	Filter by action type.	`None`
`resource_type`	`str \| None`	Filter by resource type.	`None`
`gateway`	`str \| None`	Filter by gateway name.	`None`
`since`	`str \| None`	ISO-format start timestamp filter.	`None`
`until`	`str \| None`	ISO-format end timestamp filter.	`None`

Returns:

Type	Description
`tuple[list[dict[str, Any]], int]`	tuple[list[dict[str, Any]], int]: Entries and total matching count.

export_csv ¶

export_csv(
    *,
    actor: str | None = None,
    action: str | None = None,
    resource_type: str | None = None,
    gateway: str | None = None,
    since: str | None = None,
    until: str | None = None,
) -> str

Export audit entries as a CSV string.

Parameters:

Name	Type	Description	Default
`actor`	`str \| None`	Filter by actor identity.	`None`
`action`	`str \| None`	Filter by action type.	`None`
`resource_type`	`str \| None`	Filter by resource type.	`None`
`gateway`	`str \| None`	Filter by gateway name.	`None`
`since`	`str \| None`	ISO-format start timestamp filter.	`None`
`until`	`str \| None`	ISO-format end timestamp filter.	`None`

Returns:

Name	Type	Description
`str`	`str`	CSV-formatted string of matching entries.

cleanup ¶

cleanup(older_than_days: int | None = None) -> int

Delete audit entries older than the given number of days.

Uses row-by-row deletion (not Query.delete()) so that the before_delete listener fires and the ContextVar bypass is honoured. Retention cleanup runs once per cleanup_interval so the slight extra cost is negligible.

Parameters:

Name	Type	Description	Default
`older_than_days`	`int \| None`	Age threshold in days.	`None`

Returns:

Name	Type	Description
`int`	`int`	Number of entries deleted.

export_json ¶

export_json(
    *,
    actor: str | None = None,
    action: str | None = None,
    resource_type: str | None = None,
    gateway: str | None = None,
    since: str | None = None,
    until: str | None = None,
) -> str

Export audit entries as a JSON array string.

Parameters:

Name	Type	Description	Default
`actor`	`str \| None`	Filter by actor identity.	`None`
`action`	`str \| None`	Filter by action type.	`None`
`resource_type`	`str \| None`	Filter by resource type.	`None`
`gateway`	`str \| None`	Filter by gateway name.	`None`
`since`	`str \| None`	ISO-format start timestamp filter.	`None`
`until`	`str \| None`	ISO-format end timestamp filter.	`None`

Returns:

Name	Type	Description
`str`	`str`	JSON-formatted list of matching entries.

audit_log `async` ¶

audit_log(
    request: Request,
    action: str,
    resource_type: str,
    resource_id: str = "",
    *,
    gateway: str | None = None,
    detail: dict[str, Any] | None = None,
) -> None

Route-handler helper that extracts actor/role/IP from the request.

Parameters:

Name	Type	Description	Default
`request`	`Request`	The incoming HTTP request.	required
`action`	`str`	Action being performed.	required
`resource_type`	`str`	Type of resource affected.	required
`resource_id`	`str`	Identifier of the affected resource.	`''`
`gateway`	`str \| None`	Optional gateway name for scoping.	`None`
`detail`	`dict[str, Any] \| None`	Optional structured detail payload.	`None`

local_gateway ¶

Local gateway lifecycle — Docker containers, openshell CLI, port management.

Only used when SHOREGUARD_LOCAL_MODE=1 or for backward-compatible v0.2 workflows.

LocalGatewayManager ¶

LocalGatewayManager(gateway_service: GatewayService)

Docker and openshell CLI lifecycle for locally-managed gateways.

Parameters:

Name	Type	Description	Default
`gateway_service`	`GatewayService`	Gateway service for connection management.	required

diagnostics ¶

diagnostics() -> dict[str, Any]

Check Docker availability, daemon status, and permissions.

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Diagnostic information about Docker and openshell.

start ¶

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

Start a gateway by name.

Parameters:

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

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Result with success status and output or error.

stop ¶

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

Stop a gateway by name.

Parameters:

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

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Result with success status and output or error.

restart ¶

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

Restart a gateway (stop + start).

Parameters:

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

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Result from the start step.

create ¶

create(
    name: str,
    port: int | None = None,
    *,
    remote_host: str | None = None,
    gpu: bool = False,
) -> dict[str, Any]

Create a new gateway via openshell CLI.

Parameters:

Name	Type	Description	Default
`name`	`str`	Gateway name.	required
`port`	`int \| None`	Port number, or None for auto-selection.	`None`
`remote_host`	`str \| None`	Remote host for the gateway.	`None`
`gpu`	`bool`	Whether to enable GPU support.	`False`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Gateway info or error result.

destroy ¶

destroy(
    name: str, *, force: bool = False
) -> dict[str, Any]

Destroy a gateway and remove its configuration.

Parameters:

Name	Type	Description	Default
`name`	`str`	Gateway name.	required
`force`	`bool`	Force destruction even if resources still exist.	`False`

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: Result with success status.

Services¶

registry ¶

GatewayRegistry ¶

register ¶

unregister ¶

get ¶

list_all ¶

update_health ¶

update_metadata ¶

update_gateway_metadata ¶

get_credentials ¶

gateway ¶

GatewayService ¶

registry property ¶

get_client ¶

set_client ¶

reset_backoff ¶

register ¶

unregister ¶

test_connection ¶

update_gateway_metadata ¶

list_all ¶

get_info ¶

get_config ¶

update_setting ¶

check_all_health ¶

get_cached_client ¶

sandbox ¶

SandboxService ¶

list ¶

get ¶

delete ¶

exec ¶

get_logs ¶

create_ssh_session ¶

get_config ¶

get_provider_environment ¶

list_providers ¶

attach_provider ¶

detach_provider ¶

revoke_ssh_session ¶

update_metadata ¶

create ¶

policy ¶

PolicyService ¶

get ¶

get_effective ¶

submit_analysis ¶

update ¶

update_merge ¶

get_version ¶

diff_revisions ¶

list_revisions ¶

apply_preset ¶

add_network_rule ¶

delete_network_rule ¶

add_filesystem_path ¶

delete_filesystem_path ¶

update_process_policy ¶

approvals ¶

ApprovalService ¶

get_draft ¶

get_pending ¶

approve ¶

reject ¶

approve_all ¶

edit ¶

undo ¶

clear ¶

get_history ¶

operations ¶

AsyncOperationService ¶

create async ¶

create_if_not_running async ¶

start async ¶

complete async ¶

fail async ¶

update_progress async ¶

get async ¶

get_by_idempotency_key async ¶

registry `property` ¶

create `async` ¶

create_if_not_running `async` ¶

start `async` ¶

complete `async` ¶

fail `async` ¶

update_progress `async` ¶

get `async` ¶

get_by_idempotency_key `async` ¶

list_ops `async` ¶

is_running `async` ¶

status_counts `async` ¶

cancel `async` ¶

recover_orphans `async` ¶

cleanup `async` ¶

to_dict `staticmethod` ¶

list_known_types `staticmethod` ¶

list_inference_providers `staticmethod` ¶

list_community_sandboxes `staticmethod` ¶

audit_log `async` ¶