// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.

// SPDX-License-Identifier: Apache-2.0

//! `OpenShell` Server library.

//!

//! This crate provides the server implementation for `OpenShell`, including:

//! - gRPC service implementation

//! - HTTP health endpoints

//! - Protocol multiplexing (gRPC + HTTP on same port)

//! - mTLS support

//!

//! Compiled-in compute drivers are installed into a registry at gateway

//! startup. Runtime selection only consults that registry or a configured

//! external endpoint; it does not switch on driver names.

mod auth;

mod bus_sweeper;

pub mod certgen;

pub mod cli;

mod compute;

mod config_delivery;

pub mod config_file;

mod config_update_operation;

mod credential_store;

mod credentials;

mod defaults;

mod gateway_listener;

mod gateway_members;

mod gateway_metrics;

mod gateway_ocsf;

mod gateway_ring;

mod grpc;

mod http;

mod middleware;

mod multiplex;

mod network_trust;

mod ocsf_log;

mod otel_tracing;

mod pagination;

mod persistence;

pub(crate) mod policy_store;

mod provider_profile_sources;

mod provider_refresh;

mod readiness;

mod reflection;

mod refresh_commit;

mod refresh_store;

mod sandbox_index;

mod sandbox_watch;

mod service_routing;

mod ssh_identity;

mod ssh_sessions;

mod storage_proto;

mod supervisor_owner;

mod supervisor_pool;

pub mod supervisor_session;

mod telemetry;

#[cfg(any(test, feature = "test-support"))]

pub mod test_support;

mod tls;

#[cfg(test)]

pub(crate) mod tls_test_utils;

pub mod tracing_bus;

mod tracing_setup;

mod watch_cursor;

mod ws_tunnel;

use axum::Router;

use hyper_util::{

rt::{TokioExecutor, TokioIo},

server::conn::auto::Builder,

service::TowerToHyperService,

};

use openshell_core::net::set_tcp_nodelay_best_effort;

use openshell_core::telemetry::TelemetryComputeDriver;

use openshell_core::{Config, Error, NetworkSupervisorTrustBundle, ObjectLabels, Result};

use openshell_extension_core::{

BearerTokenSlot, ExtensionAudience, ExtensionCallerKind, ExtensionKind, MAX_EXTENSION_TOKEN_TTL,

};

use openshell_ocsf::{ConfigStateChangeBuilder, SeverityId, StateId, StatusId};

use openshell_supervisor_middleware::MiddlewareRegistry;

use std::collections::{BTreeMap, HashMap};

use std::io::ErrorKind;

use std::net::SocketAddr;

use std::path::{Path, PathBuf};

#[cfg(test)]

use std::sync::LazyLock;

use std::sync::{

Arc, Mutex, OnceLock,

atomic::{AtomicBool, AtomicU64, Ordering},

};

use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use tokio::net::{TcpListener, TcpStream};

use tokio::sync::watch;

use tracing::{debug, error, info, warn};

/// Serializes tests that assert on captured spans, which share one exporter.

#[cfg(test)]

pub(crate) static TEST_TRACING_LOCK: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(()));

pub(crate) fn install_jsonwebtoken_crypto_provider() {

openshell_crypto::install_jwt_provider();

}

use compute::ComputeRuntime;

use gateway_listener::{BoundGatewayListener, bind_gateway_listener};

pub use grpc::OpenShellService;

pub use http::{health_router, http_router, metrics_router, service_http_router};

/// Deriving `Debug` is safe here: `BearerTokenSlot` renders only its expiry,

/// and an extension audience is configuration rather than secret material.

#[derive(Debug)]

struct GatewayExtensionCredential {

name: String,

audience: ExtensionAudience,

slot: BearerTokenSlot,

ttl: Duration,

}

fn extension_token_ttl(issuer: &auth::sandbox_jwt::ExtensionJwtIssuer) -> Duration {

issuer.token_ttl().min(MAX_EXTENSION_TOKEN_TTL)

}

/// Mint the gateway-caller credential for one extension registration.

///

/// Returns `Ok(None)` when the operator has explicitly opted the registration

/// out of extension authentication. The opt-out is deliberately loud: it

/// downgrades a security boundary, so it is reported once per registration at

/// startup rather than being silently tolerated.

