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 ===" diff --git a/crates/openshell-server/src/gateway_listener.rs b/crates/openshell-server/src/gateway_listener.rs new file mode 100644 index 000000000..0957c4ebe --- /dev/null +++ b/crates/openshell-server/src/gateway_listener.rs @@ -0,0 +1,250 @@ +// 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)] +pub struct CoveredGatewayAddress { + pub address: SocketAddr, + pub scope: GatewayListenerScope, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +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 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::new( + bind_address, + GatewayListenerScope::Primary, + )]; + for address in extra_addresses { + let scope = GatewayListenerScope::ComputeDriverCallback; + if let Some(existing) = specs + .iter() + .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::new(*address, scope)); + } + } + 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, + spec: spec.bind_to(local_addr), + }); + } + Ok(listeners) +} + +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; + } + 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::{ + CoveredGatewayAddress, 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_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(); + + assert_eq!( + gateway_listener_specs(primary, &[docker, docker]), + 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!( + spec.scope_for_local_addr(docker), + GatewayListenerScope::ComputeDriverCallback, + ); + assert_eq!( + spec.scope_for_local_addr(loopback), + 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, + covered_addresses: Vec::new(), + }, + GatewayListenerSpec { + address: docker, + scope: GatewayListenerScope::ComputeDriverCallback, + covered_addresses: Vec::new(), + }, + ] + ); + } + + #[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..e079f460e 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,9 @@ use tracing::{debug, error, info, warn}; pub(crate) static TEST_ENV_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); use compute::ComputeRuntime; +#[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}; @@ -522,10 +526,9 @@ 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, service.clone(), tls_acceptor.clone(), enable_loopback_service_http, @@ -552,62 +555,16 @@ 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, + bound_listener: BoundGatewayListener, service: MultiplexService, tls_acceptor: Option, enable_loopback_service_http: bool, mut shutdown: watch::Receiver, ) { + let BoundGatewayListener { listener, spec } = bound_listener; + let listen_addr = spec.address; + loop { let accepted = tokio::select! { changed = shutdown.changed() => { @@ -626,11 +583,19 @@ async fn serve_gateway_listener( continue; } }; + let listener_scope = match stream.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"); + spec.scope + } + }; spawn_gateway_connection( stream, addr, listen_addr, + listener_scope, service.clone(), tls_acceptor.clone(), enable_loopback_service_http, @@ -699,6 +664,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 +679,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 +701,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 +731,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 +1008,10 @@ pub(crate) async fn ensure_default_workspace(store: &Store) -> Result<()> { #[cfg(test)] mod tests { use super::{ - ConfiguredComputeDriver, ConnectionProtocol, MultiplexService, ServerState, TlsAcceptor, + BoundGatewayListener, ConfiguredComputeDriver, ConnectionProtocol, GatewayListenerScope, + GatewayListenerSpec, MultiplexService, ServerState, TlsAcceptor, allow_plaintext_service_http, bind_gateway_listeners, classify_initial_bytes, - configured_compute_driver, gateway_listener_addresses, is_benign_tls_handshake_failure, + configured_compute_driver, is_benign_tls_handshake_failure, kubernetes_sandbox_jwt_expiry_disabled, serve_gateway_listener, }; use openshell_core::{ @@ -1132,8 +1106,10 @@ 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, + BoundGatewayListener { + listener, + spec: GatewayListenerSpec::new(listen_addr, GatewayListenerScope::Primary), + }, service, Some(tls_acceptor), enable_loopback_service_http, @@ -1511,28 +1487,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;