mirror of
https://github.com/NVIDIA/OpenShell.git
synced 2026-11-02 08:04:37 +08:00
Divesh
834b79a8c2
* feat(server): place supervisor sessions by consistent hash and hand them off on shutdown Signed-off-by: divesh <dgude@nvidia.com> * feat: route long-lived sandbox connections to the owning gateway replica Signed-off-by: divesh <dgude@nvidia.com> * feat(helm): add opt-in per-replica GRPCRoute routing Signed-off-by: divesh <dgude@nvidia.com> --------- Signed-off-by: divesh <dgude@nvidia.com>
606 lines
21 KiB
Rust
606 lines
21 KiB
Rust
// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
use miette::{IntoDiagnostic, Result, WrapErr};
|
|
use openshell_core::auth::EdgeAuthInterceptor;
|
|
use openshell_core::net::set_tcp_nodelay_best_effort;
|
|
use openshell_core::proto::open_shell_client::OpenShellClient;
|
|
use rustls::{
|
|
RootCertStore,
|
|
client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier},
|
|
pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime},
|
|
};
|
|
use std::collections::HashMap;
|
|
use std::future::Future;
|
|
use std::io::Cursor;
|
|
use std::net::SocketAddr;
|
|
use std::path::PathBuf;
|
|
use std::sync::OnceLock;
|
|
use std::time::Duration;
|
|
use tokio::sync::Mutex;
|
|
use tonic::metadata::AsciiMetadataValue;
|
|
use tonic::service::interceptor::InterceptedService;
|
|
use tonic::transport::{Certificate, Channel, ClientTlsConfig, Endpoint, Identity};
|
|
use tracing::debug;
|
|
use url::{Host, Url};
|
|
|
|
/// Concrete gRPC client type used by all commands.
|
|
pub type GrpcClient = OpenShellClient<InterceptedService<Channel, EdgeAuthInterceptor>>;
|
|
|
|
#[derive(Clone, Debug, Default)]
|
|
pub struct TlsOptions {
|
|
ca: Option<PathBuf>,
|
|
cert: Option<PathBuf>,
|
|
key: Option<PathBuf>,
|
|
/// Gateway name for resolving default cert directory.
|
|
gateway_name: Option<String>,
|
|
/// Edge auth bearer token — when set, disables mTLS client certs and
|
|
/// injects authentication headers on every gRPC request instead.
|
|
pub edge_token: Option<String>,
|
|
/// OIDC bearer token — when set, injects `authorization: Bearer <token>`
|
|
/// on every gRPC request. Takes precedence over `edge_token`.
|
|
pub oidc_token: Option<String>,
|
|
/// Skip TLS certificate verification for gateway connections.
|
|
pub gateway_insecure: bool,
|
|
}
|
|
|
|
impl TlsOptions {
|
|
pub fn new(ca: Option<PathBuf>, cert: Option<PathBuf>, key: Option<PathBuf>) -> Self {
|
|
Self {
|
|
ca,
|
|
cert,
|
|
key,
|
|
gateway_name: None,
|
|
edge_token: None,
|
|
oidc_token: None,
|
|
gateway_insecure: false,
|
|
}
|
|
}
|
|
|
|
pub fn has_any(&self) -> bool {
|
|
self.ca.is_some() || self.cert.is_some() || self.key.is_some()
|
|
}
|
|
|
|
/// Return the gateway name, if set.
|
|
pub fn gateway_name(&self) -> Option<&str> {
|
|
self.gateway_name.as_deref()
|
|
}
|
|
|
|
/// Set the gateway name for cert directory resolution.
|
|
#[must_use]
|
|
pub fn with_gateway_name(&self, name: &str) -> Self {
|
|
Self {
|
|
gateway_name: Some(name.to_string()),
|
|
..self.clone()
|
|
}
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn with_default_paths(&self, server: &str) -> Self {
|
|
let base = self
|
|
.gateway_name
|
|
.as_deref()
|
|
.and_then(tls_dir_for_gateway)
|
|
.or_else(|| default_tls_dir(server));
|
|
Self {
|
|
ca: self
|
|
.ca
|
|
.clone()
|
|
.or_else(|| base.as_ref().map(|dir| dir.join("ca.crt"))),
|
|
cert: self
|
|
.cert
|
|
.clone()
|
|
.or_else(|| base.as_ref().map(|dir| dir.join("tls.crt"))),
|
|
key: self
|
|
.key
|
|
.clone()
|
|
.or_else(|| base.as_ref().map(|dir| dir.join("tls.key"))),
|
|
gateway_name: self.gateway_name.clone(),
|
|
..self.clone()
|
|
}
|
|
}
|
|
|
|
/// Returns `true` when using bearer token auth.
|
|
pub fn is_bearer_auth(&self) -> bool {
|
|
self.edge_token.is_some() || self.oidc_token.is_some()
|
|
}
|
|
}
|
|
|
|
pub struct TlsMaterials {
|
|
ca: Vec<u8>,
|
|
cert: Vec<u8>,
|
|
key: Vec<u8>,
|
|
}
|
|
|
|
/// Resolve the TLS cert directory for a known gateway name.
|
|
fn tls_dir_for_gateway(name: &str) -> Option<PathBuf> {
|
|
let safe_name = sanitize_name(name);
|
|
let base = xdg_config_dir().ok()?.join("openshell").join("gateways");
|
|
Some(base.join(safe_name).join("mtls"))
|
|
}
|
|
|
|
/// Fallback TLS directory resolution from a server URL.
|
|
///
|
|
/// Used when no gateway name is set (e.g., `SshProxy` which receives a raw URL).
|
|
fn default_tls_dir(server: &str) -> Option<PathBuf> {
|
|
let mut name = std::env::var("OPENSHELL_GATEWAY")
|
|
.ok()
|
|
.filter(|value| !value.trim().is_empty());
|
|
|
|
if name.is_none()
|
|
&& let Ok(uri) = server.parse::<hyper::Uri>()
|
|
&& let Some(host) = uri.host()
|
|
{
|
|
name = Some(
|
|
if host == "127.0.0.1" || host.eq_ignore_ascii_case("localhost") {
|
|
"openshell".to_string()
|
|
} else {
|
|
host.to_string()
|
|
},
|
|
);
|
|
}
|
|
|
|
let name = name.unwrap_or_else(|| "openshell".to_string());
|
|
let safe_name = sanitize_name(&name);
|
|
let base = xdg_config_dir().ok()?.join("openshell").join("gateways");
|
|
Some(base.join(safe_name).join("mtls"))
|
|
}
|
|
|
|
fn sanitize_name(value: &str) -> String {
|
|
value
|
|
.chars()
|
|
.map(|ch| {
|
|
if ch.is_ascii_alphanumeric() || matches!(ch, '.' | '-' | '_') {
|
|
ch
|
|
} else {
|
|
'_'
|
|
}
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
fn xdg_config_dir() -> Result<PathBuf> {
|
|
openshell_core::paths::xdg_config_dir()
|
|
}
|
|
|
|
pub fn require_tls_materials(server: &str, tls: &TlsOptions) -> Result<TlsMaterials> {
|
|
let resolved = tls.with_default_paths(server);
|
|
let default_hint = default_tls_dir(server).map_or_else(String::new, |dir| {
|
|
format!(" or place certs in {}", dir.display())
|
|
});
|
|
let ca_path = resolved
|
|
.ca
|
|
.as_ref()
|
|
.ok_or_else(|| miette::miette!("TLS CA is required for https endpoints{default_hint}"))?;
|
|
let cert_path = resolved.cert.as_ref().ok_or_else(|| {
|
|
miette::miette!("TLS client cert is required for https endpoints{default_hint}")
|
|
})?;
|
|
let key_path = resolved.key.as_ref().ok_or_else(|| {
|
|
miette::miette!("TLS client key is required for https endpoints{default_hint}")
|
|
})?;
|
|
|
|
let ca = std::fs::read(ca_path)
|
|
.into_diagnostic()
|
|
.wrap_err_with(|| format!("failed to read TLS CA from {}", ca_path.display()))?;
|
|
let cert = std::fs::read(cert_path)
|
|
.into_diagnostic()
|
|
.wrap_err_with(|| format!("failed to read TLS cert from {}", cert_path.display()))?;
|
|
let key = std::fs::read(key_path)
|
|
.into_diagnostic()
|
|
.wrap_err_with(|| format!("failed to read TLS key from {}", key_path.display()))?;
|
|
|
|
Ok(TlsMaterials { ca, cert, key })
|
|
}
|
|
|
|
fn load_private_key(pem: &[u8]) -> Result<PrivateKeyDer<'static>> {
|
|
let mut cursor = Cursor::new(pem);
|
|
let key = rustls_pemfile::private_key(&mut cursor)
|
|
.into_diagnostic()?
|
|
.ok_or_else(|| miette::miette!("no private key found in TLS key PEM"))?;
|
|
Ok(key)
|
|
}
|
|
|
|
pub fn build_rustls_config(materials: &TlsMaterials) -> Result<rustls::ClientConfig> {
|
|
let mut roots = RootCertStore::empty();
|
|
let mut ca_cursor = Cursor::new(&materials.ca);
|
|
let ca_certs = rustls_pemfile::certs(&mut ca_cursor)
|
|
.collect::<Result<Vec<CertificateDer<'static>>, _>>()
|
|
.into_diagnostic()?;
|
|
for cert in ca_certs {
|
|
roots.add(cert).into_diagnostic()?;
|
|
}
|
|
|
|
let mut cert_cursor = Cursor::new(&materials.cert);
|
|
let cert_chain = rustls_pemfile::certs(&mut cert_cursor)
|
|
.collect::<Result<Vec<CertificateDer<'static>>, _>>()
|
|
.into_diagnostic()?;
|
|
let key = load_private_key(&materials.key)?;
|
|
|
|
rustls::ClientConfig::builder()
|
|
.with_root_certificates(roots)
|
|
.with_client_auth_cert(cert_chain, key)
|
|
.into_diagnostic()
|
|
}
|
|
|
|
pub fn build_tonic_tls_config(server: &str, materials: &TlsMaterials) -> Result<ClientTlsConfig> {
|
|
let ca_cert = Certificate::from_pem(materials.ca.clone());
|
|
let identity = Identity::from_pem(materials.cert.clone(), materials.key.clone());
|
|
tls_config_with_server_name(
|
|
server,
|
|
ClientTlsConfig::new()
|
|
.ca_certificate(ca_cert)
|
|
.identity(identity),
|
|
)
|
|
}
|
|
|
|
fn tls_config_with_server_name(server: &str, config: ClientTlsConfig) -> Result<ClientTlsConfig> {
|
|
Ok(config.domain_name(tls_server_name(server)?))
|
|
}
|
|
|
|
fn tls_server_name(server: &str) -> Result<String> {
|
|
let endpoint = Url::parse(server)
|
|
.into_diagnostic()
|
|
.wrap_err_with(|| format!("invalid gateway endpoint '{server}'"))?;
|
|
let host = endpoint
|
|
.host()
|
|
.ok_or_else(|| miette::miette!("gateway endpoint '{server}' has no host"))?;
|
|
Ok(match host {
|
|
Host::Domain(domain) => domain.to_string(),
|
|
Host::Ipv4(address) => address.to_string(),
|
|
Host::Ipv6(address) => address.to_string(),
|
|
})
|
|
}
|
|
|
|
#[derive(Debug)]
|
|
struct InsecureServerCertVerifier;
|
|
|
|
impl ServerCertVerifier for InsecureServerCertVerifier {
|
|
fn verify_server_cert(
|
|
&self,
|
|
_end_entity: &CertificateDer<'_>,
|
|
_intermediates: &[CertificateDer<'_>],
|
|
_server_name: &ServerName<'_>,
|
|
_ocsp_response: &[u8],
|
|
_now: UnixTime,
|
|
) -> Result<ServerCertVerified, rustls::Error> {
|
|
Ok(ServerCertVerified::assertion())
|
|
}
|
|
|
|
fn verify_tls12_signature(
|
|
&self,
|
|
_message: &[u8],
|
|
_cert: &CertificateDer<'_>,
|
|
_dss: &rustls::DigitallySignedStruct,
|
|
) -> Result<HandshakeSignatureValid, rustls::Error> {
|
|
Ok(HandshakeSignatureValid::assertion())
|
|
}
|
|
|
|
fn verify_tls13_signature(
|
|
&self,
|
|
_message: &[u8],
|
|
_cert: &CertificateDer<'_>,
|
|
_dss: &rustls::DigitallySignedStruct,
|
|
) -> Result<HandshakeSignatureValid, rustls::Error> {
|
|
Ok(HandshakeSignatureValid::assertion())
|
|
}
|
|
|
|
fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
|
|
rustls::crypto::aws_lc_rs::default_provider()
|
|
.signature_verification_algorithms
|
|
.supported_schemes()
|
|
}
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct InsecureTlsConnector {
|
|
tls_connector: tokio_rustls::TlsConnector,
|
|
}
|
|
|
|
impl tower::Service<hyper::Uri> for InsecureTlsConnector {
|
|
type Response = hyper_util::rt::TokioIo<tokio_rustls::client::TlsStream<tokio::net::TcpStream>>;
|
|
type Error = Box<dyn std::error::Error + Send + Sync>;
|
|
type Future =
|
|
std::pin::Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
|
|
|
|
fn poll_ready(
|
|
&mut self,
|
|
_cx: &mut std::task::Context<'_>,
|
|
) -> std::task::Poll<Result<(), Self::Error>> {
|
|
std::task::Poll::Ready(Ok(()))
|
|
}
|
|
|
|
fn call(&mut self, uri: hyper::Uri) -> Self::Future {
|
|
let tls_connector = self.tls_connector.clone();
|
|
Box::pin(async move {
|
|
let host = uri.host().unwrap_or("localhost").to_string();
|
|
let port = uri.port_u16().unwrap_or(443);
|
|
let addr = format!("{host}:{port}");
|
|
let tcp = tokio::net::TcpStream::connect(addr).await?;
|
|
set_tcp_nodelay_best_effort(&tcp);
|
|
let server_name = ServerName::try_from(host)?;
|
|
let tls_stream = tls_connector.connect(server_name, tcp).await?;
|
|
Ok(hyper_util::rt::TokioIo::new(tls_stream))
|
|
})
|
|
}
|
|
}
|
|
|
|
pub fn build_insecure_rustls_config() -> Result<rustls::ClientConfig> {
|
|
let config = rustls::ClientConfig::builder()
|
|
.dangerous()
|
|
.with_custom_certificate_verifier(std::sync::Arc::new(InsecureServerCertVerifier))
|
|
.with_no_client_auth();
|
|
Ok(config)
|
|
}
|
|
|
|
/// Tunnel proxy addresses keyed by upstream endpoint + token.
|
|
///
|
|
/// Each distinct edge-authenticated gateway gets its own local proxy instead of
|
|
/// reusing the first gateway touched in the current process.
|
|
static EDGE_TUNNEL_ADDRS: OnceLock<Mutex<HashMap<(String, String), SocketAddr>>> = OnceLock::new();
|
|
|
|
async fn edge_tunnel_addr(server: &str, token: &str) -> Result<SocketAddr> {
|
|
let key = (server.to_string(), token.to_string());
|
|
let registry = EDGE_TUNNEL_ADDRS.get_or_init(|| Mutex::new(HashMap::new()));
|
|
|
|
{
|
|
let addrs = registry.lock().await;
|
|
if let Some(addr) = addrs.get(&key).copied() {
|
|
return Ok(addr);
|
|
}
|
|
}
|
|
|
|
let proxy = crate::edge_tunnel::start_tunnel_proxy(server, token).await?;
|
|
debug!(
|
|
local_addr = %proxy.local_addr,
|
|
server,
|
|
"edge tunnel proxy started, routing gRPC through local proxy"
|
|
);
|
|
|
|
let mut addrs = registry.lock().await;
|
|
Ok(*addrs.entry(key).or_insert(proxy.local_addr))
|
|
}
|
|
|
|
pub async fn build_channel(server: &str, tls: &TlsOptions) -> Result<Channel> {
|
|
if server.starts_with("http://") {
|
|
let endpoint = Endpoint::from_shared(server.to_string())
|
|
.into_diagnostic()?
|
|
.connect_timeout(Duration::from_secs(10))
|
|
.http2_adaptive_window(true)
|
|
.http2_keep_alive_interval(Duration::from_secs(10))
|
|
.keep_alive_while_idle(true);
|
|
return endpoint.connect().await.into_diagnostic();
|
|
}
|
|
|
|
// When Cloudflare edge bearer auth is active and the server is HTTPS,
|
|
// route traffic through a local WebSocket tunnel proxy instead.
|
|
// OIDC tokens bypass the tunnel — they connect directly.
|
|
if tls.edge_token.is_some() && server.starts_with("https://") {
|
|
let token = tls
|
|
.edge_token
|
|
.as_deref()
|
|
.ok_or_else(|| miette::miette!("edge token required for tunnel"))?;
|
|
let local_addr = edge_tunnel_addr(server, token).await?;
|
|
|
|
// Connect to the local tunnel proxy over plaintext HTTP/2.
|
|
let local_url = format!("http://{local_addr}");
|
|
let endpoint = Endpoint::from_shared(local_url)
|
|
.into_diagnostic()?
|
|
.connect_timeout(Duration::from_secs(10))
|
|
.http2_adaptive_window(true)
|
|
.http2_keep_alive_interval(Duration::from_secs(10))
|
|
.keep_alive_while_idle(true);
|
|
return endpoint.connect().await.into_diagnostic();
|
|
}
|
|
|
|
if tls.gateway_insecure && server.starts_with("https://") {
|
|
tracing::warn!("TLS certificate verification is disabled — do not use in production");
|
|
let rustls_config = build_insecure_rustls_config()?;
|
|
let tls_connector = tokio_rustls::TlsConnector::from(std::sync::Arc::new(rustls_config));
|
|
let connector = InsecureTlsConnector { tls_connector };
|
|
// Use http:// so tonic does not layer its own TLS on top — our
|
|
// connector performs TLS with the insecure config.
|
|
let http_uri = server.replacen("https://", "http://", 1);
|
|
let endpoint = Endpoint::from_shared(http_uri)
|
|
.into_diagnostic()?
|
|
.connect_timeout(Duration::from_secs(10))
|
|
.http2_keep_alive_interval(Duration::from_secs(10))
|
|
.keep_alive_while_idle(true);
|
|
return endpoint
|
|
.connect_with_connector(connector)
|
|
.await
|
|
.into_diagnostic();
|
|
}
|
|
|
|
let mut endpoint = Endpoint::from_shared(server.to_string())
|
|
.into_diagnostic()?
|
|
.connect_timeout(Duration::from_secs(10))
|
|
.http2_adaptive_window(true)
|
|
.http2_keep_alive_interval(Duration::from_secs(10))
|
|
.keep_alive_while_idle(true);
|
|
|
|
let tls_config = if tls.oidc_token.is_some() {
|
|
// Bearer auth over HTTPS: use mTLS certs for the transport layer when
|
|
// available (server may still require client certs), and layer the
|
|
// Bearer token on top via the interceptor.
|
|
if let Ok(materials) = require_tls_materials(server, tls) {
|
|
build_tonic_tls_config(server, &materials)?
|
|
} else {
|
|
let resolved = tls.with_default_paths(server);
|
|
let config = resolved
|
|
.ca
|
|
.as_ref()
|
|
.and_then(|ca_path| std::fs::read(ca_path).ok())
|
|
.map_or_else(
|
|
|| ClientTlsConfig::new().with_enabled_roots(),
|
|
|ca_pem| ClientTlsConfig::new().ca_certificate(Certificate::from_pem(ca_pem)),
|
|
);
|
|
tls_config_with_server_name(server, config)?
|
|
}
|
|
} else if tls.edge_token.is_some() {
|
|
// Edge bearer mode — routed through tunnel above; if we reach here
|
|
// the server is not HTTPS so connect plaintext.
|
|
return endpoint.connect().await.into_diagnostic();
|
|
} else {
|
|
// Standard mTLS: private CA + client cert.
|
|
let materials = require_tls_materials(server, tls)?;
|
|
build_tonic_tls_config(server, &materials)?
|
|
};
|
|
endpoint = endpoint.tls_config(tls_config).into_diagnostic()?;
|
|
endpoint.connect().await.into_diagnostic()
|
|
}
|
|
|
|
/// Build a gRPC [`OpenShellClient`].
|
|
///
|
|
/// When `tls.edge_token` is set, the returned client is wrapped with an
|
|
/// interceptor that injects authentication headers on every request.
|
|
/// Otherwise, standard mTLS is used (interceptor is a no-op).
|
|
pub async fn grpc_client(server: &str, tls: &TlsOptions) -> Result<GrpcClient> {
|
|
let channel = build_channel(server, tls).await?;
|
|
let interceptor = interceptor_from_tls(tls)?;
|
|
Ok(OpenShellClient::with_interceptor(channel, interceptor))
|
|
}
|
|
|
|
fn interceptor_from_tls(tls: &TlsOptions) -> Result<EdgeAuthInterceptor> {
|
|
EdgeAuthInterceptor::new(tls.oidc_token.as_deref(), tls.edge_token.as_deref())
|
|
}
|
|
|
|
/// Gateway replica that owns the sandbox named in `response`, when it says.
|
|
pub fn owner_replica<T>(response: &tonic::Response<T>) -> Option<AsciiMetadataValue> {
|
|
response
|
|
.metadata()
|
|
.get(openshell_core::replica_routing::OWNER_REPLICA_HEADER)
|
|
.cloned()
|
|
}
|
|
|
|
pub fn with_owner_hint<T>(message: T) -> tonic::Request<T> {
|
|
let mut request = tonic::Request::new(message);
|
|
request.metadata_mut().insert(
|
|
openshell_core::replica_routing::OWNER_REQUEST_HEADER,
|
|
AsciiMetadataValue::from_static("1"),
|
|
);
|
|
request
|
|
}
|
|
|
|
/// Wrap `message` so the edge proxy can route it straight to `replica`.
|
|
pub fn routed_to<T>(message: T, replica: Option<&AsciiMetadataValue>) -> tonic::Request<T> {
|
|
let mut request = tonic::Request::new(message);
|
|
if let Some(replica) = replica {
|
|
request.metadata_mut().insert(
|
|
openshell_core::replica_routing::ROUTE_REPLICA_HEADER,
|
|
replica.clone(),
|
|
);
|
|
}
|
|
request
|
|
}
|
|
|
|
/// A routed call is retried once without routing only when no gateway
|
|
/// received it.
|
|
///
|
|
/// Envoy Gateway answers a replica route whose Service has no ready endpoints
|
|
/// with `grpc-status: 14` and no `grpc-message`, while gateway errors and
|
|
/// mid-request resets always carry a message.
|
|
pub fn retry_unrouted(replica: Option<&AsciiMetadataValue>, status: &tonic::Status) -> bool {
|
|
let message = status.message();
|
|
replica.is_some()
|
|
&& status.code() == tonic::Code::Unavailable
|
|
&& (message.is_empty()
|
|
|| message.contains("no healthy upstream")
|
|
|| message.contains("connection failure"))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{owner_replica, retry_unrouted, routed_to, tls_server_name, with_owner_hint};
|
|
use openshell_core::replica_routing::{
|
|
OWNER_REPLICA_HEADER, OWNER_REQUEST_HEADER, ROUTE_REPLICA_HEADER,
|
|
};
|
|
|
|
#[test]
|
|
fn owner_hint_is_requested_explicitly() {
|
|
assert!(
|
|
with_owner_hint(())
|
|
.metadata()
|
|
.contains_key(OWNER_REQUEST_HEADER)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn owner_replica_is_carried_to_the_routed_request() {
|
|
let mut response = tonic::Response::new(());
|
|
response
|
|
.metadata_mut()
|
|
.insert(OWNER_REPLICA_HEADER, "openshell-1".parse().unwrap());
|
|
let replica = owner_replica(&response);
|
|
let request = routed_to((), replica.as_ref());
|
|
assert_eq!(
|
|
request.metadata().get(ROUTE_REPLICA_HEADER).unwrap(),
|
|
"openshell-1"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn requests_without_a_known_owner_are_not_routed() {
|
|
let replica = owner_replica(&tonic::Response::new(()));
|
|
assert!(replica.is_none());
|
|
let request = routed_to((), replica.as_ref());
|
|
assert!(request.metadata().get(ROUTE_REPLICA_HEADER).is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn only_unavailable_routed_calls_are_retried() {
|
|
let replica = "openshell-1".parse().unwrap();
|
|
let unavailable = tonic::Status::unavailable("no healthy upstream");
|
|
assert!(retry_unrouted(Some(&replica), &unavailable));
|
|
assert!(!retry_unrouted(None, &unavailable));
|
|
assert!(!retry_unrouted(
|
|
Some(&replica),
|
|
&tonic::Status::permission_denied("denied")
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn calls_are_retried_only_when_no_gateway_received_them() {
|
|
let replica = "openshell-1".parse().unwrap();
|
|
for never_arrived in [
|
|
"",
|
|
"no healthy upstream",
|
|
"upstream connect error or disconnect/reset before headers. reset reason: connection failure, transport failure reason: delayed connect error: Connection refused",
|
|
] {
|
|
assert!(retry_unrouted(
|
|
Some(&replica),
|
|
&tonic::Status::unavailable(never_arrived)
|
|
));
|
|
}
|
|
for may_have_started in [
|
|
"supervisor session not connected",
|
|
"upstream connect error or disconnect/reset before headers. reset reason: connection termination",
|
|
] {
|
|
assert!(!retry_unrouted(
|
|
Some(&replica),
|
|
&tonic::Status::unavailable(may_have_started)
|
|
));
|
|
}
|
|
assert!(!retry_unrouted(
|
|
None,
|
|
&tonic::Status::unavailable("no healthy upstream")
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn tls_server_name_normalizes_bracketed_ipv6_endpoint() {
|
|
assert_eq!(
|
|
tls_server_name("https://[::1]:17670").unwrap(),
|
|
"::1",
|
|
"rustls accepts an unbracketed IPv6 address as an IP server name"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn tls_server_name_preserves_dns_endpoint() {
|
|
assert_eq!(
|
|
tls_server_name("https://localhost:17670").unwrap(),
|
|
"localhost"
|
|
);
|
|
}
|
|
}
|