fn mint_gateway_extension_credential(

issuer: &Arc<auth::sandbox_jwt::ExtensionJwtIssuer>,

kind: ExtensionKind,

name: &str,

audience: &str,

endpoint: &str,

allow_insecure_transport: bool,

) -> Result<Option<GatewayExtensionCredential>> {

// A middleware endpoint must be reachable from sandbox supervisors, so a

// gateway-local Unix socket is only an option for interceptors.

let (accepted, supports_unix) = match kind {

ExtensionKind::Middleware => ("https://", false),

ExtensionKind::Interceptor => ("https:// or unix://", true),

_ => {

return Err(Error::config(format!(

"extension kind '{kind}' is not supported by gateway authentication"

)));

}

};

if allow_insecure_transport {

warn!(

extension = %name,

endpoint = %endpoint,

"extension authentication is DISABLED for this registration by \

allow_insecure_transport; OpenShell attaches no caller credential \

and the service cannot distinguish OpenShell from any other \

network client. Use {accepted} with the opt-out removed outside \

trusted-network development deployments."

);

return Ok(None);

}

let transport_supported =

endpoint.starts_with("https://") || (supports_unix && endpoint.starts_with("unix://"));

if !transport_supported {

return Err(Error::config(format!(

"authenticated {kind} '{name}' must use {accepted}; set \

allow_insecure_transport = true to opt this registration out of \

extension authentication instead"

)));

}

let audience = ExtensionAudience::new(audience.to_string()).map_err(|error| {

Error::config(format!(

"extension '{name}' has an invalid audience: {error}"

))

})?;

let ttl = extension_token_ttl(issuer);

let minted = issuer

.mint_extension_token(&audience, ExtensionCallerKind::Gateway, None, ttl)

.map_err(|status| {

Error::config(format!(

"failed to mint credential for extension '{name}': {}",

status.message()

))

})?;

let slot = BearerTokenSlot::new(&minted.token, minted.expires_at_ms).map_err(|error| {

Error::config(format!(

"failed to install credential for extension '{name}': {error}"

))

})?;

Ok(Some(GatewayExtensionCredential {

name: name.to_string(),

audience,

slot,

ttl,

}))

}

fn spawn_gateway_extension_token_refresh(

issuer: Arc<auth::sandbox_jwt::ExtensionJwtIssuer>,

credentials: Vec<GatewayExtensionCredential>,

) {

if credentials.is_empty() {

return;

}

tokio::spawn(async move {

loop {

let now_ms = SystemTime::now()

.duration_since(UNIX_EPOCH)

.map_or(0, |duration| {

i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)

});

let remaining_ms = credentials

.iter()

.filter_map(|credential| credential.slot.expires_at_ms())

.min()

.map_or(60_000, |expiry_ms| expiry_ms.saturating_sub(now_ms));

let refresh_delay = if remaining_ms <= 0 {

Duration::from_millis(100)

} else {

Duration::from_millis(

u64::try_from(remaining_ms)

.unwrap_or(u64::MAX)

.saturating_mul(4)

.checked_div(5)

.unwrap_or(100)

.max(100),

)

};

tokio::time::sleep(refresh_delay).await;

for credential in &credentials {

match issuer.mint_extension_token(

&credential.audience,

ExtensionCallerKind::Gateway,

None,

credential.ttl,

) {

Ok(minted) => {

if let Err(error) =

credential.slot.update(&minted.token, minted.expires_at_ms)

{

warn!(

extension = %credential.name,

error = %error,

"failed to rotate gateway extension credential"

);

}

}

Err(status) => warn!(

extension = %credential.name,

error = %status,

"failed to mint gateway extension credential"

),

}

}

}

});

}

/// Mint the shared supervisor's token and renew it at half its lifetime so

/// the delivered copy is never close to expiry when the supervisor restarts

/// or reconnects. Minting failures are logged and retried shortly.

