From 27f154d443cfb5371c494f8fa7544f33c3a8aada Mon Sep 17 00:00:00 2001 From: Evan Lezar Date: Mon, 27 Jul 2026 12:43:34 +0200 Subject: [PATCH 1/5] refactor(server): isolate gateway listener context Signed-off-by: Evan Lezar --- .../openshell-server/src/gateway_listener.rs | 153 ++++++++++++++++++ crates/openshell-server/src/lib.rs | 102 +++--------- crates/openshell-server/src/multiplex.rs | 114 ++++++++++++- 3 files changed, 285 insertions(+), 84 deletions(-) create mode 100644 crates/openshell-server/src/gateway_listener.rs diff --git a/crates/openshell-server/src/gateway_listener.rs b/crates/openshell-server/src/gateway_listener.rs new file mode 100644 index 000000000..20f32fe81 --- /dev/null +++ b/crates/openshell-server/src/gateway_listener.rs @@ -0,0 +1,153 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use openshell_core::{Error, Result}; +use std::net::SocketAddr; +use tokio::net::TcpListener; +use tracing::info; + +/// Authorization scope associated with a gateway listener. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum GatewayListenerScope { + Primary, + ComputeDriverCallback, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct GatewayListenerSpec { + address: SocketAddr, + scope: GatewayListenerScope, +} + +/// A gateway listener together with the context needed to serve it. +pub struct BoundGatewayListener { + pub listener: TcpListener, + pub address: SocketAddr, + pub scope: GatewayListenerScope, +} + +fn gateway_listener_specs( + bind_address: SocketAddr, + extra_addresses: &[SocketAddr], +) -> Vec { + let mut specs = vec![GatewayListenerSpec { + address: bind_address, + scope: GatewayListenerScope::Primary, + }]; + for address in extra_addresses { + if !specs + .iter() + .any(|existing| listener_covers(existing.address, *address)) + { + specs.push(GatewayListenerSpec { + address: *address, + scope: GatewayListenerScope::ComputeDriverCallback, + }); + } + } + specs +} + +pub async fn bind_gateway_listeners( + bind_address: SocketAddr, + extra_addresses: &[SocketAddr], +) -> Result> { + let specs = gateway_listener_specs(bind_address, extra_addresses); + let mut listeners = Vec::with_capacity(specs.len()); + for spec in specs { + let listener = TcpListener::bind(spec.address) + .await + .map_err(|e| Error::transport(format!("failed to bind to {}: {e}", spec.address)))?; + let local_addr = listener.local_addr().unwrap_or(spec.address); + info!(address = %local_addr, "Server listening"); + listeners.push(BoundGatewayListener { + listener, + address: local_addr, + scope: spec.scope, + }); + } + Ok(listeners) +} + +fn listener_covers(existing: SocketAddr, requested: SocketAddr) -> bool { + if existing == requested { + return true; + } + if existing.port() != requested.port() { + return false; + } + + match (existing.ip(), requested.ip()) { + (std::net::IpAddr::V4(existing), std::net::IpAddr::V4(_)) => existing.is_unspecified(), + (std::net::IpAddr::V6(existing), std::net::IpAddr::V6(_)) => existing.is_unspecified(), + _ => false, + } +} + +#[cfg(test)] +mod tests { + use super::{ + GatewayListenerScope, GatewayListenerSpec, bind_gateway_listeners, gateway_listener_specs, + }; + use std::net::SocketAddr; + use std::sync::atomic::{AtomicBool, Ordering}; + use tokio::net::TcpListener; + + #[test] + fn gateway_listener_specs_skip_driver_address_covered_by_wildcard() { + let primary: SocketAddr = "0.0.0.0:8080".parse().unwrap(); + let docker: SocketAddr = "172.18.0.1:8080".parse().unwrap(); + + assert_eq!( + gateway_listener_specs(primary, &[docker, docker]), + vec![GatewayListenerSpec { + address: primary, + scope: GatewayListenerScope::Primary, + }] + ); + } + + #[test] + fn gateway_listener_specs_preserve_driver_callback_scope() { + let primary: SocketAddr = "127.0.0.1:8080".parse().unwrap(); + let docker: SocketAddr = "172.18.0.1:8080".parse().unwrap(); + + assert_eq!( + gateway_listener_specs(primary, &[docker, docker]), + vec![ + GatewayListenerSpec { + address: primary, + scope: GatewayListenerScope::Primary, + }, + GatewayListenerSpec { + address: docker, + scope: GatewayListenerScope::ComputeDriverCallback, + }, + ] + ); + } + + #[tokio::test] + async fn failed_bind_does_not_return_partially_bound_listeners() { + let occupied_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let occupied_address = occupied_listener.local_addr().unwrap(); + let continuation_reached = AtomicBool::new(false); + let primary_address: SocketAddr = "127.0.0.1:0".parse().unwrap(); + + let result: openshell_core::Result<()> = async { + let _listeners = bind_gateway_listeners(primary_address, &[occupied_address]).await?; + continuation_reached.store(true, Ordering::SeqCst); + Ok(()) + } + .await; + + assert!( + result.is_err(), + "binding the occupied extra gateway address should fail" + ); + assert!( + !continuation_reached.load(Ordering::SeqCst), + "binding must fail before returning a partial listener set" + ); + } +} diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index 525bbd375..96b4685fa 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -29,6 +29,7 @@ pub mod cli; mod compute; pub mod config_file; mod defaults; +mod gateway_listener; mod grpc; mod http; mod inference; @@ -71,6 +72,7 @@ use tracing::{debug, error, info, warn}; pub(crate) static TEST_ENV_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); use compute::ComputeRuntime; +use gateway_listener::{GatewayListenerScope, bind_gateway_listeners}; pub use grpc::OpenShellService; pub use http::{health_router, http_router, metrics_router, service_http_router}; pub use multiplex::{MultiplexService, MultiplexedService}; @@ -522,10 +524,11 @@ pub(crate) async fn run_server( let mut listener_tasks = Vec::with_capacity(gateway_listeners.len()); let enable_loopback_service_http = config.service_routing.enable_loopback_service_http; - for (listener, listen_addr) in gateway_listeners { + for listener in gateway_listeners { listener_tasks.push(tokio::spawn(serve_gateway_listener( - listener, - listen_addr, + listener.listener, + listener.address, + listener.scope, service.clone(), tls_acceptor.clone(), enable_loopback_service_http, @@ -552,57 +555,10 @@ pub(crate) async fn run_server( Ok(()) } -fn gateway_listener_addresses( - bind_address: SocketAddr, - extra_addresses: &[SocketAddr], -) -> Vec { - let mut addresses = vec![bind_address]; - for address in extra_addresses { - if !addresses - .iter() - .any(|existing| listener_covers(*existing, *address)) - { - addresses.push(*address); - } - } - addresses -} - -async fn bind_gateway_listeners( - bind_address: SocketAddr, - extra_addresses: &[SocketAddr], -) -> Result> { - let addresses = gateway_listener_addresses(bind_address, extra_addresses); - let mut listeners = Vec::with_capacity(addresses.len()); - for address in addresses { - let listener = TcpListener::bind(address) - .await - .map_err(|e| Error::transport(format!("failed to bind to {address}: {e}")))?; - let local_addr = listener.local_addr().unwrap_or(address); - info!(address = %local_addr, "Server listening"); - listeners.push((listener, local_addr)); - } - Ok(listeners) -} - -fn listener_covers(existing: SocketAddr, requested: SocketAddr) -> bool { - if existing == requested { - return true; - } - if existing.port() != requested.port() { - return false; - } - - match (existing.ip(), requested.ip()) { - (std::net::IpAddr::V4(existing), std::net::IpAddr::V4(_)) => existing.is_unspecified(), - (std::net::IpAddr::V6(existing), std::net::IpAddr::V6(_)) => existing.is_unspecified(), - _ => false, - } -} - async fn serve_gateway_listener( listener: TcpListener, listen_addr: SocketAddr, + scope: GatewayListenerScope, service: MultiplexService, tls_acceptor: Option, enable_loopback_service_http: bool, @@ -631,6 +587,7 @@ async fn serve_gateway_listener( stream, addr, listen_addr, + scope, service.clone(), tls_acceptor.clone(), enable_loopback_service_http, @@ -699,6 +656,7 @@ fn spawn_gateway_connection( stream: TcpStream, addr: SocketAddr, listen_addr: SocketAddr, + listener_scope: GatewayListenerScope, service: MultiplexService, tls_acceptor: Option, enable_loopback_service_http: bool, @@ -713,7 +671,10 @@ fn spawn_gateway_connection( addr, ) => { - if let Err(e) = service.serve_service_http(stream).await { + if let Err(e) = service + .serve_service_http_on_listener(stream, listener_scope) + .await + { if is_benign_connection_close(e.as_ref()) { debug!(error = %e, client = %addr, listen = %listen_addr, "Plaintext service HTTP connection closed"); } else { @@ -732,7 +693,11 @@ fn spawn_gateway_connection( Ok(tls_stream) => { let peer_identity = multiplex::extract_peer_identity(&tls_stream); if let Err(e) = service - .serve_with_peer_identity(tls_stream, peer_identity) + .serve_with_peer_identity_on_listener( + tls_stream, + peer_identity, + listener_scope, + ) .await { if is_benign_connection_close(e.as_ref()) { @@ -758,7 +723,7 @@ fn spawn_gateway_connection( }); } else { tokio::spawn(async move { - if let Err(e) = service.serve(stream).await { + if let Err(e) = service.serve_on_listener(stream, listener_scope).await { if is_benign_connection_close(e.as_ref()) { debug!(error = %e, client = %addr, "Connection closed"); } else { @@ -1035,9 +1000,9 @@ pub(crate) async fn ensure_default_workspace(store: &Store) -> Result<()> { #[cfg(test)] mod tests { use super::{ - ConfiguredComputeDriver, ConnectionProtocol, MultiplexService, ServerState, TlsAcceptor, - allow_plaintext_service_http, bind_gateway_listeners, classify_initial_bytes, - configured_compute_driver, gateway_listener_addresses, is_benign_tls_handshake_failure, + ConfiguredComputeDriver, ConnectionProtocol, GatewayListenerScope, MultiplexService, + ServerState, TlsAcceptor, allow_plaintext_service_http, bind_gateway_listeners, + classify_initial_bytes, configured_compute_driver, is_benign_tls_handshake_failure, kubernetes_sandbox_jwt_expiry_disabled, serve_gateway_listener, }; use openshell_core::{ @@ -1134,6 +1099,7 @@ mod tests { let handle = tokio::spawn(serve_gateway_listener( listener, listen_addr, + GatewayListenerScope::Primary, service, Some(tls_acceptor), enable_loopback_service_http, @@ -1511,28 +1477,6 @@ mod tests { assert!(!kubernetes_sandbox_jwt_expiry_disabled(&Config::new(None))); } - #[test] - fn gateway_listener_addresses_skip_driver_address_covered_by_wildcard() { - let primary: SocketAddr = "0.0.0.0:8080".parse().unwrap(); - let docker: SocketAddr = "172.18.0.1:8080".parse().unwrap(); - - assert_eq!( - gateway_listener_addresses(primary, &[docker, docker]), - vec![primary] - ); - } - - #[test] - fn gateway_listener_addresses_include_driver_address_on_distinct_ip() { - let primary: SocketAddr = "127.0.0.1:8080".parse().unwrap(); - let docker: SocketAddr = "172.18.0.1:8080".parse().unwrap(); - - assert_eq!( - gateway_listener_addresses(primary, &[docker, docker]), - vec![primary, docker] - ); - } - #[tokio::test] async fn failed_gateway_listener_bind_does_not_attempt_persisted_sandbox_resume() { let occupied_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); diff --git a/crates/openshell-server/src/multiplex.rs b/crates/openshell-server/src/multiplex.rs index 559fdc6c7..44ec231d2 100644 --- a/crates/openshell-server/src/multiplex.rs +++ b/crates/openshell-server/src/multiplex.rs @@ -41,6 +41,7 @@ use crate::{ auth::identity::Identity, auth::oidc::{self, OidcAuthenticator}, auth::principal::{Principal, UserPrincipal}, + gateway_listener::GatewayListenerScope, http_router, inference::InferenceService, service_http_router, @@ -144,7 +145,22 @@ impl MultiplexService { where S: AsyncRead + AsyncWrite + Unpin + Send + 'static, { - self.serve_with_peer_identity(stream, None).await + self.serve_on_listener(stream, GatewayListenerScope::Primary) + .await + } + + /// Serve a connection and preserve its listener purpose in request + /// extensions for downstream routing and policy decisions. + pub(crate) async fn serve_on_listener( + &self, + stream: S, + listener_scope: GatewayListenerScope, + ) -> Result<(), Box> + where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, + { + self.serve_with_peer_identity_on_listener(stream, None, listener_scope) + .await } /// Serve a TLS connection with an optional mTLS peer identity. @@ -153,6 +169,25 @@ impl MultiplexService { stream: S, peer_identity: Option, ) -> Result<(), Box> + where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, + { + self.serve_with_peer_identity_on_listener( + stream, + peer_identity, + GatewayListenerScope::Primary, + ) + .await + } + + /// Serve a TLS connection and preserve its listener purpose in request + /// extensions for downstream routing and policy decisions. + pub(crate) async fn serve_with_peer_identity_on_listener( + &self, + stream: S, + peer_identity: Option, + listener_scope: GatewayListenerScope, + ) -> Result<(), Box> where S: AsyncRead + AsyncWrite + Unpin + Send + 'static, { @@ -188,7 +223,10 @@ impl MultiplexService { let grpc_service = request_id_middleware!(grpc_service); let http_service = request_id_middleware!(http_service); - let service = MultiplexedService::new(grpc_service, http_service); + let service = GatewayListenerContextService::new( + MultiplexedService::new(grpc_service, http_service), + listener_scope, + ); let mut builder = Builder::new(TokioExecutor::new()); // Server-side HTTP/2 keepalive: supervisors hold long-lived sessions, and without @@ -217,9 +255,26 @@ impl MultiplexService { where S: AsyncRead + AsyncWrite + Unpin + Send + 'static, { - let http_service = TowerToHyperService::new(request_id_middleware!(service_http_router( - self.state.clone() - ))); + self.serve_service_http_on_listener(stream, GatewayListenerScope::Primary) + .await + } + + /// Serve a plaintext service HTTP connection and preserve its listener + /// purpose in request extensions. + pub(crate) async fn serve_service_http_on_listener( + &self, + stream: S, + listener_scope: GatewayListenerScope, + ) -> Result<(), Box> + where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, + { + let http_service = GatewayListenerContextService::new( + TowerToHyperService::new(request_id_middleware!(service_http_router( + self.state.clone() + ))), + listener_scope, + ); Builder::new(TokioExecutor::new()) .serve_connection_with_upgrades(TokioIo::new(stream), http_service) @@ -229,6 +284,36 @@ impl MultiplexService { } } +/// Adds the immutable listener authorization scope to every served request. +#[derive(Clone)] +struct GatewayListenerContextService { + inner: S, + listener_scope: GatewayListenerScope, +} + +impl GatewayListenerContextService { + fn new(inner: S, listener_scope: GatewayListenerScope) -> Self { + Self { + inner, + listener_scope, + } + } +} + +impl hyper::service::Service> for GatewayListenerContextService +where + S: hyper::service::Service>, +{ + type Response = S::Response; + type Error = S::Error; + type Future = S::Future; + + fn call(&self, mut request: Request) -> Self::Future { + request.extensions_mut().insert(self.listener_scope); + self.inner.call(request) + } +} + /// `OpenShell` gRPC wrapper that applies configured gateway interceptors before /// tonic dispatches to a specific RPC handler. #[derive(Clone)] @@ -1101,6 +1186,25 @@ mod tests { use tokio_stream::wrappers::TcpListenerStream; use tower::Service; + #[tokio::test] + async fn listener_context_service_preserves_listener_scope() { + let observed = Arc::new(Mutex::new(None)); + let captured = observed.clone(); + let inner = hyper::service::service_fn(move |request: Request>| { + *captured.lock().unwrap() = request.extensions().get::().copied(); + async move { Ok::<_, Infallible>(Response::new(Empty::::new())) } + }); + let service = GatewayListenerContextService::new(inner, GatewayListenerScope::Primary); + hyper::service::Service::call(&service, Request::new(Empty::::new())) + .await + .unwrap(); + + assert_eq!( + *observed.lock().unwrap(), + Some(GatewayListenerScope::Primary) + ); + } + #[derive(Clone)] struct PostCommitTestInterceptor; From 4c226e46559cd381baf4a14d24f4b81f6dc5b15d Mon Sep 17 00:00:00 2001 From: Evan Lezar Date: Wed, 29 Jul 2026 17:20:58 +0200 Subject: [PATCH 2/5] refactor(server): preserve gateway listener binding Signed-off-by: Evan Lezar --- crates/openshell-server/src/lib.rs | 33 +++++++++++++++++------------- 1 file changed, 19 insertions(+), 14 deletions(-) diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index 96b4685fa..adb6f8a36 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -72,7 +72,7 @@ use tracing::{debug, error, info, warn}; pub(crate) static TEST_ENV_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); use compute::ComputeRuntime; -use gateway_listener::{GatewayListenerScope, bind_gateway_listeners}; +use gateway_listener::{BoundGatewayListener, GatewayListenerScope, bind_gateway_listeners}; pub use grpc::OpenShellService; pub use http::{health_router, http_router, metrics_router, service_http_router}; pub use multiplex::{MultiplexService, MultiplexedService}; @@ -526,9 +526,7 @@ pub(crate) async fn run_server( let enable_loopback_service_http = config.service_routing.enable_loopback_service_http; for listener in gateway_listeners { listener_tasks.push(tokio::spawn(serve_gateway_listener( - listener.listener, - listener.address, - listener.scope, + listener, service.clone(), tls_acceptor.clone(), enable_loopback_service_http, @@ -556,14 +554,18 @@ pub(crate) async fn run_server( } async fn serve_gateway_listener( - listener: TcpListener, - listen_addr: SocketAddr, - scope: GatewayListenerScope, + bound_listener: BoundGatewayListener, service: MultiplexService, tls_acceptor: Option, enable_loopback_service_http: bool, mut shutdown: watch::Receiver, ) { + let BoundGatewayListener { + listener, + address: listen_addr, + scope, + } = bound_listener; + loop { let accepted = tokio::select! { changed = shutdown.changed() => { @@ -1000,10 +1002,11 @@ pub(crate) async fn ensure_default_workspace(store: &Store) -> Result<()> { #[cfg(test)] mod tests { use super::{ - ConfiguredComputeDriver, ConnectionProtocol, GatewayListenerScope, MultiplexService, - ServerState, TlsAcceptor, allow_plaintext_service_http, bind_gateway_listeners, - classify_initial_bytes, configured_compute_driver, is_benign_tls_handshake_failure, - kubernetes_sandbox_jwt_expiry_disabled, serve_gateway_listener, + BoundGatewayListener, ConfiguredComputeDriver, ConnectionProtocol, GatewayListenerScope, + MultiplexService, ServerState, TlsAcceptor, allow_plaintext_service_http, + bind_gateway_listeners, classify_initial_bytes, configured_compute_driver, + is_benign_tls_handshake_failure, kubernetes_sandbox_jwt_expiry_disabled, + serve_gateway_listener, }; use openshell_core::{ ComputeDriverKind, Config, @@ -1097,9 +1100,11 @@ mod tests { let (tls_dir, tls_acceptor) = test_tls_acceptor(); let (shutdown_tx, shutdown_rx) = watch::channel(false); let handle = tokio::spawn(serve_gateway_listener( - listener, - listen_addr, - GatewayListenerScope::Primary, + BoundGatewayListener { + listener, + address: listen_addr, + scope: GatewayListenerScope::Primary, + }, service, Some(tls_acceptor), enable_loopback_service_http, From ef92a2717f838a975ae7c3ac6bc724c8f325dccc Mon Sep 17 00:00:00 2001 From: Evan Lezar Date: Wed, 29 Jul 2026 17:54:21 +0200 Subject: [PATCH 3/5] fix(server): preserve covered listener callback scope Signed-off-by: Evan Lezar --- .../openshell-server/src/gateway_listener.rs | 106 +++++++++++++++++- crates/openshell-server/src/lib.rs | 18 ++- 2 files changed, 117 insertions(+), 7 deletions(-) diff --git a/crates/openshell-server/src/gateway_listener.rs b/crates/openshell-server/src/gateway_listener.rs index 20f32fe81..ae80bf229 100644 --- a/crates/openshell-server/src/gateway_listener.rs +++ b/crates/openshell-server/src/gateway_listener.rs @@ -14,9 +14,16 @@ pub enum GatewayListenerScope { } #[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct CoveredGatewayAddress { + pub address: SocketAddr, + pub scope: GatewayListenerScope, +} + +#[derive(Clone, Debug, Eq, PartialEq)] struct GatewayListenerSpec { address: SocketAddr, scope: GatewayListenerScope, + covered_addresses: Vec, } /// A gateway listener together with the context needed to serve it. @@ -24,6 +31,7 @@ pub struct BoundGatewayListener { pub listener: TcpListener, pub address: SocketAddr, pub scope: GatewayListenerScope, + pub covered_addresses: Vec, } fn gateway_listener_specs( @@ -33,15 +41,31 @@ fn gateway_listener_specs( let mut specs = vec![GatewayListenerSpec { address: bind_address, scope: GatewayListenerScope::Primary, + covered_addresses: Vec::new(), }]; for address in extra_addresses { - if !specs + let scope = GatewayListenerScope::ComputeDriverCallback; + if let Some(existing) = specs .iter() - .any(|existing| listener_covers(existing.address, *address)) + .position(|existing| listener_covers(existing.address, *address)) { + let existing = &mut specs[existing]; + if existing.address != *address + && !existing + .covered_addresses + .iter() + .any(|covered| covered.address == *address) + { + existing.covered_addresses.push(CoveredGatewayAddress { + address: *address, + scope, + }); + } + } else { specs.push(GatewayListenerSpec { address: *address, - scope: GatewayListenerScope::ComputeDriverCallback, + scope, + covered_addresses: Vec::new(), }); } } @@ -64,11 +88,57 @@ pub async fn bind_gateway_listeners( listener, address: local_addr, scope: spec.scope, + covered_addresses: resolve_bound_covered_addresses( + &spec.covered_addresses, + spec.address, + local_addr, + ), }); } Ok(listeners) } +pub fn gateway_listener_scope_for_local_addr( + default_scope: GatewayListenerScope, + covered_addresses: &[CoveredGatewayAddress], + local_addr: SocketAddr, +) -> GatewayListenerScope { + covered_addresses + .iter() + .find(|covered| covered.address == local_addr) + .map_or(default_scope, |covered| covered.scope) +} + +fn resolve_bound_covered_addresses( + covered_addresses: &[CoveredGatewayAddress], + requested_listener_addr: SocketAddr, + bound_listener_addr: SocketAddr, +) -> Vec { + covered_addresses + .iter() + .map(|covered| CoveredGatewayAddress { + address: resolve_ephemeral_port( + covered.address, + requested_listener_addr, + bound_listener_addr, + ), + scope: covered.scope, + }) + .collect() +} + +fn resolve_ephemeral_port( + address: SocketAddr, + requested_listener_addr: SocketAddr, + bound_listener_addr: SocketAddr, +) -> SocketAddr { + if requested_listener_addr.port() == 0 && address.port() == 0 { + SocketAddr::new(address.ip(), bound_listener_addr.port()) + } else { + address + } +} + fn listener_covers(existing: SocketAddr, requested: SocketAddr) -> bool { if existing == requested { return true; @@ -87,14 +157,15 @@ fn listener_covers(existing: SocketAddr, requested: SocketAddr) -> bool { #[cfg(test)] mod tests { use super::{ - GatewayListenerScope, GatewayListenerSpec, bind_gateway_listeners, gateway_listener_specs, + CoveredGatewayAddress, GatewayListenerScope, GatewayListenerSpec, bind_gateway_listeners, + gateway_listener_scope_for_local_addr, gateway_listener_specs, }; use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, Ordering}; use tokio::net::TcpListener; #[test] - fn gateway_listener_specs_skip_driver_address_covered_by_wildcard() { + fn gateway_listener_specs_track_driver_address_covered_by_wildcard() { let primary: SocketAddr = "0.0.0.0:8080".parse().unwrap(); let docker: SocketAddr = "172.18.0.1:8080".parse().unwrap(); @@ -103,10 +174,33 @@ mod tests { vec![GatewayListenerSpec { address: primary, scope: GatewayListenerScope::Primary, + covered_addresses: vec![CoveredGatewayAddress { + address: docker, + scope: GatewayListenerScope::ComputeDriverCallback, + }], }] ); } + #[test] + fn gateway_listener_scope_for_local_addr_uses_covered_address_scope() { + let primary: SocketAddr = "0.0.0.0:8080".parse().unwrap(); + let docker: SocketAddr = "172.18.0.1:8080".parse().unwrap(); + let loopback: SocketAddr = "127.0.0.1:8080".parse().unwrap(); + let [spec] = gateway_listener_specs(primary, &[docker]) + .try_into() + .unwrap(); + + assert_eq!( + gateway_listener_scope_for_local_addr(spec.scope, &spec.covered_addresses, docker), + GatewayListenerScope::ComputeDriverCallback, + ); + assert_eq!( + gateway_listener_scope_for_local_addr(spec.scope, &spec.covered_addresses, loopback), + GatewayListenerScope::Primary, + ); + } + #[test] fn gateway_listener_specs_preserve_driver_callback_scope() { let primary: SocketAddr = "127.0.0.1:8080".parse().unwrap(); @@ -118,10 +212,12 @@ mod tests { GatewayListenerSpec { address: primary, scope: GatewayListenerScope::Primary, + covered_addresses: Vec::new(), }, GatewayListenerSpec { address: docker, scope: GatewayListenerScope::ComputeDriverCallback, + covered_addresses: Vec::new(), }, ] ); diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index adb6f8a36..0fb60ec43 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -72,7 +72,10 @@ use tracing::{debug, error, info, warn}; pub(crate) static TEST_ENV_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); use compute::ComputeRuntime; -use gateway_listener::{BoundGatewayListener, GatewayListenerScope, bind_gateway_listeners}; +use gateway_listener::{ + BoundGatewayListener, GatewayListenerScope, bind_gateway_listeners, + gateway_listener_scope_for_local_addr, +}; pub use grpc::OpenShellService; pub use http::{health_router, http_router, metrics_router, service_http_router}; pub use multiplex::{MultiplexService, MultiplexedService}; @@ -564,6 +567,7 @@ async fn serve_gateway_listener( listener, address: listen_addr, scope, + covered_addresses, } = bound_listener; loop { @@ -584,12 +588,21 @@ async fn serve_gateway_listener( continue; } }; + let listener_scope = match stream.local_addr() { + Ok(local_addr) => { + gateway_listener_scope_for_local_addr(scope, &covered_addresses, local_addr) + } + Err(e) => { + debug!(error = %e, client = %addr, listen = %listen_addr, "Failed to inspect accepted local address"); + scope + } + }; spawn_gateway_connection( stream, addr, listen_addr, - scope, + listener_scope, service.clone(), tls_acceptor.clone(), enable_loopback_service_http, @@ -1104,6 +1117,7 @@ mod tests { listener, address: listen_addr, scope: GatewayListenerScope::Primary, + covered_addresses: Vec::new(), }, service, Some(tls_acceptor), From c4b2b688cbddc85ec7968844e816420f359a46ec Mon Sep 17 00:00:00 2001 From: Evan Lezar Date: Wed, 29 Jul 2026 18:01:54 +0200 Subject: [PATCH 4/5] refactor(server): reuse gateway listener spec Signed-off-by: Evan Lezar --- .../openshell-server/src/gateway_listener.rs | 75 ++++++++++--------- crates/openshell-server/src/lib.rs | 33 +++----- 2 files changed, 50 insertions(+), 58 deletions(-) diff --git a/crates/openshell-server/src/gateway_listener.rs b/crates/openshell-server/src/gateway_listener.rs index ae80bf229..0957c4ebe 100644 --- a/crates/openshell-server/src/gateway_listener.rs +++ b/crates/openshell-server/src/gateway_listener.rs @@ -20,29 +20,51 @@ pub struct CoveredGatewayAddress { } #[derive(Clone, Debug, Eq, PartialEq)] -struct GatewayListenerSpec { - address: SocketAddr, - scope: GatewayListenerScope, +pub struct GatewayListenerSpec { + pub address: SocketAddr, + pub scope: GatewayListenerScope, covered_addresses: Vec, } /// A gateway listener together with the context needed to serve it. pub struct BoundGatewayListener { pub listener: TcpListener, - pub address: SocketAddr, - pub scope: GatewayListenerScope, - pub covered_addresses: Vec, + pub spec: GatewayListenerSpec, +} + +impl GatewayListenerSpec { + pub fn new(address: SocketAddr, scope: GatewayListenerScope) -> Self { + Self { + address, + scope, + covered_addresses: Vec::new(), + } + } + + pub fn scope_for_local_addr(&self, local_addr: SocketAddr) -> GatewayListenerScope { + self.covered_addresses + .iter() + .find(|covered| covered.address == local_addr) + .map_or(self.scope, |covered| covered.scope) + } + + fn bind_to(mut self, local_addr: SocketAddr) -> Self { + let requested_addr = self.address; + self.address = local_addr; + self.covered_addresses = + resolve_bound_covered_addresses(&self.covered_addresses, requested_addr, local_addr); + self + } } fn gateway_listener_specs( bind_address: SocketAddr, extra_addresses: &[SocketAddr], ) -> Vec { - let mut specs = vec![GatewayListenerSpec { - address: bind_address, - scope: GatewayListenerScope::Primary, - covered_addresses: Vec::new(), - }]; + let mut specs = vec![GatewayListenerSpec::new( + bind_address, + GatewayListenerScope::Primary, + )]; for address in extra_addresses { let scope = GatewayListenerScope::ComputeDriverCallback; if let Some(existing) = specs @@ -62,11 +84,7 @@ fn gateway_listener_specs( }); } } else { - specs.push(GatewayListenerSpec { - address: *address, - scope, - covered_addresses: Vec::new(), - }); + specs.push(GatewayListenerSpec::new(*address, scope)); } } specs @@ -86,29 +104,12 @@ pub async fn bind_gateway_listeners( info!(address = %local_addr, "Server listening"); listeners.push(BoundGatewayListener { listener, - address: local_addr, - scope: spec.scope, - covered_addresses: resolve_bound_covered_addresses( - &spec.covered_addresses, - spec.address, - local_addr, - ), + spec: spec.bind_to(local_addr), }); } Ok(listeners) } -pub fn gateway_listener_scope_for_local_addr( - default_scope: GatewayListenerScope, - covered_addresses: &[CoveredGatewayAddress], - local_addr: SocketAddr, -) -> GatewayListenerScope { - covered_addresses - .iter() - .find(|covered| covered.address == local_addr) - .map_or(default_scope, |covered| covered.scope) -} - fn resolve_bound_covered_addresses( covered_addresses: &[CoveredGatewayAddress], requested_listener_addr: SocketAddr, @@ -158,7 +159,7 @@ fn listener_covers(existing: SocketAddr, requested: SocketAddr) -> bool { mod tests { use super::{ CoveredGatewayAddress, GatewayListenerScope, GatewayListenerSpec, bind_gateway_listeners, - gateway_listener_scope_for_local_addr, gateway_listener_specs, + gateway_listener_specs, }; use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, Ordering}; @@ -192,11 +193,11 @@ mod tests { .unwrap(); assert_eq!( - gateway_listener_scope_for_local_addr(spec.scope, &spec.covered_addresses, docker), + spec.scope_for_local_addr(docker), GatewayListenerScope::ComputeDriverCallback, ); assert_eq!( - gateway_listener_scope_for_local_addr(spec.scope, &spec.covered_addresses, loopback), + spec.scope_for_local_addr(loopback), GatewayListenerScope::Primary, ); } diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index 0fb60ec43..e079f460e 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -72,10 +72,9 @@ use tracing::{debug, error, info, warn}; pub(crate) static TEST_ENV_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); use compute::ComputeRuntime; -use gateway_listener::{ - BoundGatewayListener, GatewayListenerScope, bind_gateway_listeners, - gateway_listener_scope_for_local_addr, -}; +#[cfg(test)] +use gateway_listener::GatewayListenerSpec; +use gateway_listener::{BoundGatewayListener, GatewayListenerScope, bind_gateway_listeners}; pub use grpc::OpenShellService; pub use http::{health_router, http_router, metrics_router, service_http_router}; pub use multiplex::{MultiplexService, MultiplexedService}; @@ -563,12 +562,8 @@ async fn serve_gateway_listener( enable_loopback_service_http: bool, mut shutdown: watch::Receiver, ) { - let BoundGatewayListener { - listener, - address: listen_addr, - scope, - covered_addresses, - } = bound_listener; + let BoundGatewayListener { listener, spec } = bound_listener; + let listen_addr = spec.address; loop { let accepted = tokio::select! { @@ -589,12 +584,10 @@ async fn serve_gateway_listener( } }; let listener_scope = match stream.local_addr() { - Ok(local_addr) => { - gateway_listener_scope_for_local_addr(scope, &covered_addresses, local_addr) - } + Ok(local_addr) => spec.scope_for_local_addr(local_addr), Err(e) => { debug!(error = %e, client = %addr, listen = %listen_addr, "Failed to inspect accepted local address"); - scope + spec.scope } }; @@ -1016,10 +1009,10 @@ pub(crate) async fn ensure_default_workspace(store: &Store) -> Result<()> { mod tests { use super::{ BoundGatewayListener, ConfiguredComputeDriver, ConnectionProtocol, GatewayListenerScope, - MultiplexService, ServerState, TlsAcceptor, allow_plaintext_service_http, - bind_gateway_listeners, classify_initial_bytes, configured_compute_driver, - is_benign_tls_handshake_failure, kubernetes_sandbox_jwt_expiry_disabled, - serve_gateway_listener, + GatewayListenerSpec, MultiplexService, ServerState, TlsAcceptor, + allow_plaintext_service_http, bind_gateway_listeners, classify_initial_bytes, + configured_compute_driver, is_benign_tls_handshake_failure, + kubernetes_sandbox_jwt_expiry_disabled, serve_gateway_listener, }; use openshell_core::{ ComputeDriverKind, Config, @@ -1115,9 +1108,7 @@ mod tests { let handle = tokio::spawn(serve_gateway_listener( BoundGatewayListener { listener, - address: listen_addr, - scope: GatewayListenerScope::Primary, - covered_addresses: Vec::new(), + spec: GatewayListenerSpec::new(listen_addr, GatewayListenerScope::Primary), }, service, Some(tls_acceptor), From 16ceacd32ab6ff4120b4d8797eefef2365619b7f Mon Sep 17 00:00:00 2001 From: Drew Newberry Date: Wed, 29 Jul 2026 20:48:12 -0700 Subject: [PATCH 5/5] ci(e2e): pin rootless podman packages Signed-off-by: Drew Newberry --- .github/workflows/e2e-test.yml | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/.github/workflows/e2e-test.yml b/.github/workflows/e2e-test.yml index eb7f83bed..a9270c378 100644 --- a/.github/workflows/e2e-test.yml +++ b/.github/workflows/e2e-test.yml @@ -141,10 +141,14 @@ jobs: # Podman 4.x. The probe records whether AppArmor blocks the drop. - runner: ubuntu-24.04 podman_major: "4" + podman_package_version: "4.9.3+ds1-1ubuntu0.2" + conmon_package_version: "2.1.10+ds1-1build2" # Ubuntu 26.04 provides the supported Podman 5.x coverage for # comparison with the Ubuntu 24.04 environment. - runner: ubuntu-26.04 podman_major: "5" + podman_package_version: "5.7.0+ds2-3build1" + conmon_package_version: "2.1.13+ds1-2" env: IMAGE_TAG: ${{ inputs.image-tag }} MISE_GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} @@ -193,9 +197,20 @@ jobs: openssh-client \ passt \ pkg-config \ - podman \ + "conmon=${{ matrix.conmon_package_version }}" \ + "podman=${{ matrix.podman_package_version }}" \ slirp4netns \ uidmap + # Hosted runners can place newer Podman and conmon binaries under + # /usr/local ahead of Ubuntu's packages. Select the distro CLI and + # use Podman's supported final config override for its conmon path. + podman_config="${RUNNER_TEMP}/openshell-containers.conf" + printf '%s\n' \ + '[engine]' \ + 'conmon_path = ["/usr/bin/conmon"]' \ + > "${podman_config}" + echo "/usr/bin" >> "${GITHUB_PATH}" + echo "CONTAINERS_CONF_OVERRIDE=${podman_config}" >> "${GITHUB_ENV}" - name: Configure rootless Podman run: | @@ -218,6 +233,10 @@ jobs: "${{ matrix.podman_major }}".*) ;; *) echo "ERROR: expected Podman ${{ matrix.podman_major }}.x, found $podman_version" >&2; exit 1 ;; esac + test "$(dpkg-query -W -f='${Version}' podman)" = "${{ matrix.podman_package_version }}" + test "$(dpkg-query -W -f='${Version}' conmon)" = "${{ matrix.conmon_package_version }}" + test "$(command -v podman)" = "/usr/bin/podman" + test "$(podman info --format '{{.Host.Conmon.Path}}')" = "/usr/bin/conmon" test "$(podman info --format '{{.Host.Security.Rootless}}')" = "true" test "$(sudo sysctl -n kernel.apparmor_restrict_unprivileged_userns)" = "1" echo "=== host ==="