fn spawn_supervisor_token_refresh(

authority: Arc<auth::sandbox_jwt::SandboxSessionJwtAuthority>,

ttl: Duration,

controller: openshell_supervisor_control::SupervisorHostController,

mut shutdown_rx: watch::Receiver<bool>,

) {

tokio::spawn(async move {

loop {

let delay = match authority.mint_supervisor_token(ttl) {

Ok(minted) => {

let remaining = minted.expires_at.saturating_sub(

i64::try_from(

SystemTime::now()

.duration_since(UNIX_EPOCH)

.map_or(0, |elapsed| elapsed.as_secs()),

)

.unwrap_or(i64::MAX),

);

controller.publish_credential(

openshell_supervisor_control::SupervisorCredential {

token: minted.token,

expires_at: minted.expires_at,

},

);

Duration::from_secs(u64::try_from(remaining).unwrap_or(0) / 2)

.max(Duration::from_secs(1))

}

Err(error) => {

warn!(%error, "failed to mint the supervisor token; retrying");

Duration::from_secs(10)

}

};

tokio::select! {

() = tokio::time::sleep(delay) => {}

changed = shutdown_rx.changed() => {

if changed.is_err() || *shutdown_rx.borrow() {

return;

}

}

}

}

});

}

pub use auth::audit::ConnectionSource;

pub use multiplex::{MultiplexService, MultiplexedService};

pub use otel_tracing::gateway_trace_headers;

use otel_tracing::supervisor_otlp_headers;

pub use persistence::Store;

use sandbox_index::SandboxIndex;

use sandbox_watch::SandboxWatchBus;

pub use tls::TlsAcceptor;

use tracing_bus::TracingLogBus;

pub(crate) struct ServerStartupConfig {

pub config: Config,

pub config_file: Option<config_file::ConfigFile>,

pub guest_tls: Option<compute::driver_config::GuestTlsPaths>,

pub network_trust: Option<NetworkSupervisorTrustBundle>,

pub compute_driver: ComputeDriverSelection,

pub legacy_compute_driver_env_seen: bool,

}

/// Server state shared across handlers.

#[derive(Debug)]

pub struct ServerState {

/// Server configuration.

pub config: Config,

/// Persistence store.

pub store: Arc<Store>,

/// Compute orchestration over the configured driver.

pub compute: ComputeRuntime,

/// Credential-driver selection and resolution runtime.

pub credentials: credentials::CredentialRuntime,

/// Shared embedded refresh runtime (separate from storage-only consumers).

pub(crate) refresh: openshell_credential_issuer::refresh::RefreshRuntime,

/// In-memory sandbox correlation index.

pub sandbox_index: SandboxIndex,

/// In-memory bus for sandbox update notifications.

pub sandbox_watch_bus: SandboxWatchBus,

/// In-memory bus for server process logs.

pub tracing_log_bus: TracingLogBus,

/// In-memory anonymous telemetry accounting for active sandbox sessions.

pub(crate) telemetry: telemetry::TelemetryState,

/// Active SSH tunnel connection counts per session token.

pub ssh_connections_by_token: Mutex<HashMap<String, u32>>,

/// Active SSH tunnel connection counts per sandbox id.

pub ssh_connections_by_sandbox: Mutex<HashMap<String, u32>>,

/// Active service forward connection counts per sandbox id.

pub service_connections_by_sandbox: Mutex<HashMap<String, u32>>,

/// Registry of active supervisor sessions and pending relay channels.

///

/// Stored as `Arc` so compiled compute drivers can be constructed before

/// `ServerState` and still

/// query session state to surface supervisor readiness.

pub supervisor_sessions: Arc<supervisor_session::SupervisorSessionRegistry>,

/// Set once graceful gateway shutdown begins so stream handlers can

/// distinguish expected transport closes from runtime failures.

pub(crate) gateway_shutting_down: AtomicBool,

/// Per-session delivery tasks for supervisor configuration.

pub(crate) config_delivery: config_delivery::ConfigDelivery,

/// Stable identity for this gateway process.

pub replica_id: String,

/// Internal endpoint other gateway replicas can dial for peer RPCs.

pub peer_endpoint: Option<String>,

/// Reused peer connections, peer token, and owner lookups for relay

/// forwarding. Keeps per-relay cost off the connection and auth paths.

pub peer_routes: Arc<supervisor_session::PeerRouteCache>,

/// Idle HTTP/1 upstreams to sandbox services, so routed requests reuse a

/// relay instead of opening one per request.

pub service_upstreams: Arc<service_routing::ServiceUpstreamPool>,

/// In-memory hostname-to-service index for opaque service routing labels.

pub service_route_index: Arc<service_routing::ServiceRouteIndex>,

/// Latest placement ring, refreshed from live gateway membership. Decides

/// which replica *should* own a sandbox; the owner record in the store

/// remains authoritative for which one does.

pub gateway_ring: Arc<std::sync::RwLock<gateway_ring::GatewayRing>>,

/// Peer endpoints of live replicas, keyed by replica ID. Refreshed

/// alongside `gateway_ring` so a redirect can name a dialable address.

pub gateway_peers: Arc<std::sync::RwLock<HashMap<String, String>>>,

/// Validated built-in and operator-registered supervisor middleware.

pub middleware_registry: Arc<MiddlewareRegistry>,

/// OIDC JWKS cache for JWT validation. `None` when OIDC is not configured.

pub oidc_cache: Option<Arc<auth::oidc::JwksCache>>,

/// Typed extension JWT issuer and public verification metadata.

pub extension_jwt_issuer: Option<Arc<auth::sandbox_jwt::ExtensionJwtIssuer>>,

/// Launch-scoped gateway and Sandbox Protocol token authority.

pub sandbox_session_jwt_authority: Option<Arc<auth::sandbox_jwt::SandboxSessionJwtAuthority>>,

/// Optional selected-driver authenticator for the `IssueSandboxToken`

/// bootstrap path.

pub compute_driver_authenticator: Option<Arc<auth::compute_driver::ComputeDriverAuthenticator>>,

/// Optional K8s `ServiceAccount` authenticator for gateway peer RPCs.

pub peer_authenticator: Option<Arc<auth::peer::PeerServiceAccountAuthenticator>>,

/// Credentials that have recently logged on, so each records one

/// authentication success rather than one per request.

pub(crate) logons: auth::audit::Logons,

/// Gateway-wide gRPC request rate limiter shared by every multiplex path.

pub(crate) grpc_rate_limiter: Option<multiplex::GrpcRateLimiter>,

/// Immutable public reflection index and its per-query rate limiter.

pub(crate) reflection_service: reflection::GatewayReflectionServer,

/// Per-sandbox bound on extension credential minting, which resolves the

/// caller's effective policy on every request.

pub(crate) extension_mint_limiter: auth::extension_mint_limit::ExtensionMintLimiter,

/// Immutable gateway interceptor execution plan. `None` when disabled.

pub(crate) gateway_interceptors:

Option<openshell_gateway_interceptors::GatewayInterceptorRuntime>,

/// Gateway-local provider profile sources. User-imported profiles are read

/// on demand when the user source is configured.

pub(crate) provider_profile_sources: provider_profile_sources::ProviderProfileSources,

/// OIDC admin role name for workspace-level authorization.

/// Empty when OIDC is not configured — `authorize_workspace()` treats

/// every authenticated user as Platform Admin in that case.

pub admin_role: String,

}

fn is_benign_tls_handshake_failure(error: &std::io::Error) -> bool {

matches!(

error.kind(),

ErrorKind::UnexpectedEof | ErrorKind::ConnectionReset

)

}

const METRICS_TLS_HANDSHAKE_WARNING_INTERVAL: Duration = Duration::from_mins(1);

#[derive(Default)]

struct MetricsTlsHandshakeLogState {

last_warning: Option<Instant>,

suppressed: u64,

}

struct MetricsTlsHandshakeLogLimiter {

interval: Duration,

state: Mutex<MetricsTlsHandshakeLogState>,

rejected: AtomicU64,

}

impl MetricsTlsHandshakeLogLimiter {

fn new(interval: Duration) -> Self {

Self {

interval,

state: Mutex::new(MetricsTlsHandshakeLogState::default()),

rejected: AtomicU64::new(0),

}

}

fn record(&self, now: Instant) -> Option<u64> {

self.rejected.fetch_add(1, Ordering::Relaxed);

let mut state = self

.state

.lock()

.unwrap_or_else(std::sync::PoisonError::into_inner);

if state

.last_warning

.is_none_or(|last_warning| now.duration_since(last_warning) >= self.interval)

{

state.last_warning = Some(now);

return Some(std::mem::take(&mut state.suppressed));

}

state.suppressed += 1;

None

}

}

/// Separate limiters for handshake rejections and timeouts so a burst of one

/// kind cannot absorb the other's warnings, and each periodic WARN reports

/// counts for a single event type.

#[derive(Clone)]

struct MetricsTlsHandshakeLogLimiters {

rejections: Arc<MetricsTlsHandshakeLogLimiter>,

timeouts: Arc<MetricsTlsHandshakeLogLimiter>,

}

impl MetricsTlsHandshakeLogLimiters {

fn new(interval: Duration) -> Self {

Self {

rejections: Arc::new(MetricsTlsHandshakeLogLimiter::new(interval)),

timeouts: Arc::new(MetricsTlsHandshakeLogLimiter::new(interval)),

}

}

}

fn log_metrics_tls_handshake_rejection(

limiter: &MetricsTlsHandshakeLogLimiter,

error: &std::io::Error,

client: SocketAddr,

) {

debug!(

event = "metrics_tls_handshake_rejected",

error = %error,

client = %client,

"Rejected metrics TLS handshake"

);

if let Some(suppressed) = limiter.record(Instant::now()) {

warn!(

event = "metrics_tls_handshake_rejected",

suppressed,

total = limiter.rejected.load(Ordering::Relaxed),

"Rejected metrics TLS handshakes; verify that clients use HTTPS and, when client authentication is enabled, certificates trusted by the metrics client CA"

);

}

}

fn log_metrics_tls_handshake_timeout(limiter: &MetricsTlsHandshakeLogLimiter, client: SocketAddr) {

debug!(

event = "metrics_tls_handshake_timed_out",

client = %client,

"Metrics TLS handshake timed out"

);

if let Some(suppressed) = limiter.record(Instant::now()) {

warn!(

event = "metrics_tls_handshake_timed_out",

suppressed,

total = limiter.rejected.load(Ordering::Relaxed),

"Metrics TLS handshakes timed out"

);

}

}

fn metrics_tls_acceptor_or_disable(

metrics_tls: &openshell_core::MetricsTlsConfig,

) -> Option<TlsAcceptor> {

match TlsAcceptor::from_files(

&metrics_tls.cert_path,

&metrics_tls.key_path,

metrics_tls.client_ca_path.as_deref(),

metrics_tls.require_client_auth,

None,

None,

Vec::new(),

) {

Ok(acceptor) => Some(acceptor),

Err(error) => {

error!(

event = "metrics_tls_listener_disabled",

listener = "metrics",

error = %error,

"Metrics TLS listener disabled because its certificate material could not be loaded"

);

let event = ConfigStateChangeBuilder::new(&gateway_ocsf::context("", ""))

.severity(SeverityId::Medium)

.status(StatusId::Failure)

.state(StateId::Disabled, "initialization_failed")

.message(format!(

"metrics TLS listener disabled because certificate material could not be loaded: {error}"

))

.build();

openshell_ocsf::ocsf_emit!(event);

None

}

}

}

fn is_benign_connection_close(error: &(dyn std::error::Error + 'static)) -> bool {

openshell_core::transport_errors::is_expected_transport_close_error(error)

}

impl ServerState {

/// Create new server state.

#[must_use]

#[allow(clippy::too_many_arguments)]

pub fn new(

config: Config,

store: Arc<Store>,

compute: ComputeRuntime,

sandbox_index: SandboxIndex,

sandbox_watch_bus: SandboxWatchBus,

tracing_log_bus: TracingLogBus,

supervisor_sessions: Arc<supervisor_session::SupervisorSessionRegistry>,

oidc_cache: Option<Arc<auth::oidc::JwksCache>>,

) -> Self {

let credentials =

credentials::CredentialRuntime::from_config_with_store(&config, Arc::clone(&store))

.expect("server config should be validated before ServerState::new");

Self::new_with_credentials(

config,

store,

compute,

sandbox_index,

sandbox_watch_bus,

tracing_log_bus,

supervisor_sessions,

oidc_cache,

credentials,

)

}

/// Create new server state with an already-initialized credential runtime.

#[must_use]

#[allow(clippy::too_many_arguments)]

pub fn new_with_credentials(

config: Config,

store: Arc<Store>,

compute: ComputeRuntime,

sandbox_index: SandboxIndex,

sandbox_watch_bus: SandboxWatchBus,

tracing_log_bus: TracingLogBus,

supervisor_sessions: Arc<supervisor_session::SupervisorSessionRegistry>,

oidc_cache: Option<Arc<auth::oidc::JwksCache>>,

credentials: credentials::CredentialRuntime,

) -> Self {

let replica_id = compute::lease::replica_id();

let compute = compute.with_sandbox_updates(config.enable_sandbox_updates);

compute.configure_ssh_identities(credentials.clone());

let service_route_index = Arc::new(service_routing::ServiceRouteIndex::new(

config.service_routing_discriminator().map(str::to_string),

));

compute.configure_service_route_index(Arc::clone(&service_route_index));

let peer_endpoint = derive_peer_endpoint(&config);

let grpc_rate_limiter = multiplex::GrpcRateLimiter::from_config(&config);

let reflection_service = reflection::build_gateway_reflection_service(&config)

.expect("compiled public gateway descriptors must be valid");

let admin_role = config

.oidc

.as_ref()

.map_or_else(String::new, |oidc| oidc.admin_role.clone());

let config_delivery =

config_delivery::ConfigDelivery::for_db_connections(store.max_connections());

let refresh = provider_refresh::runtime(&store, Some(&credentials), Some(&compute));

Self {

config,

store,

compute,

credentials,

refresh,

sandbox_index,

sandbox_watch_bus,

tracing_log_bus,

telemetry: telemetry::TelemetryState::new(),

ssh_connections_by_token: Mutex::new(HashMap::new()),

ssh_connections_by_sandbox: Mutex::new(HashMap::new()),

service_connections_by_sandbox: Mutex::new(HashMap::new()),

supervisor_sessions,

gateway_shutting_down: AtomicBool::new(false),

config_delivery,

replica_id,

peer_endpoint,

peer_routes: Arc::new(supervisor_session::PeerRouteCache::default()),

service_upstreams: Arc::new(service_routing::ServiceUpstreamPool::default()),

service_route_index,

gateway_ring: Arc::new(std::sync::RwLock::new(gateway_ring::GatewayRing::default())),

gateway_peers: Arc::new(std::sync::RwLock::new(HashMap::new())),

extension_mint_limiter: auth::extension_mint_limit::ExtensionMintLimiter::default(),

middleware_registry: Arc::new(MiddlewareRegistry::default()),

oidc_cache,

extension_jwt_issuer: None,

sandbox_session_jwt_authority: None,

compute_driver_authenticator: None,

peer_authenticator: None,

logons: auth::audit::Logons::default(),

grpc_rate_limiter,

reflection_service,

gateway_interceptors: None,

provider_profile_sources:

provider_profile_sources::ProviderProfileSources::with_default_sources(),

admin_role,

}

}

}

fn derive_peer_endpoint(config: &Config) -> Option<String> {

if let Ok(endpoint) = std::env::var("OPENSHELL_PEER_ENDPOINT")

&& !endpoint.trim().is_empty()

{

return Some(endpoint.trim().to_string());

}

let pod_name = std::env::var("OPENSHELL_POD_NAME").ok()?;

let namespace = std::env::var("OPENSHELL_POD_NAMESPACE").ok()?;

let service = std::env::var("OPENSHELL_PEER_SERVICE_NAME").ok()?;

if pod_name.trim().is_empty() || namespace.trim().is_empty() || service.trim().is_empty() {

return None;

}

let scheme = if config.tls.is_some() {

"https"

} else {

"http"

};

Some(format!(

"{scheme}://{pod}.{service}.{namespace}.svc.cluster.local:{port}",

pod = pod_name.trim(),

service = service.trim(),

namespace = namespace.trim(),

port = config.bind_address.port()

))

}

/// Reject a plaintext peer endpoint on a gateway that serves TLS.

///

/// Peer relay traffic carries whole supervisor sessions between replicas. The

/// chart renders a plaintext peer endpoint only when the gateway itself serves

fn validate_peer_endpoint_scheme(config: &Config, peer_endpoint: &str) -> Result<()> {

if peer_endpoint.starts_with("https://") {

return Ok(());

}

if config.tls.is_some() {

return Err(Error::config(format!(

"gateway peer endpoint {peer_endpoint} is plaintext but this gateway serves TLS; \

set an https:// OPENSHELL_PEER_ENDPOINT so peer relay traffic is not downgraded"

)));

}

warn!(

peer_endpoint,

"gateway peer relay traffic is plaintext because this gateway does not serve TLS"

);

Ok(())

}

/// Run the `OpenShell` server.

///

/// This starts a multiplexed gRPC/HTTP server on the configured bind address.

///

/// # Errors

///

/// Returns an error if the server fails to start or encounters a fatal error.

pub(crate) async fn run_server(

startup: ServerStartupConfig,

tracing_log_bus: TracingLogBus,

compute_drivers: ComputeDriverRegistry,

) -> Result<()> {

let ServerStartupConfig {

config,

config_file,

guest_tls,

network_trust,

compute_driver,

legacy_compute_driver_env_seen: _,

} = startup;

let (shutdown_tx, shutdown_rx) = watch::channel(false);

auth::descriptor_authz::init()

.map_err(|error| Error::config(format!("invalid gRPC authorization metadata: {error}")))?;

let database_url = config.database_url.trim();

if database_url.is_empty() {

return Err(Error::config("database_url is required"));

}

// Load signing material before connecting remote extensions so their

// startup Describe calls can authenticate with gateway-caller tokens.

let (extension_jwt_issuer, sandbox_session_jwt_authority) =

if let Some(ref jwt) = config.gateway_jwt {

let authorities = auth::launch_signing::load(jwt)?;

info!(

gateway_id = %jwt.gateway_id,

ttl_secs = jwt.ttl_secs.map(std::num::NonZeroU64::get),

"gateway-minted sandbox JWT enabled"

);

(

Some(authorities.extension),

Some(authorities.sandbox_session),

)

} else {

(None, None)

};

let middleware_registrations = config_file

.as_ref()

.map(|file| {

file.openshell

.supervisor

.middleware

.iter()

.map(openshell_core::proto::SupervisorMiddlewareService::try_from)

.collect::<std::result::Result<Vec<_>, _>>()

})

.transpose()

.map_err(|error| Error::config(format!("middleware registration failed: {error}")))?

.unwrap_or_default();

let mut gateway_extension_credentials = Vec::new();

let middleware_registry = Arc::new(

if let Some(issuer) = extension_jwt_issuer.as_ref() {

let mut slots = HashMap::new();

for registration in &middleware_registrations {

if let Some(credential) = mint_gateway_extension_credential(

issuer,

ExtensionKind::Middleware,

&registration.name,

&registration.audience,

&registration.grpc_endpoint,

registration.allow_insecure_transport,

)? {

slots.insert(registration.name.clone(), credential.slot.clone());

gateway_extension_credentials.push(credential);

}

}

MiddlewareRegistry::connect_services_authenticated(

openshell_supervisor_middleware_builtins::services(),

middleware_registrations,

&slots,

)

.await

} else {

MiddlewareRegistry::connect_services(

openshell_supervisor_middleware_builtins::services(),

middleware_registrations,

)

.await

}

.map_err(|error| Error::config(format!("middleware registration failed: {error}")))?,

);

let shutdown = shutdown_signal();

tokio::pin!(shutdown);

let mut health_listener = HealthListener::new(config.health_bind_address);

let Some(store) = connect_store(

&config,

database_url,

&mut health_listener,

shutdown.as_mut(),

)

.await?

else {

info!("Shutdown signal received while waiting for the database; stopping gateway");

return Ok(());

};

let store = Arc::new(store);

let credentials = credentials::CredentialRuntime::from_config_file_with_store(

&config,

config_file.as_ref(),

Arc::clone(&store),

)

.await?;

let oidc_cache = if let Some(ref oidc) = config.oidc {

// Validate RBAC configuration before starting.

let policy = auth::authz::AuthzPolicy {

admin_role: oidc.admin_role.clone(),

user_role: oidc.user_role.clone(),

scopes_enabled: !oidc.scopes_claim.is_empty(),

};

policy.validate().map_err(Error::config)?;

let cache = auth::oidc::JwksCache::new(oidc)

.await

.map_err(|e| Error::config(format!("OIDC initialization failed: {e}")))?;

info!("OIDC JWT validation enabled (issuer: {})", oidc.issuer);

Some(Arc::new(cache))

} else {

None

};

let sandbox_index = SandboxIndex::new();

let sandbox_watch_bus = SandboxWatchBus::new();

let supervisor_sessions = Arc::new(supervisor_session::SupervisorSessionRegistry::new());

let driver_startup = compute::driver_config::DriverStartupContext {

file: config_file.as_ref(),

guest_tls: guest_tls.as_ref(),

network_trust: network_trust.as_ref(),

network_trust_configured: network_trust.is_some(),

gateway_port: config.bind_address.port(),

gateway_tls_enabled: config.tls.is_some(),

endpoint_overrides: &config.compute_driver_endpoints,

};

let compute = build_compute_runtime(

&compute_drivers,

&compute_driver,

&config,

driver_startup,

store.clone(),

sandbox_index.clone(),

sandbox_watch_bus.clone(),

tracing_log_bus.clone(),

supervisor_sessions.clone(),

shutdown_rx.clone(),

)

.await?;

compute.configure_supervisor_otlp_headers(supervisor_otlp_headers(

config_file

.as_ref()

.and_then(|file| file.openshell.gateway.otlp.as_ref()),

));

compute.configure_supervisor_otlp_tls(otel_tracing::supervisor_otlp_tls(

config_file

.as_ref()

.and_then(|file| file.openshell.gateway.otlp.as_ref()),

));

let gateway_interceptors = if let Some(issuer) = extension_jwt_issuer.as_ref() {

let mut slots = BTreeMap::new();

for interceptor in &config.gateway_interceptors {

let audience = interceptor.resolved_audience();

if let Some(credential) = mint_gateway_extension_credential(

issuer,

ExtensionKind::Interceptor,

&interceptor.name,

audience.as_ref(),

&interceptor.grpc_endpoint,

interceptor.allow_insecure_transport,

)? {

slots.insert(interceptor.name.clone(), credential.slot.clone());

gateway_extension_credentials.push(credential);

}

}

openshell_gateway_interceptors::initialize_authenticated(

config.gateway_interceptors.clone(),

slots,

)

.await

} else {

openshell_gateway_interceptors::initialize(config.gateway_interceptors.clone()).await

}

.map_err(|e| Error::config(format!("gateway interceptor initialization failed: {e}")))?;

// Push delivery needs the refresh to notice catalog changes, so only push

// mode serves interceptor profiles from a cached snapshot.

let provider_profile_sources = provider_profile_sources::ProviderProfileSources::from_config(

&config.provider_profile_sources,

gateway_interceptors.as_ref(),

(config.config_delivery_mode == openshell_core::config::ConfigDeliveryMode::Push)

.then_some(provider_profile_sources::PROVIDER_PROFILE_SOURCE_REFRESH_INTERVAL),

)

.map_err(|err| {

Error::config(format!(

"provider profile source configuration failed: {err}"

))

})?;

info!(

sources = ?provider_profile_sources.source_ids(),

"provider profile sources configured"

);

// Builds read cached interceptor snapshots, so fetch them before serving.

provider_profile_sources.refresh_cached(&store).await;

let mut state = ServerState::new_with_credentials(

config.clone(),

store.clone(),

compute,

sandbox_index,

sandbox_watch_bus,

tracing_log_bus,

supervisor_sessions,

oidc_cache,

credentials,

);

state.middleware_registry = middleware_registry;

state.gateway_interceptors = gateway_interceptors;

state.provider_profile_sources = provider_profile_sources;

state.extension_jwt_issuer = extension_jwt_issuer.clone();

state.sandbox_session_jwt_authority = sandbox_session_jwt_authority;

if let Some(issuer) = extension_jwt_issuer {

spawn_gateway_extension_token_refresh(issuer, gateway_extension_credentials);

}

// The shared supervisor authenticates with a token this gateway mints

// from its launch-signing key and carries only over TLS, so the topology

// needs both.

if let Some(controller) = state.compute.supervisor_host() {

let (authority, jwt) = state

.sandbox_session_jwt_authority

.clone()

.zip(config.gateway_jwt.as_ref())

.ok_or_else(|| {

Error::config(

"the shared supervisor requires gateway_jwt signing material to mint \

supervisor tokens; configure [openshell.gateway.gateway_jwt] or disable \

supervisor_topology",

)

})?;

if config.tls.is_none() {

return Err(Error::config(

"the shared supervisor requires gateway TLS; its token travels only over TLS. \

Enable TLS or disable supervisor_topology",

));

}

let ttl = jwt.token_ttl().clamp(

openshell_core::jwt::MIN_SESSION_TOKEN_TTL,

openshell_core::jwt::MAX_SESSION_TOKEN_TTL,

);

info!(

ttl_secs = ttl.as_secs(),

"shared supervisor assignment stream enabled; minting supervisor tokens"