Files
OpenShell/crates/openshell-server/src/multiplex.rs
T
krishicks 35a3390dbc feat(auth): classify rejected gateway credentials in authentication events (#4655)
Authenticators now return the mechanism and a reason category with the
status they reject a request with, and the gateway router records both
in the Authentication event. Callers still receive the original status.

Refs #2912

Signed-off-by: Kris Hicks <khicks@nvidia.com>
2026-10-16 20:14:45 +00:00

4087 lines
155 KiB
Rust

// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0
//! Protocol multiplexing for gRPC and HTTP on the same port.
//!
//! This module implements connection-level multiplexing that routes requests
//! to either the gRPC service or HTTP endpoints based on the request headers.
use bytes::{Bytes, BytesMut};
use http::{Extensions, HeaderValue, Request, Response};
use http_body::Body;
use http_body_util::{BodyExt, Full, LengthLimitError, Limited, StreamBody};
use hyper::body::Incoming;
use hyper_util::{
rt::{TokioExecutor, TokioIo, TokioTimer},
server::conn::auto::Builder,
service::TowerToHyperService,
};
use metrics::{counter, histogram};
use openshell_core::proto::open_shell_server::OpenShellServer;
use openshell_core::{
Config,
proto::{Provider, UpdateProviderRequest},
};
use openshell_gateway_interceptors::{EvaluationContext, GatewayInterceptorRuntime};
use openshell_otel::HeaderMapExtractor;
use opentelemetry::propagation::TextMapPropagator;
use opentelemetry::trace::TraceContextExt as _;
use opentelemetry_sdk::propagation::TraceContextPropagator;
use prost::Message;
use std::collections::BTreeMap;
use std::convert::Infallible;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use tokio::io::{AsyncRead, AsyncWrite};
use tower::ServiceExt;
use tower_http::request_id::{MakeRequestId, RequestId};
use tracing::{Span, warn};
use tracing_opentelemetry::OpenTelemetrySpanExt as _;
use crate::{
OpenShellService, ServerState,
auth::audit,
auth::authenticator::{AuthFailureReason, AuthMechanism, Authentication, AuthenticatorChain},
auth::authz::AuthzPolicy,
auth::identity::Identity,
auth::oidc::{self, OidcAuthenticator},
auth::principal::{OperatorPrincipal, Principal, UserPrincipal},
auth::workspace_authz::{MinWorkspaceRole, authorize_workspace},
http_router, service_http_router,
};
/// Request-ID generator that produces a UUID v4 for each inbound request.
#[derive(Clone)]
struct UuidRequestId;
impl MakeRequestId for UuidRequestId {
fn make_request_id<B>(&mut self, _req: &Request<B>) -> Option<RequestId> {
let id = uuid::Uuid::new_v4().to_string();
Some(RequestId::new(HeaderValue::from_str(&id).unwrap()))
}
}
/// Paths called on a timer rather than by someone waiting on the result.
const POLLED_PATHS: &[&str] = &[
"/health",
"/healthz",
"/readyz",
"/openshell.v1.OpenShell/GetSandboxConfig",
"/openshell.v1.OpenShell/ReportProviderReadiness",
"/openshell.v1.OpenShell/PeerReportProviderReadiness",
];
/// Build a tracing span for an inbound request, recording the `request_id`
/// header (set by [`UuidRequestId`] or supplied by the client).
fn make_request_span<B>(req: &Request<B>) -> Span {
let path = req.uri().path();
let request_id = req
.headers()
.get("x-request-id")
.and_then(|v| v.to_str().ok())
.unwrap_or("-");
// `otel.name` and `otel.kind` are consumed by `tracing-opentelemetry` to
// set the exported span's name and kind; they are not emitted as
// attributes. See [`otel_span_name`] for why the name cannot simply be
// the callsite name.
let otel_name = otel_span_name(req.method(), path);
let span = if POLLED_PATHS.contains(&path) {
tracing::debug_span!(
"request",
method = %req.method(),
path,
request_id,
otel.name = %otel_name,
otel.kind = "server",
otel.status_code = tracing::field::Empty,
http.response.status_code = tracing::field::Empty,
)
} else {
let span = tracing::info_span!(
"request",
method = %req.method(),
path,
request_id,
otel.name = %otel_name,
otel.kind = "server",
otel.status_code = tracing::field::Empty,
http.response.status_code = tracing::field::Empty,
rpc.system = tracing::field::Empty,
rpc.service = tracing::field::Empty,
rpc.method = tracing::field::Empty,
rpc.grpc.status_code = tracing::field::Empty,
);
// RPC-aware backends build service maps from these; without them a
// gRPC call is just an HTTP span.
if let Some((service, method)) = grpc_service_method(path) {
span.record("rpc.system", "grpc");
span.record("rpc.service", service);
span.record("rpc.method", method);
}
span
};
let propagator = TraceContextPropagator::new();
let parent = propagator.extract_with_context(
&opentelemetry::Context::new(),
&HeaderMapExtractor::new(req.headers()),
);
if parent.span().span_context().is_valid() {
let _ = span.set_parent(parent);
}
span
}
/// Log response status and latency, record protocol status, and mark failures.
fn log_response<B>(res: &Response<B>, latency: Duration, span: &Span) {
let status = res.status();
span.record("http.response.status_code", status.as_u16());
record_grpc_status(res.headers(), span);
if status.is_server_error() {
crate::otel_tracing::mark_error(span);
}
// Polled requests get a DEBUG span, which is `None` when filtered out.
let polled = span
.metadata()
.is_none_or(|m| *m.level() == tracing::Level::DEBUG);
let server_error = status.is_server_error();
let (status, latency_ms) = (status.as_u16(), latency.as_millis());
match (polled, server_error) {
(false, _) => tracing::info!(status, latency_ms, "response"),
(true, true) => tracing::warn!(status, latency_ms, "response"),
(true, false) => tracing::debug!(status, latency_ms, "response"),
}
}
fn record_response_trailers(
trailers: Option<&http::HeaderMap>,
_stream_duration: Duration,
span: &Span,
) {
if let Some(trailers) = trailers {
record_grpc_status(trailers, span);
}
}
fn record_grpc_status(headers: &http::HeaderMap, span: &Span) {
let Some(code) = headers
.get("grpc-status")
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse::<i64>().ok())
else {
return;
};
span.record("rpc.grpc.status_code", code);
if code != 0 {
crate::otel_tracing::mark_error(span);
}
}
/// Wrap a service with the standard request-ID middleware stack.
///
/// Layer order: `SetRequestId` → `TraceLayer` → `PropagateRequestId`.
macro_rules! request_id_middleware {
($service:expr) => {{
let x_request_id = ::http::HeaderName::from_static("x-request-id");
::tower::ServiceBuilder::new()
.layer(::tower_http::request_id::SetRequestIdLayer::new(
x_request_id.clone(),
UuidRequestId,
))
.layer(
::tower_http::trace::TraceLayer::new_for_http()
.make_span_with(make_request_span)
.on_request(())
.on_response(log_response)
.on_eos(record_response_trailers),
)
.layer(::tower_http::request_id::PropagateRequestIdLayer::new(
x_request_id,
))
.service($service)
}};
}
/// Maximum inbound gRPC message size (1 MB).
///
/// Replaces tonic's implicit 4 MB default with a conservative limit to
/// bound memory allocation from a single request. Sandbox creation is
/// the largest payload and well within this cap under normal use.
const MAX_GRPC_DECODE_SIZE: usize = 1_048_576;
const MAX_INTERCEPTED_GRPC_BODY_SIZE: usize = MAX_GRPC_DECODE_SIZE + 5;
/// Concurrent HTTP/2 streams allowed per connection. Sits above the
/// per-replica pending relay budget so pooled peer connections are bounded by
/// the relay caps rather than by the transport.
const MAX_HTTP2_CONCURRENT_STREAMS: u32 = 1024;
/// Multiplexed gRPC/HTTP service.
#[derive(Clone)]
pub struct MultiplexService {
state: Arc<ServerState>,
}
impl MultiplexService {
/// Create a new multiplex service.
#[must_use]
#[allow(clippy::missing_const_for_fn)]
pub fn new(state: Arc<ServerState>) -> Self {
Self { state }
}
/// Serve a connection, routing to gRPC or HTTP based on content-type.
pub async fn serve<S>(&self, stream: S) -> Result<(), Box<dyn std::error::Error + Send + Sync>>
where
S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
self.serve_with_peer_identity(stream, None).await
}
/// Serve a TLS connection with an optional mTLS peer identity.
pub async fn serve_with_peer_identity<S>(
&self,
stream: S,
peer_identity: Option<Identity>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>>
where
S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
self.serve_with_verified_certificate(stream, peer_identity, None)
.await
}
/// TLS listener only: fingerprint and identity must originate from its verified handshake.
pub async fn serve_with_verified_certificate<S>(
&self,
stream: S,
peer_identity: Option<Identity>,
certificate_sha256: Option<String>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>>
where
S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
let operator = if self.state.config.mtls_auth.operator_enabled {
peer_identity
.as_ref()
.filter(|identity| {
identity.provider == crate::auth::identity::IdentityProvider::Mtls
&& identity.roles.iter().any(|role| role == "operator")
})
.zip(certificate_sha256)
.map(|(identity, certificate_sha256)| OperatorPrincipal {
identity: identity.clone(),
certificate_sha256,
})
} else {
None
};
let openshell = OpenShellServer::new(OpenShellService::new(self.state.clone()))
.max_decoding_message_size(MAX_GRPC_DECODE_SIZE);
let openshell = GatewayInterceptorGrpcService::new(
openshell,
self.state.gateway_interceptors.clone(),
Some(self.state.clone()),
);
let reflection = self.state.reflection_service.clone();
let authz_policy = self.state.config.oidc.as_ref().map(|oidc| AuthzPolicy {
admin_role: oidc.admin_role.clone(),
user_role: oidc.user_role.clone(),
scopes_enabled: !oidc.scopes_claim.is_empty(),
});
let authenticator_chain = build_authenticator_chain(&self.state);
let grpc_service = AuthGrpcRouter::with_peer_identity(
GrpcRouter::new(openshell, reflection),
authenticator_chain,
authz_policy,
self.state
.config
.mtls_auth
.enabled
.then_some(peer_identity)
.flatten(),
self.state.config.mtls_auth.enabled,
self.state.config.auth.allow_unauthenticated_users,
)
.with_operator(operator);
let grpc_service =
GrpcRateLimitService::new(grpc_service, self.state.grpc_rate_limiter.clone());
let http_service = http_router(self.state.clone());
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 mut builder = Builder::new(TokioExecutor::new());
// Server-side HTTP/2 keepalive: supervisors hold long-lived sessions, and without
// it the gateway never PINGs them, so idle/half-dead connections linger and orphan
// in-flight relay execs. The timer is required — hyper panics on the keepalive
// interval without one.
//
// Peer relays from one replica now share a single pooled connection, so every
// forwarded session for every sandbox counts against this one limit. hyper's
// default of 200 would cap the whole replica pair below MAX_PENDING_RELAYS.
builder
.http2()
.timer(TokioTimer::new())
.adaptive_window(true)
.max_concurrent_streams(MAX_HTTP2_CONCURRENT_STREAMS)
.keep_alive_interval(Some(Duration::from_secs(20)))
.keep_alive_timeout(Duration::from_secs(10));
builder
.serve_connection_with_upgrades(TokioIo::new(stream), service)
.await?;
Ok(())
}
/// Serve a plaintext HTTP connection for sandbox service endpoints only.
pub async fn serve_service_http<S>(
&self,
stream: S,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>>
where
S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
let http_service = TowerToHyperService::new(request_id_middleware!(service_http_router(
self.state.clone()
)));
Builder::new(TokioExecutor::new())
.serve_connection_with_upgrades(TokioIo::new(stream), http_service)
.await?;
Ok(())
}
}
/// `OpenShell` gRPC wrapper that applies configured gateway interceptors before
/// tonic dispatches to a specific RPC handler.
#[derive(Clone)]
struct GatewayInterceptorGrpcService<S> {
inner: S,
interceptors: Option<GatewayInterceptorRuntime>,
state: Option<Arc<ServerState>>,
}
impl<S> GatewayInterceptorGrpcService<S> {
fn new(
inner: S,
interceptors: Option<GatewayInterceptorRuntime>,
state: Option<Arc<ServerState>>,
) -> Self {
Self {
inner,
interceptors,
state,
}
}
}
impl<S> tower::Service<Request<BoxBody>> for GatewayInterceptorGrpcService<S>
where
S: tower::Service<Request<BoxBody>, Response = Response<tonic::body::Body>>
+ Clone
+ Send
+ 'static,
S::Future: Send + 'static,
S::Error: Send + 'static,
{
type Response = S::Response;
type Error = S::Error;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, req: Request<BoxBody>) -> Self::Future {
let interceptors = self.interceptors.clone();
let state = self.state.clone();
let mut inner = self.inner.clone();
Box::pin(async move {
let Some(interceptors) = interceptors else {
return inner.ready().await?.call(req).await;
};
let path = req.uri().path().to_string();
if !interceptors.should_intercept_path(&path) {
return inner.ready().await?.call(req).await;
}
let context = gateway_interceptor_context(req.extensions());
let principal = req.extensions().get::<Principal>().cloned();
let (mut parts, body) = req.into_parts();
let mut body = match collect_intercepted_grpc_body(body).await {
Ok(body) => body,
Err(status) => return Ok(status.into_http()),
};
// Retain only in memory. evaluate_request below validates the single
// uncompressed frame before this extension reaches typed dispatch.
let original_body = body.clone();
if let Some(state) = state.as_ref() {
body =
match hydrate_update_provider_identity(&path, body, state, principal.as_ref())
.await
{
Ok(body) => body,
Err(status) => return Ok(status.into_http()),
};
}
let intercepted = match interceptors.evaluate_request(&path, &body, &context).await {
Ok(intercepted) => intercepted,
Err(status) => return Ok(status.into_http()),
};
parts
.extensions
.insert(crate::grpc::mutation_replay::OriginalMutation(
original_body[GRPC_FRAME_HEADER_LEN..].to_vec(),
));
let req = Request::from_parts(
parts,
boxed_body_from_bytes(Bytes::from(intercepted.body.clone())),
);
let response = inner.ready().await?.call(req).await?;
if grpc_status_from_response(&response) != "0"
|| response
.headers()
.get("openshell-replayed")
.is_some_and(|value| value == "true")
|| !interceptors.has_post_commit(&intercepted)
{
return Ok(response);
}
let (response, observation) = observe_intercepted_grpc_response(response).await;
let (response_body, trailers) = match observation {
Ok(observation) => observation,
Err(error) => {
warn!(
error = %error,
"gateway post-commit response observation failed; preserving committed response"
);
counter!(
"openshell_gateway_interceptor_post_commit_observation_failures_total",
"stage" => "response_body"
)
.increment(1);
return Ok(response);
}
};
if grpc_status_from_response_and_trailers(&response, trailers.as_ref()) == "0"
&& let Err(status) = interceptors
.evaluate_post_commit(&intercepted, &response_body, &context)
.await
{
warn!(
error = %status,
"gateway post-commit evaluation failed; preserving committed response"
);
counter!(
"openshell_gateway_interceptor_post_commit_observation_failures_total",
"stage" => "evaluation"
)
.increment(1);
}
Ok(response)
})
}
}
const UPDATE_PROVIDER_PATH: &str = "/openshell.v1.OpenShell/UpdateProvider";
const GRPC_FRAME_HEADER_LEN: usize = 5;
/// Complete immutable provider identity before policy interception.
///
/// Update requests intentionally omit immutable fields. Loading them here keeps
/// `provider update` a write-only operation while giving policy interceptors a
/// canonical proposed operation derived from trusted gateway state.
async fn hydrate_update_provider_identity(
path: &str,
body: Bytes,
state: &ServerState,
principal: Option<&Principal>,
) -> Result<Bytes, tonic::Status> {
if path != UPDATE_PROVIDER_PATH {
return Ok(body);
}
let mut request = decode_unary_grpc_message::<UpdateProviderRequest>(&body)?;
let Some(provider) = request.provider.as_mut() else {
return Ok(body);
};
if !provider.r#type.is_empty() && !provider.profile_workspace.is_empty() {
return Ok(body);
}
let name = provider
.metadata
.as_ref()
.map(|metadata| metadata.name.as_str())
.unwrap_or_default();
if name.is_empty() {
return Ok(body);
}
let principal =
principal.ok_or_else(|| tonic::Status::unauthenticated("authentication required"))?;
let authorized = authorize_workspace(
state.store.as_ref(),
&state.admin_role,
principal,
crate::auth::workspace_authz::selected_workspace_name(request.workspace_scope.as_ref())?,
MinWorkspaceRole::Admin,
)
.await?;
let workspace =
crate::grpc::workspace::resolve_workspace(state.store.as_ref(), &authorized.workspace)
.await?
.name;
let existing = state
.store
.get_message_by_name::<Provider>(&workspace, name)
.await
.map_err(|error| tonic::Status::internal(format!("provider lookup failed: {error}")))?
.ok_or_else(|| tonic::Status::not_found(format!("provider '{name}' not found")))?;
if provider.r#type.is_empty() {
provider.r#type = existing.r#type;
}
if provider.profile_workspace.is_empty() {
provider.profile_workspace = existing.profile_workspace;
}
encode_unary_grpc_message(&request)
}
fn decode_unary_grpc_message<M>(body: &[u8]) -> Result<M, tonic::Status>
where
M: Message + Default,
{
if body.len() < GRPC_FRAME_HEADER_LEN {
return Err(tonic::Status::invalid_argument("gRPC frame is too short"));
}
if body[0] != 0 {
return Err(tonic::Status::unimplemented(
"gateway interceptors do not support compressed gRPC frames",
));
}
let message_len = u32::from_be_bytes([body[1], body[2], body[3], body[4]]) as usize;
if body.len() != GRPC_FRAME_HEADER_LEN + message_len {
return Err(tonic::Status::invalid_argument(
"gRPC body must contain exactly one frame",
));
}
M::decode(&body[GRPC_FRAME_HEADER_LEN..])
.map_err(|error| tonic::Status::invalid_argument(format!("invalid gRPC message: {error}")))
}
fn encode_unary_grpc_message<M: Message>(message: &M) -> Result<Bytes, tonic::Status> {
let message = message.encode_to_vec();
let message_len = u32::try_from(message.len())
.map_err(|_| tonic::Status::resource_exhausted("gRPC message exceeds u32"))?;
let mut frame = Vec::with_capacity(GRPC_FRAME_HEADER_LEN + message.len());
frame.push(0);
frame.extend_from_slice(&message_len.to_be_bytes());
frame.extend_from_slice(&message);
Ok(Bytes::from(frame))
}
async fn collect_intercepted_grpc_body(body: BoxBody) -> Result<Bytes, tonic::Status> {
Limited::new(body, MAX_INTERCEPTED_GRPC_BODY_SIZE)
.collect()
.await
.map(http_body_util::Collected::to_bytes)
.map_err(|err| {
if err.downcast_ref::<LengthLimitError>().is_some() {
tonic::Status::resource_exhausted(format!(
"gRPC request body exceeds interceptor evaluation limit of {MAX_INTERCEPTED_GRPC_BODY_SIZE} bytes"
))
} else {
tonic::Status::internal(format!(
"failed to read gRPC request body for interceptor evaluation: {err}"
))
}
})
}
fn boxed_body_from_bytes(bytes: Bytes) -> BoxBody {
let body = Full::new(bytes)
.map_err(|never: Infallible| -> Box<dyn std::error::Error + Send + Sync> { match never {} })
.boxed_unsync();
BoxBody(body)
}
async fn observe_intercepted_grpc_response(
response: Response<tonic::body::Body>,
) -> (
Response<tonic::body::Body>,
Result<(Bytes, Option<http::HeaderMap>), String>,
) {
let (parts, mut body) = response.into_parts();
let mut frames = Vec::new();
let mut bytes = BytesMut::new();
let mut trailers = None;
while let Some(frame) = body.frame().await {
match frame {
Ok(frame) => {
if let Some(data) = frame.data_ref() {
bytes.extend_from_slice(data);
}
if let Some(frame_trailers) = frame.trailers_ref() {
trailers = Some(frame_trailers.clone());
}
frames.push(Ok(frame));
}
Err(status) => {
let error =
format!("failed to read gRPC response for post-commit evaluation: {status}");
frames.push(Err(status));
return (
Response::from_parts(parts, tonic_body_from_frames(frames)),
Err(error),
);
}
}
}
(
Response::from_parts(parts, tonic_body_from_frames(frames)),
Ok((bytes.freeze(), trailers)),
)
}
#[cfg(test)]
fn tonic_body_from_bytes_and_trailers(
bytes: Bytes,
trailers: Option<http::HeaderMap>,
) -> tonic::body::Body {
let mut frames: Vec<Result<http_body::Frame<Bytes>, tonic::Status>> = Vec::with_capacity(2);
if !bytes.is_empty() {
frames.push(Ok(http_body::Frame::data(bytes)));
}
if let Some(trailers) = trailers {
frames.push(Ok(http_body::Frame::trailers(trailers)));
}
tonic_body_from_frames(frames)
}
fn tonic_body_from_frames(
frames: Vec<Result<http_body::Frame<Bytes>, tonic::Status>>,
) -> tonic::body::Body {
tonic::body::Body::new(StreamBody::new(futures::stream::iter(frames)))
}
fn gateway_interceptor_context(extensions: &Extensions) -> EvaluationContext {
EvaluationContext {
principal: extensions
.get::<Principal>()
.map_or_else(unknown_gateway_principal, gateway_principal_fields),
validate_current_state: None,
}
}
fn gateway_principal_fields(principal: &Principal) -> BTreeMap<String, String> {
use crate::auth::principal::SandboxIdentitySource;
let mut fields = BTreeMap::new();
match principal {
Principal::User(user) => {
fields.insert("kind".to_string(), "user".to_string());
fields.insert("subject".to_string(), user.identity.subject.clone());
if let Some(display_name) = &user.identity.display_name {
fields.insert("display_name".to_string(), display_name.clone());
}
fields.insert(
"provider".to_string(),
identity_provider_name(user.identity.provider).to_string(),
);
if !user.identity.roles.is_empty() {
fields.insert("roles".to_string(), user.identity.roles.join(","));
}
if !user.identity.scopes.is_empty() {
fields.insert("scopes".to_string(), user.identity.scopes.join(","));
}
}
Principal::Operator(operator) => {
fields.insert("kind".to_string(), "operator".to_string());
fields.insert("subject".to_string(), operator.identity.subject.clone());
fields.insert("provider".to_string(), "mtls".to_string());
fields.insert(
"certificate_sha256".to_string(),
operator.certificate_sha256.clone(),
);
}
Principal::Sandbox(sandbox) => {
fields.insert("kind".to_string(), "sandbox".to_string());
fields.insert("sandbox_id".to_string(), sandbox.sandbox_id.clone());
fields.insert(
"source".to_string(),
match &sandbox.source {
SandboxIdentitySource::BootstrapJwt { .. } => "bootstrap_jwt",
SandboxIdentitySource::ComputeDriver { .. } => "compute_driver",
}
.to_string(),
);
if let Some(trust_domain) = &sandbox.trust_domain {
fields.insert("trust_domain".to_string(), trust_domain.clone());
}
}
Principal::Peer(peer) => {
fields.insert("kind".to_string(), "peer".to_string());
fields.insert("replica_id".to_string(), peer.replica_id.clone());
fields.insert("pod_uid".to_string(), peer.pod_uid.clone());
}
Principal::Anonymous => {
fields.insert("kind".to_string(), "anonymous".to_string());
}
}
fields
}
fn unknown_gateway_principal() -> BTreeMap<String, String> {
BTreeMap::from([("kind".to_string(), "unknown".to_string())])
}
fn identity_provider_name(provider: crate::auth::identity::IdentityProvider) -> &'static str {
match provider {
crate::auth::identity::IdentityProvider::Oidc => "oidc",
crate::auth::identity::IdentityProvider::Mtls => "mtls",
crate::auth::identity::IdentityProvider::CloudflareAccess => "cloudflare_access",
crate::auth::identity::IdentityProvider::LocalDev => "local_dev",
}
}
#[derive(Clone, Debug)]
pub struct GrpcRateLimiter {
requests: u64,
window: Duration,
state: Arc<Mutex<GrpcRateLimitState>>,
}
#[derive(Debug)]
struct GrpcRateLimitState {
window_started: Instant,
remaining: u64,
}
impl GrpcRateLimiter {
pub fn from_config(config: &Config) -> Option<Self> {
let (requests, window) = config.grpc_rate_limit()?;
Some(Self {
requests,
window,
state: Arc::new(Mutex::new(GrpcRateLimitState {
window_started: Instant::now(),
remaining: requests,
})),
})
}
pub(crate) fn allow(&self) -> bool {
let now = Instant::now();
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if now.duration_since(state.window_started) >= self.window {
state.window_started = now;
state.remaining = self.requests;
}
if state.remaining == 0 {
false
} else {
state.remaining -= 1;
true
}
}
/// Report whether the limiter currently has capacity without consuming a
/// token, rolling the window over first so an elapsed window reports
/// capacity again.
///
/// Used by `poll_ready` so an exhausted limiter reports readiness instead
/// of blocking on inner-service backpressure: `call` can then return
/// `RESOURCE_EXHAUSTED` immediately rather than waiting for the inner gRPC
/// service to become ready.
fn has_capacity(&self) -> bool {
let now = Instant::now();
let mut state = self
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if now.duration_since(state.window_started) >= self.window {
state.window_started = now;
state.remaining = self.requests;
}
state.remaining > 0
}
}
#[derive(Clone)]
struct GrpcRateLimitService<S> {
inner: S,
limiter: Option<GrpcRateLimiter>,
/// Set by `poll_ready` when it reports synthetic readiness for an
/// exhausted limiter without polling the inner service. The paired `call`
/// must then reject with `RESOURCE_EXHAUSTED` instead of forwarding to an
/// inner service that never reported readiness — even if the rate-limit
/// window rolls over in between. Reset whenever `poll_ready` defers to the
/// inner service.
rate_limited: bool,
}
impl<S> GrpcRateLimitService<S> {
fn new(inner: S, limiter: Option<GrpcRateLimiter>) -> Self {
Self {
inner,
limiter,
rate_limited: false,
}
}
}
impl<S, B> tower::Service<Request<B>> for GrpcRateLimitService<S>
where
S: tower::Service<Request<B>, Response = Response<tonic::body::Body>>,
S::Future: Send + 'static,
B: Send + 'static,
{
type Response = S::Response;
type Error = S::Error;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
// When the limiter is exhausted, report ready so `call` can return
// RESOURCE_EXHAUSTED immediately. Delegating to the inner service here
// would make rate-limited requests wait on inner backpressure (a
// pending inner `poll_ready`) before they are rejected. The check is
// non-consuming: the token is only consumed in `call` via `allow`.
//
// Crucially, this path does NOT poll the inner service, so the inner
// service has not reported readiness. Record that decision so the
// paired `call` rejects rather than forwarding to a service that never
// became ready — even if the rate-limit window rolls over in between.
if self
.limiter
.as_ref()
.is_some_and(|limiter| !limiter.has_capacity())
{
self.rate_limited = true;
return Poll::Ready(Ok(()));
}
self.rate_limited = false;
self.inner.poll_ready(cx)
}
fn call(&mut self, req: Request<B>) -> Self::Future {
// If `poll_ready` short-circuited an exhausted limiter, it never polled
// the inner service to readiness. Honor that decision regardless of the
// limiter's current state (the window may have rolled over since): the
// Tower contract forbids forwarding to an inner service that did not
// report readiness.
if std::mem::take(&mut self.rate_limited) {
let response =
tonic::Status::resource_exhausted("gRPC rate limit exceeded").into_http();
return Box::pin(async move { Ok(response) });
}
if self
.limiter
.as_ref()
.is_some_and(|limiter| !limiter.allow())
{
let response =
tonic::Status::resource_exhausted("gRPC rate limit exceeded").into_http();
return Box::pin(async move { Ok(response) });
}
let future = self.inner.call(req);
Box::pin(future)
}
}
/// Combined gRPC service that routes between `OpenShell` and reflection.
#[derive(Clone)]
pub struct GrpcRouter<N, R> {
openshell: N,
reflection: R,
}
impl<N, R> GrpcRouter<N, R> {
fn new(openshell: N, reflection: R) -> Self {
Self {
openshell,
reflection,
}
}
}
pub const REFLECTION_PATH_PREFIX: &str = "/grpc.reflection.v1.";
impl<N, R, B> tower::Service<Request<B>> for GrpcRouter<N, R>
where
N: tower::Service<Request<B>> + Clone + Send + 'static,
N::Response: Send,
N::Future: Send,
N::Error: Send,
R: tower::Service<Request<B>, Response = N::Response, Error = N::Error>
+ Clone
+ Send
+ 'static,
R::Future: Send,
B: Send + 'static,
{
type Response = N::Response;
type Error = N::Error;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: Request<B>) -> Self::Future {
if req.uri().path().starts_with(REFLECTION_PATH_PREFIX) {
let mut svc = self.reflection.clone();
Box::pin(async move { svc.ready().await?.call(req).await })
} else {
let mut svc = self.openshell.clone();
Box::pin(async move { svc.ready().await?.call(req).await })
}
}
}
/// Assemble the authenticator chain for the gateway.
///
/// Chain order (first-match-wins):
/// 1. `PeerServiceAccountAuthenticator` (path-scoped to peer RPCs)
/// — validates gateway replica projected `ServiceAccount` tokens with
/// `TokenReview` for internal peer relay calls. No-op on every other path.
/// 2. `ComputeDriverAuthenticator` (path-scoped to `IssueSandboxToken`)
/// — delegates a driver-native credential and receives a sandbox identity
/// so the handler can mint a gateway JWT. No-op on every other path.
/// 3. `SandboxSessionJwtAuthenticator` — validates generation-bound gateway
/// JWTs against the durable sandbox identity. Untyped legacy sandbox JWTs
/// are not admitted.
/// 4. `OidcAuthenticator` — validates user Bearer tokens against the
/// configured OIDC issuer. Returns `Unauthenticated` for missing
/// Bearer headers so non-OIDC clients can't sneak through.
///
/// Once sandbox authentication is configured, callers must present an
/// explicit credential for authenticated gRPC methods. Missing bearer auth
/// is promoted to an mTLS user only when `mtls_auth.enabled` is configured
/// and the connection presents a verified client certificate, or to an unsafe
/// local developer user when `auth.allow_unauthenticated_users` is explicitly
/// enabled.
///
/// When neither OIDC nor sandbox credentials are configured (a barebones
/// dev gateway), the chain is left as `None` so the router short-circuits
/// to pass-through unless mTLS or local unauthenticated users are enabled.
fn build_authenticator_chain(state: &ServerState) -> Option<AuthenticatorChain> {
let mut authenticators: Vec<Arc<dyn crate::auth::authenticator::Authenticator>> = Vec::new();
if let Some(peer) = state.peer_authenticator.clone() {
authenticators.push(peer);
}
if let Some(driver) = state.compute_driver_authenticator.clone() {
authenticators.push(driver);
}
if let Some(authority) = state.sandbox_session_jwt_authority.clone() {
authenticators.push(Arc::new(
crate::auth::sandbox_jwt::SandboxSessionJwtAuthenticator::new(
authority,
state.store.clone(),
),
));
}
if let Some(cache) = state.oidc_cache.clone() {
authenticators.push(Arc::new(OidcAuthenticator::new(cache)));
}
if authenticators.is_empty() {
return None;
}
Some(AuthenticatorChain::new(authenticators))
}
/// gRPC router wrapper that runs the [`AuthenticatorChain`] and inserts the
/// resulting [`Principal`] into the request's extensions.
///
/// Behavior:
/// - Strip any external `x-openshell-auth-source` marker first (so callers
/// cannot spoof a sandbox identity).
/// - Health probes / reflection bypass the chain entirely.
/// - When no chain is configured (OIDC not configured), forward without
/// authentication — preserves today's pass-through behavior.
/// - Otherwise, run the chain. The first match produces a `Principal`.
/// `Principal::User` is gated by the RBAC `AuthzPolicy`.
/// `Principal::Sandbox` is gated by a supervisor-method allowlist, then
/// handlers enforce same-sandbox scope on request bodies.
#[derive(Clone)]
pub struct AuthGrpcRouter<S> {
inner: S,
authenticator_chain: Option<AuthenticatorChain>,
authz_policy: Option<AuthzPolicy>,
/// mTLS peer identity extracted from the TLS handshake.
peer_identity: Option<Identity>,
mtls_auth_enabled: bool,
allow_unauthenticated_users: bool,
operator: Option<OperatorPrincipal>,
}
impl<S> AuthGrpcRouter<S> {
#[cfg(test)]
fn new(
inner: S,
authenticator_chain: Option<AuthenticatorChain>,
authz_policy: Option<AuthzPolicy>,
) -> Self {
Self::with_peer_identity(inner, authenticator_chain, authz_policy, None, false, false)
}
fn with_peer_identity(
inner: S,
authenticator_chain: Option<AuthenticatorChain>,
authz_policy: Option<AuthzPolicy>,
peer_identity: Option<Identity>,
mtls_auth_enabled: bool,
allow_unauthenticated_users: bool,
) -> Self {
Self {
inner,
authenticator_chain,
authz_policy,
peer_identity,
mtls_auth_enabled,
allow_unauthenticated_users,
operator: None,
}
}
fn with_operator(mut self, operator: Option<OperatorPrincipal>) -> Self {
self.operator = operator;
self
}
}
fn unauthenticated_dev_user_principal() -> Principal {
Principal::User(UserPrincipal {
identity: Identity {
subject: "unauthenticated-local-dev".to_string(),
display_name: Some("Unauthenticated Local Dev".to_string()),
roles: vec!["openshell-user".to_string(), "openshell-admin".to_string()],
scopes: vec!["openshell:all".to_string()],
provider: crate::auth::identity::IdentityProvider::LocalDev,
},
})
}
fn status_response(status: tonic::Status) -> Response<tonic::body::Body> {
status.into_http()
}
impl<S, B> tower::Service<Request<B>> for AuthGrpcRouter<S>
where
S: tower::Service<Request<B>, Response = Response<tonic::body::Body>> + Clone + Send + 'static,
S::Future: Send,
S::Error: Send + Into<Box<dyn std::error::Error + Send + Sync>>,
B: Send + 'static,
{
type Response = S::Response;
type Error = S::Error;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: Request<B>) -> Self::Future {
let chain = self.authenticator_chain.clone();
let authz_policy = self.authz_policy.clone();
let peer_identity = self.peer_identity.clone();
let mtls_auth_enabled = self.mtls_auth_enabled;
let allow_unauthenticated_users = self.allow_unauthenticated_users;
let operator = self.operator.clone();
let mut inner = self.inner.clone();
Box::pin(async move {
let mut req = req;
let path = req.uri().path().to_string();
// Health probes and reflection — truly unauthenticated.
if oidc::is_unauthenticated_method(&path) {
return inner.ready().await?.call(req).await;
}
let principal = if let Some(operator) = operator {
if req.headers().contains_key(http::header::AUTHORIZATION) {
audit::emit_authentication_failure(
&path,
Some(AuthMechanism::Mtls),
AuthFailureReason::ConflictingCredentials,
);
return Ok(status_response(tonic::Status::unauthenticated(
"operator certificates must not be combined with authorization headers",
)));
}
Principal::Operator(operator)
} else if let Some(chain) = chain {
match chain.authenticate(req.headers(), &path).await {
Ok(Authentication::Accepted(p)) => p,
Ok(outcome) => match (mtls_auth_enabled, peer_identity) {
(true, Some(identity)) => Principal::User(UserPrincipal { identity }),
_ if allow_unauthenticated_users => unauthenticated_dev_user_principal(),
_ => {
let (mechanism, reason) = match outcome {
Authentication::Declined(decline) => {
(Some(decline.mechanism), decline.reason)
}
_ => (None, AuthFailureReason::MissingCredential),
};
audit::emit_authentication_failure(&path, mechanism, reason);
return Ok(status_response(tonic::Status::unauthenticated(
"missing authorization header",
)));
}
},
Err(rejection) => {
audit::emit_authentication_failure(
&path,
Some(rejection.mechanism),
rejection.reason,
);
return Ok(status_response(rejection.status));
}
}
} else if mtls_auth_enabled {
let Some(identity) = peer_identity else {
audit::emit_authentication_failure(
&path,
Some(AuthMechanism::Mtls),
AuthFailureReason::MissingCredential,
);
return Ok(status_response(tonic::Status::unauthenticated(
"missing client certificate",
)));
};
Principal::User(UserPrincipal { identity })
} else if allow_unauthenticated_users {
unauthenticated_dev_user_principal()
} else {
// No auth configured — dev / fronting-proxy deployments.
// Inject a local-dev principal so downstream handlers that
// call extract_principal() always find one.
unauthenticated_dev_user_principal()
};
match principal {
Principal::Operator(_) => {
if !crate::auth::method_authz::is_operator_callable(&path) {
return Ok(status_response(tonic::Status::permission_denied(
"operators may not call supervisor, peer, or unknown methods",
)));
}
}
Principal::User(ref user) => {
if !crate::auth::method_authz::is_user_callable(&path) {
return Ok(status_response(tonic::Status::permission_denied(
"this method requires a sandbox principal",
)));
}
if let Some(ref policy) = authz_policy
&& let Err(status) = policy.check(&user.identity, &path)
{
return Ok(status_response(status));
}
}
Principal::Sandbox(_) => {
if !crate::auth::sandbox_methods::is_sandbox_callable(&path) {
return Ok(status_response(tonic::Status::permission_denied(
"sandbox principals may not call this method",
)));
}
}
Principal::Peer(_) => {
if !crate::auth::method_authz::is_peer_callable(&path) {
return Ok(status_response(tonic::Status::permission_denied(
"gateway peer principals may not call this method",
)));
}
}
Principal::Anonymous => {
return Ok(status_response(tonic::Status::unauthenticated(
"anonymous callers may not call authenticated methods",
)));
}
}
req.extensions_mut().insert(principal);
inner.ready().await?.call(req).await
})
}
}
/// Service that multiplexes between gRPC and HTTP.
#[derive(Clone)]
pub struct MultiplexedService<G, H> {
grpc: G,
http: H,
}
impl<G, H> MultiplexedService<G, H> {
/// Create a new multiplexed service from gRPC and HTTP services.
#[must_use]
pub fn new(grpc: G, http: H) -> Self {
Self { grpc, http }
}
}
impl<G, H, GBody, HBody> hyper::service::Service<Request<Incoming>> for MultiplexedService<G, H>
where
G: tower::Service<Request<BoxBody>, Response = Response<GBody>> + Clone + Send + 'static,
G::Future: Send,
G::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
GBody: Body<Data = Bytes> + Send + 'static,
GBody::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
H: tower::Service<Request<BoxBody>, Response = Response<HBody>> + Clone + Send + 'static,
H::Future: Send,
H::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
HBody: Body<Data = Bytes> + Send + 'static,
HBody::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
{
type Response = Response<BoxBody>;
type Error = Box<dyn std::error::Error + Send + Sync>;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn call(&self, req: Request<Incoming>) -> Self::Future {
let is_grpc = req
.headers()
.get("content-type")
.is_some_and(|v| v.as_bytes().starts_with(b"application/grpc"));
if is_grpc {
let method = grpc_method_from_path(req.uri().path());
let start = Instant::now();
let mut grpc = self.grpc.clone();
Box::pin(async move {
let (parts, body) = req.into_parts();
let body = body.map_err(Into::into).boxed_unsync();
let req = Request::from_parts(parts, BoxBody(body));
let res = grpc
.ready()
.await
.map_err(Into::into)?
.call(req)
.await
.map_err(Into::into)?;
let code = grpc_status_from_response(&res);
let elapsed = start.elapsed().as_secs_f64();
counter!("openshell_server_grpc_requests_total", "method" => method.clone(), "code" => code.clone()).increment(1);
histogram!("openshell_server_grpc_request_duration_seconds", "method" => method, "code" => code).record(elapsed);
let (parts, body) = res.into_parts();
let body = body.map_err(Into::into).boxed_unsync();
Ok(Response::from_parts(parts, BoxBody(body)))
})
} else {
let path = normalize_http_path(req.uri().path());
let start = Instant::now();
let mut http = self.http.clone();
Box::pin(async move {
let (parts, body) = req.into_parts();
let body = body.map_err(Into::into).boxed_unsync();
let req = Request::from_parts(parts, BoxBody(body));
let res = http
.ready()
.await
.map_err(Into::into)?
.call(req)
.await
.map_err(Into::into)?;
let status = res.status().as_u16().to_string();
let elapsed = start.elapsed().as_secs_f64();
counter!("openshell_server_http_requests_total", "path" => path, "status" => status.clone()).increment(1);
histogram!("openshell_server_http_request_duration_seconds", "path" => path, "status" => status).record(elapsed);
let (parts, body) = res.into_parts();
let body = body.map_err(Into::into).boxed_unsync();
Ok(Response::from_parts(parts, BoxBody(body)))
})
}
}
}
fn grpc_method_from_path(path: &str) -> String {
path.rsplit('/').next().unwrap_or(path).to_string()
}
/// Name for the exported `OpenTelemetry` span, per the `OTel` semantic
/// conventions: `$service/$method` for RPCs and the method for plain HTTP.
///
/// The gateway cannot determine route templates for proxied sandbox
/// applications, so including the literal path would create high-cardinality
/// operation names. The path remains available as a span attribute.
///
/// The `tracing` callsite name is the constant `"request"` because `tracing`
/// requires `'static` span names, so the per-request name is carried in the
/// `otel.name` field instead.
fn otel_span_name(method: &http::Method, path: &str) -> String {
grpc_service_method(path).map_or_else(
|| method.to_string(),
|(service, rpc_method)| format!("{service}/{rpc_method}"),
)
}
/// Split a gRPC path into its service and method.
///
/// A gRPC path is exactly "/package.Service/Method". Anything else — a health
/// check, /metrics, a sandbox service URL — is plain HTTP.
fn grpc_service_method(path: &str) -> Option<(&str, &str)> {
let mut segments = path.strip_prefix('/')?.split('/');
let service = segments.next()?;
let method = segments.next()?;
if segments.next().is_some() || !service.contains('.') || method.is_empty() {
return None;
}
Some((service, method))
}
fn grpc_status_from_response<B>(res: &Response<B>) -> String {
res.headers()
.get("grpc-status")
.and_then(|v| v.to_str().ok())
.map_or_else(|| "0".to_string(), ToString::to_string)
}
fn grpc_status_from_response_and_trailers<B>(
res: &Response<B>,
trailers: Option<&http::HeaderMap>,
) -> String {
trailers
.and_then(|trailers| trailers.get("grpc-status"))
.or_else(|| res.headers().get("grpc-status"))
.and_then(|value| value.to_str().ok())
.map_or_else(|| "0".to_string(), ToString::to_string)
}
fn normalize_http_path(path: &str) -> &'static str {
match path {
p if p.starts_with("/_ws_tunnel") => "/_ws_tunnel",
p if p.starts_with("/auth/") => "/auth",
_ => "unknown",
}
}
/// Extract an [`Identity`] from the peer certificates presented during a TLS
/// handshake. Returns `None` if no client certificate was presented.
pub fn extract_peer_identity<S>(tls_stream: &tokio_rustls::server::TlsStream<S>) -> Option<Identity>
where
S: AsyncRead + AsyncWrite + Unpin,
{
use crate::auth::identity::IdentityProvider;
use x509_parser::prelude::*;
let (_, server_conn) = tls_stream.get_ref();
let certs = server_conn.peer_certificates()?;
let first = certs.first()?;
let (_, cert) = X509Certificate::from_der(first.as_ref()).ok()?;
let subject = cert.subject();
let cn = subject
.iter_common_name()
.next()
.and_then(|attr| attr.as_str().ok())
.unwrap_or("unknown")
.to_string();
let roles: Vec<String> = subject
.iter_organizational_unit()
.filter_map(|attr| attr.as_str().ok().map(String::from))
.collect();
Some(Identity {
subject: cn.clone(),
display_name: Some(cn),
roles,
scopes: Vec::new(),
provider: IdentityProvider::Mtls,
})
}
/// Fingerprint of the leaf certificate verified during the TLS handshake.
pub fn extract_peer_certificate_sha256<S>(
stream: &tokio_rustls::server::TlsStream<S>,
) -> Option<String> {
use sha2::{Digest, Sha256};
let certificate = stream.get_ref().1.peer_certificates()?.first()?;
Some(format!("{:x}", Sha256::digest(certificate.as_ref())))
}
/// Boxed body type for uniform handling.
pub struct BoxBody(
http_body_util::combinators::UnsyncBoxBody<Bytes, Box<dyn std::error::Error + Send + Sync>>,
);
impl Body for BoxBody {
type Data = Bytes;
type Error = Box<dyn std::error::Error + Send + Sync>;
fn poll_frame(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
Pin::new(&mut self.0).poll_frame(cx)
}
fn is_end_stream(&self) -> bool {
self.0.is_end_stream()
}
fn size_hint(&self) -> http_body::SizeHint {
self.0.size_hint()
}
}
#[cfg(test)]
mod tests {
use super::*;
use bytes::Bytes;
use http::StatusCode;
use http_body_util::Empty;
use openshell_core::GatewayInterceptorConfig;
use openshell_core::proto::CreateSandboxRequest;
use openshell_core::proto::gateway_interceptor::v1::{
DescribeRequest, GatewayInterceptorPhase, InterceptorBinding, InterceptorEvaluation,
InterceptorManifest, InterceptorResult, InterceptorSelector, ProviderProfileSnapshot,
ProviderProfileSnapshotRequest,
gateway_interceptor_server::{GatewayInterceptor, GatewayInterceptorServer},
};
use std::convert::Infallible;
use std::sync::Mutex;
use std::sync::atomic::{AtomicUsize, Ordering};
use tokio_stream::wrappers::TcpListenerStream;
use tower::Service;
#[derive(Clone)]
struct PostCommitTestInterceptor;
#[tonic::async_trait]
impl GatewayInterceptor for PostCommitTestInterceptor {
async fn describe(
&self,
_request: tonic::Request<DescribeRequest>,
) -> Result<tonic::Response<InterceptorManifest>, tonic::Status> {
Ok(tonic::Response::new(InterceptorManifest {
name: "post-commit-test".to_string(),
failure_policy: "fail_open".to_string(),
bindings: vec![InterceptorBinding {
id: "audit-create-sandbox".to_string(),
selector: Some(InterceptorSelector {
rpc: "openshell.v1.OpenShell/CreateSandbox".to_string(),
service: String::new(),
method: String::new(),
}),
phases: vec![GatewayInterceptorPhase::PostCommit as i32],
failure_policy: "fail_open".to_string(),
}],
provider_profiles: false,
expected_audience: String::new(),
extension: Some(openshell_core::extension_protocol::extension_metadata(
openshell_core::extension_protocol::ExtensionFamily::GatewayInterceptor,
"openshell/post-commit-test",
"test",
[],
)),
}))
}
async fn evaluate(
&self,
_request: tonic::Request<InterceptorEvaluation>,
) -> Result<tonic::Response<InterceptorResult>, tonic::Status> {
Ok(tonic::Response::new(InterceptorResult {
allowed: true,
..InterceptorResult::default()
}))
}
async fn snapshot_provider_profiles(
&self,
_request: tonic::Request<ProviderProfileSnapshotRequest>,
) -> Result<tonic::Response<ProviderProfileSnapshot>, tonic::Status> {
Err(tonic::Status::unimplemented("not a profile source"))
}
}
fn grpc_frame(message: &[u8]) -> Bytes {
let mut frame = Vec::with_capacity(5 + message.len());
frame.push(0);
frame.extend_from_slice(&u32::try_from(message.len()).unwrap().to_be_bytes());
frame.extend_from_slice(message);
Bytes::from(frame)
}
async fn post_commit_test_runtime() -> (GatewayInterceptorRuntime, tokio::task::JoinHandle<()>)
{
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let task = tokio::spawn(async move {
tonic::transport::Server::builder()
.add_service(GatewayInterceptorServer::new(PostCommitTestInterceptor))
.serve_with_incoming(TcpListenerStream::new(listener))
.await
.unwrap();
});
let runtime = openshell_gateway_interceptors::initialize(vec![GatewayInterceptorConfig {
name: "post-commit-test".to_string(),
grpc_endpoint: format!("http://{address}"),
..GatewayInterceptorConfig::default()
}])
.await
.unwrap()
.unwrap();
(runtime, task)
}
#[test]
fn uuid_request_id_generates_valid_uuid() {
let mut maker = UuidRequestId;
let req = Request::builder().body(()).unwrap();
let id = maker.make_request_id(&req).expect("should produce an ID");
let value = id.header_value().to_str().unwrap();
uuid::Uuid::parse_str(value).expect("should be a valid UUID");
}
#[test]
fn uuid_request_id_generates_unique_ids() {
let mut maker = UuidRequestId;
let req = Request::builder().body(()).unwrap();
let id1 = maker.make_request_id(&req).unwrap();
let id2 = maker.make_request_id(&req).unwrap();
assert_ne!(id1.header_value(), id2.header_value());
}
async fn test_health_store() -> Arc<crate::Store> {
Arc::new(
crate::Store::connect("sqlite::memory:")
.await
.expect("connect in-memory sqlite store for tests"),
)
}
async fn start_http_server_with_middleware() -> std::net::SocketAddr {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let http_service = crate::http::health_router(test_health_store().await);
let http_service = request_id_middleware!(http_service);
let service = MultiplexedService::new(http_service.clone(), http_service);
tokio::spawn(async move {
loop {
let Ok((stream, _)) = listener.accept().await else {
continue;
};
let svc = service.clone();
tokio::spawn(async move {
let _ = Builder::new(TokioExecutor::new())
.serve_connection(TokioIo::new(stream), svc)
.await;
});
}
});
addr
}
async fn http1_request(
addr: std::net::SocketAddr,
method: &str,
path: &str,
headers: &[(&str, &str)],
) -> Response<Incoming> {
let stream = tokio::net::TcpStream::connect(addr).await.unwrap();
let (mut sender, conn) = hyper::client::conn::http1::Builder::new()
.handshake(TokioIo::new(stream))
.await
.unwrap();
tokio::spawn(async move {
let _ = conn.await;
});
let mut builder = Request::builder()
.method(method)
.uri(format!("http://{addr}{path}"));
for (k, v) in headers {
builder = builder.header(*k, *v);
}
let req = builder.body(Empty::<Bytes>::new()).unwrap();
sender.send_request(req).await.unwrap()
}
async fn http1_get(
addr: std::net::SocketAddr,
path: &str,
headers: &[(&str, &str)],
) -> Response<Incoming> {
http1_request(addr, "GET", path, headers).await
}
#[tokio::test]
async fn intercepted_grpc_body_collection_rejects_oversized_body() {
let oversized = Bytes::from(vec![0_u8; MAX_INTERCEPTED_GRPC_BODY_SIZE + 1]);
let status = collect_intercepted_grpc_body(boxed_body_from_bytes(oversized))
.await
.expect_err("oversized body should be rejected");
assert_eq!(status.code(), tonic::Code::ResourceExhausted);
}
#[tokio::test]
async fn update_provider_interception_hydrates_identity_from_trusted_state() {
let state = crate::grpc::test_support::test_server_state().await;
let existing = Provider {
metadata: Some(openshell_core::proto::datamodel::v1::ObjectMeta {
id: "provider-id".to_string(),
name: "managed-provider".to_string(),
workspace: "default".to_string(),
..Default::default()
}),
r#type: "agent-tool-gateway".to_string(),
profile_workspace: "default".to_string(),
..Default::default()
};
state.store.put_message(&existing).await.unwrap();
let request = UpdateProviderRequest {
provider: Some(Provider {
metadata: Some(openshell_core::proto::datamodel::v1::ObjectMeta {
name: "managed-provider".to_string(),
workspace: "default".to_string(),
..Default::default()
}),
credentials: std::collections::HashMap::from([(
"TOKEN".to_string(),
"rotated".to_string(),
)]),
..Default::default()
}),
workspace_scope: Some(openshell_core::proto::workspace_selector(
"default".to_string(),
)),
..Default::default()
};
let authed = crate::grpc::test_support::authed_request(());
let principal = authed.extensions().get::<Principal>().unwrap();
let hydrated = hydrate_update_provider_identity(
UPDATE_PROVIDER_PATH,
grpc_frame(&request.encode_to_vec()),
state.as_ref(),
Some(principal),
)
.await
.unwrap();
let hydrated = decode_unary_grpc_message::<UpdateProviderRequest>(&hydrated).unwrap();
let provider = hydrated.provider.unwrap();
assert_eq!(provider.r#type, "agent-tool-gateway");
assert_eq!(provider.profile_workspace, "default");
assert_eq!(provider.credentials.get("TOKEN").unwrap(), "rotated");
}
#[tokio::test]
async fn intercepted_grpc_response_preserves_body_and_trailers() {
let bytes = Bytes::from_static(b"committed-response");
let mut trailers = http::HeaderMap::new();
trailers.insert("grpc-status", HeaderValue::from_static("0"));
let response = Response::new(tonic_body_from_bytes_and_trailers(
bytes.clone(),
Some(trailers.clone()),
));
let (response, observation) = observe_intercepted_grpc_response(response).await;
let (observed, observed_trailers) = observation.unwrap();
assert_eq!(observed, bytes);
assert_eq!(observed_trailers.as_ref(), Some(&trailers));
assert_eq!(
grpc_status_from_response_and_trailers(&response, observed_trailers.as_ref()),
"0"
);
let collected = response.into_body().collect().await.unwrap();
assert_eq!(collected.trailers(), Some(&trailers));
assert_eq!(collected.to_bytes(), bytes);
}
#[tokio::test]
async fn intercepted_grpc_response_preserves_body_error() {
let bytes = Bytes::from_static(b"committed-response-prefix");
let response = Response::new(tonic_body_from_frames(vec![
Ok(http_body::Frame::data(bytes.clone())),
Err(tonic::Status::unavailable("response stream failed")),
]));
let (response, observation) = observe_intercepted_grpc_response(response).await;
let observation_error = observation.unwrap_err();
assert!(observation_error.contains("failed to read gRPC response"));
assert!(observation_error.contains("response stream failed"));
let mut body = response.into_body();
let data = body.frame().await.unwrap().unwrap();
assert_eq!(data.data_ref(), Some(&bytes));
let status = body.frame().await.unwrap().unwrap_err();
assert_eq!(status.code(), tonic::Code::Unavailable);
assert_eq!(status.message(), "response stream failed");
assert!(body.frame().await.is_none());
}
#[tokio::test]
async fn post_commit_decode_failure_preserves_committed_response() {
let (runtime, interceptor_task) = post_commit_test_runtime().await;
let committed = Arc::new(AtomicUsize::new(0));
let committed_for_service = committed.clone();
let committed_body = Bytes::from_static(b"malformed committed gRPC response");
let committed_body_for_service = committed_body.clone();
let inner = tower::service_fn(move |_request: Request<BoxBody>| {
let committed = committed_for_service.clone();
let body = committed_body_for_service.clone();
async move {
committed.fetch_add(1, Ordering::SeqCst);
let mut trailers = http::HeaderMap::new();
trailers.insert("grpc-status", HeaderValue::from_static("0"));
Ok::<_, Infallible>(Response::new(tonic_body_from_bytes_and_trailers(
body,
Some(trailers),
)))
}
});
let mut service = GatewayInterceptorGrpcService::new(inner, Some(runtime), None);
let request_body = grpc_frame(&CreateSandboxRequest::default().encode_to_vec());
let request = Request::builder()
.uri("/openshell.v1.OpenShell/CreateSandbox")
.body(boxed_body_from_bytes(request_body))
.unwrap();
let response = service.ready().await.unwrap().call(request).await.unwrap();
assert_eq!(committed.load(Ordering::SeqCst), 1);
let collected = response.into_body().collect().await.unwrap();
assert_eq!(collected.to_bytes(), committed_body);
interceptor_task.abort();
}
#[derive(Clone, Default)]
struct ReplayTestInterceptor {
modifications: Arc<AtomicUsize>,
validations: Arc<AtomicUsize>,
observations: Arc<AtomicUsize>,
deny: Arc<std::sync::atomic::AtomicBool>,
name: Arc<Mutex<String>>,
}
#[tonic::async_trait]
impl GatewayInterceptor for ReplayTestInterceptor {
async fn describe(
&self,
request: tonic::Request<DescribeRequest>,
) -> Result<tonic::Response<InterceptorManifest>, tonic::Status> {
let mut manifest = PostCommitTestInterceptor
.describe(request)
.await?
.into_inner();
manifest.bindings.push(InterceptorBinding {
id: "validate-create".into(),
selector: manifest.bindings[0].selector.clone(),
phases: vec![
GatewayInterceptorPhase::ModifyOperation.into(),
GatewayInterceptorPhase::Validate.into(),
],
failure_policy: "fail_closed".into(),
});
Ok(tonic::Response::new(manifest))
}
async fn evaluate(
&self,
request: tonic::Request<InterceptorEvaluation>,
) -> Result<tonic::Response<InterceptorResult>, tonic::Status> {
use openshell_core::proto::gateway_interceptor::v1::{
JsonPatch, interceptor_evaluation::Phase,
};
let mut result = InterceptorResult {
allowed: true,
..Default::default()
};
match request.into_inner().phase.unwrap() {
Phase::ModifyOperation(_) => {
self.modifications.fetch_add(1, Ordering::SeqCst);
result.patches.push(JsonPatch {
op: "replace".into(),
path: "/name".into(),
value: Some(prost_types::Value {
kind: Some(prost_types::value::Kind::StringValue(
self.name.lock().unwrap().clone(),
)),
}),
..Default::default()
});
}
Phase::Validate(_) => {
self.validations.fetch_add(1, Ordering::SeqCst);
result.allowed = !self.deny.load(Ordering::SeqCst);
result.reason = "current policy denies creation".into();
result.status_code = "PERMISSION_DENIED".into();
}
Phase::PostCommit(_) => {
self.observations.fetch_add(1, Ordering::SeqCst);
}
}
Ok(tonic::Response::new(result))
}
async fn snapshot_provider_profiles(
&self,
request: tonic::Request<ProviderProfileSnapshotRequest>,
) -> Result<tonic::Response<ProviderProfileSnapshot>, tonic::Status> {
PostCommitTestInterceptor
.snapshot_provider_profiles(request)
.await
}
}
#[tokio::test]
async fn mutation_replay_revalidates_interceptors_and_observes_commit_once() {
use openshell_core::proto::{
SandboxResponse, SandboxSpec, open_shell_server::OpenShellServer,
};
let interceptor = ReplayTestInterceptor::default();
*interceptor.name.lock().unwrap() = "intercepted-create".into();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let address = listener.local_addr().unwrap();
let service = interceptor.clone();
let task = tokio::spawn(async move {
tonic::transport::Server::builder()
.add_service(GatewayInterceptorServer::new(service))
.serve_with_incoming(TcpListenerStream::new(listener))
.await
.unwrap();
});
let runtime = openshell_gateway_interceptors::initialize(vec![GatewayInterceptorConfig {
name: "post-commit-test".into(),
grpc_endpoint: format!("http://{address}"),
..Default::default()
}])
.await
.unwrap();
let directory = tempfile::tempdir().unwrap();
let key = directory.path().join("key");
std::fs::write(&key, b"test-only-private-key").unwrap();
let mut state = crate::grpc::test_support::test_server_state().await;
Arc::get_mut(&mut state).unwrap().config.gateway_jwt =
Some(openshell_core::config::GatewayJwtConfig {
signing_key_path: key,
public_key_path: directory.path().join("public"),
kid_path: directory.path().join("kid"),
gateway_id: "test".into(),
ttl_secs: None,
});
let inner = OpenShellServer::new(OpenShellService::new(state.clone()));
let mut service = GatewayInterceptorGrpcService::new(inner, runtime, Some(state));
let input = CreateSandboxRequest {
name: "client-original".into(),
request_id: uuid::Uuid::new_v4().to_string(),
workspace_scope: Some(openshell_core::proto::workspace_selector(
"default".to_string(),
)),
spec: Some(SandboxSpec::default()),
..Default::default()
};
let request = || {
let principal = crate::grpc::test_support::authed_request(())
.extensions()
.get::<Principal>()
.unwrap()
.clone();
Request::builder()
.uri("/openshell.v1.OpenShell/CreateSandbox")
.header("content-type", "application/grpc")
.header("openshell-replayed", "true")
.extension(principal)
.body(boxed_body_from_bytes(grpc_frame(&input.encode_to_vec())))
.unwrap()
};
let first = service
.ready()
.await
.unwrap()
.call(request())
.await
.unwrap();
assert!(!first.headers().contains_key("openshell-replayed"));
let first = first.into_body().collect().await.unwrap();
assert_eq!(first.trailers().unwrap().get("grpc-status").unwrap(), "0");
let first = decode_unary_grpc_message::<SandboxResponse>(&first.to_bytes()).unwrap();
assert_eq!(
first
.sandbox
.as_ref()
.unwrap()
.metadata
.as_ref()
.unwrap()
.name,
"intercepted-create"
);
let replay = service
.ready()
.await
.unwrap()
.call(request())
.await
.unwrap();
assert_eq!(replay.headers().get("openshell-replayed").unwrap(), "true");
let replay = replay.into_body().collect().await.unwrap();
assert_eq!(
decode_unary_grpc_message::<SandboxResponse>(&replay.to_bytes()).unwrap(),
first
);
assert_eq!(interceptor.modifications.load(Ordering::SeqCst), 2);
assert_eq!(interceptor.validations.load(Ordering::SeqCst), 2);
assert_eq!(interceptor.observations.load(Ordering::SeqCst), 1);
interceptor.deny.store(true, Ordering::SeqCst);
let denied = service
.ready()
.await
.unwrap()
.call(request())
.await
.unwrap();
assert_eq!(denied.headers().get("grpc-status").unwrap(), "7");
assert_eq!(interceptor.validations.load(Ordering::SeqCst), 3);
assert_eq!(interceptor.observations.load(Ordering::SeqCst), 1);
task.abort();
}
#[test]
fn grpc_trailer_status_takes_precedence_over_headers() {
let response = Response::builder()
.header("grpc-status", "0")
.body(())
.unwrap();
let mut trailers = http::HeaderMap::new();
trailers.insert("grpc-status", HeaderValue::from_static("7"));
assert_eq!(
grpc_status_from_response_and_trailers(&response, Some(&trailers)),
"7"
);
}
#[tokio::test]
async fn http_response_includes_request_id() {
let addr = start_http_server_with_middleware().await;
let resp = http1_get(addr, "/healthz", &[]).await;
assert_eq!(resp.status(), 200);
let request_id = resp
.headers()
.get("x-request-id")
.expect("response should include x-request-id header");
let id_str = request_id.to_str().unwrap();
uuid::Uuid::parse_str(id_str).expect("should be a valid UUID");
}
#[tokio::test]
async fn http_preserves_client_request_id() {
let addr = start_http_server_with_middleware().await;
let client_id = "my-custom-correlation-id";
let resp = http1_get(addr, "/healthz", &[("x-request-id", client_id)]).await;
assert_eq!(resp.status(), 200);
let request_id = resp
.headers()
.get("x-request-id")
.expect("response should include x-request-id header");
assert_eq!(request_id.to_str().unwrap(), client_id);
}
#[tokio::test]
async fn each_request_gets_unique_id() {
let addr = start_http_server_with_middleware().await;
let mut ids = Vec::new();
for _ in 0..3 {
let resp = http1_get(addr, "/healthz", &[]).await;
let id = resp
.headers()
.get("x-request-id")
.unwrap()
.to_str()
.unwrap()
.to_string();
ids.push(id);
}
assert_ne!(ids[0], ids[1]);
assert_ne!(ids[1], ids[2]);
assert_ne!(ids[0], ids[2]);
}
#[tokio::test]
async fn grpc_path_includes_request_id() {
let addr = start_http_server_with_middleware().await;
let resp = http1_get(
addr,
"/openshell.v1.OpenShell/Health",
&[
("content-type", "application/grpc"),
("x-request-id", "grpc-corr-id"),
],
)
.await;
let request_id = resp
.headers()
.get("x-request-id")
.expect("gRPC-routed response should include x-request-id header");
assert_eq!(request_id.to_str().unwrap(), "grpc-corr-id");
}
#[derive(Clone)]
struct CountingGrpcService {
calls: Arc<AtomicUsize>,
}
impl Service<Request<()>> for CountingGrpcService {
type Response = Response<tonic::body::Body>;
type Error = Infallible;
type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, _req: Request<()>) -> Self::Future {
self.calls.fetch_add(1, Ordering::Relaxed);
std::future::ready(Ok(Response::new(tonic::body::Body::empty())))
}
}
/// Inner service that is never ready, used to prove the rate limiter does
/// not wait on inner-service backpressure when it is already exhausted.
/// Counts `call` invocations so tests can assert the limiter never forwards
/// to an inner service that did not report readiness.
#[derive(Clone)]
struct PendingInnerService {
calls: Arc<AtomicUsize>,
}
impl PendingInnerService {
fn new() -> Self {
Self {
calls: Arc::new(AtomicUsize::new(0)),
}
}
}
impl Service<Request<()>> for PendingInnerService {
type Response = Response<tonic::body::Body>;
type Error = Infallible;
type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Pending
}
fn call(&mut self, _req: Request<()>) -> Self::Future {
self.calls.fetch_add(1, Ordering::Relaxed);
std::future::ready(Ok(Response::new(tonic::body::Body::empty())))
}
}
#[tokio::test]
async fn grpc_rate_limit_poll_ready_short_circuits_exhausted_limiter() {
// An exhausted limiter must report ready even when the inner service is
// pending, so `call` returns RESOURCE_EXHAUSTED instead of waiting on
// inner backpressure.
let config = Config::new(None).with_grpc_rate_limit(Some(1), Some(60));
let limiter = GrpcRateLimiter::from_config(&config).expect("limiter should be enabled");
// Consume the single token so the limiter is exhausted.
assert!(limiter.allow());
let mut exhausted = GrpcRateLimitService::new(PendingInnerService::new(), Some(limiter));
let mut cx = Context::from_waker(std::task::Waker::noop());
assert!(
matches!(exhausted.poll_ready(&mut cx), Poll::Ready(Ok(()))),
"exhausted limiter should report ready despite a pending inner service",
);
let response = exhausted.call(Request::new(())).await.unwrap();
assert_eq!(grpc_status_from_response(&response), "8");
// A limiter with capacity must still respect inner backpressure.
let config = Config::new(None).with_grpc_rate_limit(Some(1), Some(60));
let limiter = GrpcRateLimiter::from_config(&config);
let mut with_capacity = GrpcRateLimitService::new(PendingInnerService::new(), limiter);
assert!(
with_capacity.poll_ready(&mut cx).is_pending(),
"limiter with capacity should defer to the pending inner service",
);
}
#[tokio::test]
async fn grpc_rate_limit_call_rejects_after_poll_ready_short_circuit_despite_window_rollover() {
// Regression: when `poll_ready` reports synthetic readiness for an
// exhausted limiter, it does NOT poll the inner service. If the
// rate-limit window then rolls over before `call`, the request must
// still be rejected rather than forwarded to an inner service that
// never reported readiness (a Tower contract violation).
let config = Config::new(None).with_grpc_rate_limit(Some(1), Some(60));
let limiter = GrpcRateLimiter::from_config(&config).expect("limiter should be enabled");
// Exhaust the single token.
assert!(limiter.allow());
// Pending inner service: its `poll_ready` never reports ready and its
// `call` increments a counter. A ready result from the wrapper
// therefore proves the limiter short-circuited rather than delegating,
// and `calls == 0` proves the wrapper never forwarded.
let inner = PendingInnerService::new();
let calls = inner.calls.clone();
let mut service = GrpcRateLimitService::new(inner, Some(limiter.clone()));
// poll_ready short-circuits the exhausted limiter and records synthetic
// readiness without polling the inner service.
let mut cx = Context::from_waker(std::task::Waker::noop());
assert!(
matches!(service.poll_ready(&mut cx), Poll::Ready(Ok(()))),
"exhausted limiter should report ready despite a pending inner service",
);
// The window rolls over between poll_ready and call: the limiter now
// has capacity again.
{
let mut state = limiter
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.window_started = state
.window_started
.checked_sub(Duration::from_secs(61))
.expect("test window rewind should be valid");
}
let response = service.call(Request::new(())).await.unwrap();
assert_eq!(grpc_status_from_response(&response), "8");
assert_eq!(
calls.load(Ordering::Relaxed),
0,
"inner service must not be called when poll_ready short-circuited the limiter",
);
}
#[tokio::test]
async fn grpc_rate_limit_returns_resource_exhausted_after_limit() {
let config = Config::new(None).with_grpc_rate_limit(Some(1), Some(60));
let limiter = GrpcRateLimiter::from_config(&config);
let calls = Arc::new(AtomicUsize::new(0));
let mut service = GrpcRateLimitService::new(
CountingGrpcService {
calls: calls.clone(),
},
limiter,
);
let first = service
.ready()
.await
.unwrap()
.call(Request::new(()))
.await
.unwrap();
assert_eq!(grpc_status_from_response(&first), "0");
let second = service
.ready()
.await
.unwrap()
.call(Request::new(()))
.await
.unwrap();
assert_eq!(grpc_status_from_response(&second), "8");
assert_eq!(calls.load(Ordering::Relaxed), 1);
}
#[tokio::test]
async fn grpc_rate_limit_disabled_passes_requests_through() {
let config = Config::new(None).with_grpc_rate_limit(Some(0), Some(60));
let limiter = GrpcRateLimiter::from_config(&config);
let calls = Arc::new(AtomicUsize::new(0));
let mut service = GrpcRateLimitService::new(
CountingGrpcService {
calls: calls.clone(),
},
limiter,
);
for _ in 0..3 {
let response = service
.ready()
.await
.unwrap()
.call(Request::new(()))
.await
.unwrap();
assert_eq!(grpc_status_from_response(&response), "0");
}
assert_eq!(calls.load(Ordering::Relaxed), 3);
}
#[tokio::test]
async fn grpc_rate_limit_resets_after_window() {
let config = Config::new(None).with_grpc_rate_limit(Some(1), Some(60));
let limiter = GrpcRateLimiter::from_config(&config).expect("limiter should be enabled");
let calls = Arc::new(AtomicUsize::new(0));
let mut service = GrpcRateLimitService::new(
CountingGrpcService {
calls: calls.clone(),
},
Some(limiter.clone()),
);
let first = service
.ready()
.await
.unwrap()
.call(Request::new(()))
.await
.unwrap();
assert_eq!(grpc_status_from_response(&first), "0");
{
let mut state = limiter
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
state.window_started = state
.window_started
.checked_sub(Duration::from_secs(61))
.expect("test window rewind should be valid");
}
let second = service
.ready()
.await
.unwrap()
.call(Request::new(()))
.await
.unwrap();
assert_eq!(grpc_status_from_response(&second), "0");
assert_eq!(calls.load(Ordering::Relaxed), 2);
}
#[tokio::test]
async fn grpc_rate_limit_state_is_shared_across_service_clones() {
let config = Config::new(None).with_grpc_rate_limit(Some(1), Some(60));
let limiter = GrpcRateLimiter::from_config(&config);
let calls = Arc::new(AtomicUsize::new(0));
let mut first_service = GrpcRateLimitService::new(
CountingGrpcService {
calls: calls.clone(),
},
limiter.clone(),
);
let mut second_service = GrpcRateLimitService::new(
CountingGrpcService {
calls: calls.clone(),
},
limiter,
);
let first = first_service
.ready()
.await
.unwrap()
.call(Request::new(()))
.await
.unwrap();
assert_eq!(grpc_status_from_response(&first), "0");
let second = second_service
.ready()
.await
.unwrap()
.call(Request::new(()))
.await
.unwrap();
assert_eq!(grpc_status_from_response(&second), "8");
assert_eq!(calls.load(Ordering::Relaxed), 1);
}
#[derive(Clone)]
struct TraceBuf(Arc<Mutex<Vec<u8>>>);
impl std::io::Write for TraceBuf {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(buf);
Ok(buf.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
#[test]
fn request_id_appears_in_trace_span() {
use tracing_subscriber::fmt::format::FmtSpan;
let log_buf: Arc<Mutex<Vec<u8>>> = Arc::new(Mutex::new(Vec::new()));
let writer = TraceBuf(log_buf.clone());
let fmt_layer = tracing_subscriber::fmt::layer()
.with_writer(move || writer.clone())
.with_ansi(false)
.with_span_events(FmtSpan::CLOSE);
let subscriber = {
use tracing_subscriber::layer::SubscriberExt as _;
tracing_subscriber::registry().with(fmt_layer)
};
{
let _traced = crate::otel_tracing::test_exporter::install_scoped(subscriber);
let req = Request::builder()
.uri("/test-path")
.header("x-request-id", "trace-test-id-12345")
.body(Empty::<Bytes>::new())
.unwrap();
let span = make_request_span(&req);
drop(span.enter());
drop(span);
}
let output = String::from_utf8(log_buf.lock().unwrap().clone()).unwrap();
assert!(
output.contains("trace-test-id-12345"),
"trace output should contain the request_id recorded in the span, got: {output}"
);
}
#[test]
fn polled_paths_get_debug_request_spans() {
let _traced = crate::otel_tracing::test_exporter::install_traced();
let level = |path: &str| {
let req = Request::builder()
.uri(path)
.body(Empty::<Bytes>::new())
.unwrap();
*make_request_span(&req)
.metadata()
.expect("span enabled")
.level()
};
for path in [
"/healthz",
"/openshell.v1.OpenShell/GetSandboxConfig",
"/openshell.v1.OpenShell/ReportProviderReadiness",
"/openshell.v1.OpenShell/PeerReportProviderReadiness",
] {
assert_eq!(level(path), tracing::Level::DEBUG, "{path}");
}
assert_eq!(
level("/openshell.v1.OpenShell/CreateSandbox"),
tracing::Level::INFO
);
}
#[test]
fn polled_path_responses_log_at_debug() {
let log_buf: Arc<Mutex<Vec<u8>>> = Arc::new(Mutex::new(Vec::new()));
let writer = TraceBuf(log_buf.clone());
let subscriber = {
use tracing_subscriber::layer::SubscriberExt as _;
tracing_subscriber::registry()
.with(tracing_subscriber::filter::LevelFilter::INFO)
.with(
tracing_subscriber::fmt::layer()
.with_writer(move || writer.clone())
.with_ansi(false),
)
};
{
let _traced = crate::otel_tracing::test_exporter::install_scoped(subscriber);
let respond = |path: &str, status: u16| {
let req = Request::builder()
.uri(path)
.body(Empty::<Bytes>::new())
.unwrap();
let res = Response::builder()
.status(status)
.body(Empty::<Bytes>::new())
.unwrap();
log_response(&res, Duration::from_millis(1), &make_request_span(&req));
};
respond("/openshell.v1.OpenShell/GetSandboxConfig", 200);
respond("/healthz", 200);
respond("/openshell.v1.OpenShell/CreateSandbox", 201);
respond("/healthz", 503);
}
let output = String::from_utf8(log_buf.lock().unwrap().clone()).unwrap();
let lines: Vec<&str> = output.lines().collect();
assert_eq!(lines.len(), 2, "got: {output}");
assert!(
lines[0].contains("INFO") && lines[0].contains("status=201"),
"got: {output}"
);
assert!(
lines[1].contains("WARN") && lines[1].contains("status=503"),
"got: {output}"
);
}
/// The `TraceLayer` creates the server span, so no gRPC handler needs
/// `#[instrument]`. The request ID carries into it so a trace can be
/// correlated with the gateway's logs.
#[tokio::test]
async fn request_span_exports_over_otlp_with_request_id() {
use crate::otel_tracing::test_exporter;
let traced = test_exporter::install_traced();
let req = Request::builder()
.uri("/openshell.v1.OpenShell/CreateSandbox")
.header("x-request-id", "otlp-req-id-9876")
.body(Empty::<Bytes>::new())
.unwrap();
let span = make_request_span(&req);
drop(span.enter());
drop(span);
let spans = traced.finished_spans();
let span = spans
.iter()
.find(|s| s.name == "openshell.v1.OpenShell/CreateSandbox")
.unwrap_or_else(|| {
panic!(
"the per-request span is recorded under its RPC name, got {:?}",
spans.iter().map(|s| &s.name).collect::<Vec<_>>()
)
});
assert_eq!(
test_exporter::attribute(span, "request_id").as_deref(),
Some("otlp-req-id-9876"),
);
assert_eq!(
test_exporter::attribute(span, "path").as_deref(),
Some("/openshell.v1.OpenShell/CreateSandbox"),
);
assert_eq!(
span.span_kind,
opentelemetry::trace::SpanKind::Server,
"trace UIs lay this out as a served call, not an internal operation"
);
test_exporter::assert_is_root(span);
assert_eq!(
test_exporter::attribute(span, "rpc.system").as_deref(),
Some("grpc"),
);
assert_eq!(
test_exporter::attribute(span, "rpc.service").as_deref(),
Some("openshell.v1.OpenShell"),
);
assert_eq!(
test_exporter::attribute(span, "rpc.method").as_deref(),
Some("CreateSandbox"),
);
}
#[tokio::test]
async fn request_span_continues_the_incoming_trace() {
use crate::otel_tracing::test_exporter;
let traced = test_exporter::install_traced();
let req = Request::builder()
.uri("/openshell.v1.OpenShell/CreateSandbox")
.header(
"traceparent",
"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
)
.body(Empty::<Bytes>::new())
.unwrap();
let span = make_request_span(&req);
drop(span.enter());
drop(span);
let span = traced.span_with(
"openshell.v1.OpenShell/CreateSandbox",
"rpc.method",
"CreateSandbox",
);
assert_eq!(
span.span_context.trace_id().to_string(),
"4bf92f3577b34da6a3ce929d0e0e4736"
);
assert_eq!(
span.parent_span_id.to_string(),
"00f067aa0ba902b7",
"the server span is a child of the caller's span"
);
}
/// A failed request must be distinguishable from a successful one in a
/// trace UI, which keys off span status rather than a logged field.
#[tokio::test]
async fn request_spans_record_the_response_outcome() {
use crate::otel_tracing::test_exporter;
let traced = test_exporter::install_traced();
for (path, status) in [
("/openshell.v1.OpenShell/CreateSandbox", 500),
("/openshell.v1.OpenShell/ListSandboxes", 200),
] {
let req = Request::builder()
.uri(path)
.body(Empty::<Bytes>::new())
.unwrap();
let span = make_request_span(&req);
let res = Response::builder()
.status(status)
.body(Empty::<Bytes>::new())
.unwrap();
let entered = span.enter();
log_response(&res, Duration::from_millis(3), &span);
drop(entered);
drop(span);
}
let spans = traced.finished_spans();
let failed = spans
.iter()
.find(|s| s.name == "openshell.v1.OpenShell/CreateSandbox")
.expect("failed request span recorded");
let succeeded = spans
.iter()
.find(|s| s.name == "openshell.v1.OpenShell/ListSandboxes")
.expect("successful request span recorded");
assert_eq!(
test_exporter::attribute(failed, "http.response.status_code").as_deref(),
Some("500"),
"the response status is an attribute, not only a log field"
);
assert!(
matches!(failed.status, opentelemetry::trace::Status::Error { .. }),
"the span carries error status so trace UIs flag it, got {:?}",
failed.status
);
assert!(
!matches!(succeeded.status, opentelemetry::trace::Status::Error { .. }),
"got {:?}",
succeeded.status
);
}
#[tokio::test]
async fn request_span_records_grpc_status_from_trailers() {
use crate::otel_tracing::test_exporter;
let traced = test_exporter::install_traced();
let req = Request::builder()
.uri("/openshell.v1.OpenShell/CreateSandbox")
.body(Empty::<Bytes>::new())
.unwrap();
let span = make_request_span(&req);
let mut trailers = http::HeaderMap::new();
trailers.insert("grpc-status", HeaderValue::from_static("13"));
record_response_trailers(Some(&trailers), Duration::from_millis(3), &span);
drop(span);
let span = traced.span_with(
"openshell.v1.OpenShell/CreateSandbox",
"rpc.method",
"CreateSandbox",
);
assert_eq!(
test_exporter::attribute(&span, "rpc.grpc.status_code").as_deref(),
Some("13")
);
assert!(
matches!(span.status, opentelemetry::trace::Status::Error { .. }),
"a non-OK gRPC trailer marks the span as failed"
);
}
/// Without upstream trace context, each inbound entrypoint roots a trace
/// named for its RPC or HTTP method.
#[tokio::test]
async fn each_entrypoint_gets_its_own_root_span() {
use crate::otel_tracing::test_exporter;
let paths = [
"/openshell.v1.OpenShell/CreateSandbox",
"/openshell.v1.OpenShell/ListSandboxes",
"/openshell.v1.OpenShell/DeleteSandbox",
"/metrics",
];
let traced = test_exporter::install_traced();
for path in paths {
let req = Request::builder()
.uri(path)
.body(Empty::<Bytes>::new())
.unwrap();
let span = make_request_span(&req);
drop(span.enter());
drop(span);
}
let names: std::collections::BTreeSet<String> = traced
.finished_spans()
.iter()
.map(|s| s.name.to_string())
.collect();
let expected = [
"GET",
"openshell.v1.OpenShell/CreateSandbox",
"openshell.v1.OpenShell/DeleteSandbox",
"openshell.v1.OpenShell/ListSandboxes",
]
.into_iter()
.map(String::from)
.collect::<std::collections::BTreeSet<_>>();
assert!(
expected.is_subset(&names),
"each entrypoint exports under its own name, got {names:?}"
);
assert!(
!names.contains("request"),
"no entrypoint falls back to the generic callsite name, got {names:?}"
);
}
/// gRPC spans are named for the RPC, per the OpenTelemetry RPC semantic
/// conventions (`$service/$method`).
#[test]
fn grpc_request_spans_are_named_for_the_rpc() {
assert_eq!(
otel_span_name(&http::Method::POST, "/openshell.v1.OpenShell/CreateSandbox"),
"openshell.v1.OpenShell/CreateSandbox"
);
}
/// Non-RPC paths use a low-cardinality method-only name because sandbox
/// application routes are opaque to the gateway.
#[test]
fn http_request_spans_do_not_include_the_literal_path() {
assert_eq!(otel_span_name(&http::Method::GET, "/users/12345"), "GET");
assert_eq!(otel_span_name(&http::Method::GET, "/users/67890"), "GET");
}
/// A path with no service segment must not produce a span named after a
/// stray slash or an empty string.
#[test]
fn bare_paths_fall_back_to_the_http_shape() {
assert_eq!(otel_span_name(&http::Method::GET, "/"), "GET");
assert_eq!(otel_span_name(&http::Method::POST, "/Foo"), "POST");
}
#[test]
fn grpc_method_extracts_last_segment() {
assert_eq!(
grpc_method_from_path("/openshell.v1.OpenShell/CreateSandbox"),
"CreateSandbox"
);
}
#[test]
fn grpc_method_handles_bare_path() {
assert_eq!(grpc_method_from_path("Health"), "Health");
}
#[test]
fn grpc_method_handles_single_slash() {
assert_eq!(grpc_method_from_path("/"), "");
}
#[test]
fn grpc_method_handles_empty_string() {
assert_eq!(grpc_method_from_path(""), "");
}
#[tokio::test]
async fn grpc_router_dispatches_gateway_and_reflection_paths() {
#[derive(Clone)]
struct RouteRecorder {
name: &'static str,
calls: Arc<Mutex<Vec<&'static str>>>,
}
impl<B: Send + 'static> Service<Request<B>> for RouteRecorder {
type Response = Response<tonic::body::Body>;
type Error = Infallible;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, _req: Request<B>) -> Self::Future {
self.calls.lock().unwrap().push(self.name);
Box::pin(async { Ok(Response::new(tonic::body::Body::empty())) })
}
}
let calls = Arc::new(Mutex::new(Vec::new()));
let service = |name| RouteRecorder {
name,
calls: calls.clone(),
};
let mut router = GrpcRouter::new(service("openshell"), service("reflection"));
for path in [
"/openshell.v1.OpenShell/Health",
"/grpc.reflection.v1.ServerReflection/ServerReflectionInfo",
] {
router
.call(
Request::builder()
.uri(path)
.body(Empty::<Bytes>::new())
.unwrap(),
)
.await
.unwrap();
}
assert_eq!(*calls.lock().unwrap(), vec!["openshell", "reflection"]);
}
#[tokio::test]
async fn reflection_protocol_serves_complete_public_descriptors_and_recovers_from_errors() {
use crate::auth::authenticator::test_support::MockAuthenticator;
use prost_reflect::{DescriptorPool, Value};
use tonic_reflection::pb::v1::{
ServerReflectionRequest, server_reflection_client::ServerReflectionClient,
server_reflection_request::MessageRequest, server_reflection_response::MessageResponse,
};
let reflection =
crate::reflection::build_gateway_reflection_service(&Config::new(None)).unwrap();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let unrouted = tower::service_fn(|_request: Request<BoxBody>| async {
Ok::<_, Infallible>(tonic::Status::unimplemented("test fallback").into_http())
});
let rejecting_oidc = Arc::new(MockAuthenticator::returning(Err(
crate::auth::authenticator::AuthRejection::new(
AuthMechanism::Oidc,
AuthFailureReason::InvalidCredential,
tonic::Status::unauthenticated("OIDC credentials required"),
),
)));
let grpc = AuthGrpcRouter::with_peer_identity(
GrpcRouter::new(unrouted, reflection),
Some(AuthenticatorChain::new(vec![rejecting_oidc])),
None,
None,
true,
false,
);
let service = MultiplexedService::new(grpc, unrouted);
let server = tokio::spawn(async move {
loop {
let (stream, _) = listener.accept().await.unwrap();
let service = service.clone();
tokio::spawn(async move {
Builder::new(TokioExecutor::new())
.serve_connection(TokioIo::new(stream), service)
.await
.unwrap();
});
}
});
let channel = tonic::transport::Channel::from_shared(format!("http://{addr}"))
.unwrap()
.connect()
.await
.unwrap();
let mut client = ServerReflectionClient::new(channel);
let list_request = ServerReflectionRequest {
host: String::new(),
message_request: Some(MessageRequest::ListServices(String::new())),
};
let descriptor_request = ServerReflectionRequest {
host: String::new(),
message_request: Some(MessageRequest::FileContainingSymbol(
"openshell.v1.OpenShell".to_string(),
)),
};
let missing_request = ServerReflectionRequest {
host: String::new(),
message_request: Some(MessageRequest::FileContainingSymbol(
"openshell.v1.DoesNotExist".to_string(),
)),
};
let list_after_error_request = ServerReflectionRequest {
host: String::new(),
message_request: Some(MessageRequest::ListServices(String::new())),
};
let mut responses = client
.server_reflection_info(tokio_stream::iter([
list_request,
descriptor_request,
missing_request,
list_after_error_request,
]))
.await
.unwrap()
.into_inner();
let response = responses.message().await.unwrap().unwrap();
let Some(MessageResponse::ListServicesResponse(response)) = response.message_response
else {
panic!("expected a reflection list-services response");
};
let mut names: Vec<_> = response
.service
.into_iter()
.map(|service| service.name)
.collect();
names.sort();
assert_eq!(names, vec!["openshell.v1.OpenShell"]);
let descriptor_response = responses.message().await.unwrap().unwrap();
let Some(MessageResponse::FileDescriptorResponse(response)) =
descriptor_response.message_response
else {
panic!("expected a reflection file-descriptor response");
};
let mut reflected_pool = DescriptorPool::new();
for descriptor in &response.file_descriptor_proto {
reflected_pool
.decode_file_descriptor_proto(descriptor.as_slice())
.unwrap();
}
assert!(
reflected_pool
.get_service_by_name("openshell.v1.OpenShell")
.is_some(),
"a fresh descriptor pool must resolve the advertised service"
);
assert!(
reflected_pool
.get_message_by_name("openshell.v1.HealthRequest")
.is_some(),
"the response must include imported public message descriptors"
);
let source_pool = DescriptorPool::decode(openshell_core::FILE_DESCRIPTOR_SET).unwrap();
let source_auth = source_pool
.get_extension_by_name("openshell.options.v1.authorization")
.unwrap();
let reflected_auth = reflected_pool
.get_extension_by_name("openshell.options.v1.authorization")
.unwrap();
let source_health = source_pool
.get_service_by_name("openshell.v1.OpenShell")
.unwrap()
.methods()
.find(|method| method.name() == "Health")
.unwrap();
let reflected_health = reflected_pool
.get_service_by_name("openshell.v1.OpenShell")
.unwrap()
.methods()
.find(|method| method.name() == "Health")
.unwrap();
let source_options = source_health.options();
let reflected_options = reflected_health.options();
let Value::Message(source_auth_value) = &*source_options.get_extension(&source_auth) else {
panic!("source authorization option must be a message");
};
let Value::Message(reflected_auth_value) =
&*reflected_options.get_extension(&reflected_auth)
else {
panic!("reflected authorization option must be a message");
};
assert_eq!(
source_auth_value.encode_to_vec(),
reflected_auth_value.encode_to_vec()
);
let source_secret = source_pool
.get_extension_by_name("openshell.options.v1.secret")
.unwrap();
let reflected_secret = reflected_pool
.get_extension_by_name("openshell.options.v1.secret")
.unwrap();
let source_field = source_pool
.get_message_by_name("openshell.v1.TcpForwardInit")
.unwrap()
.get_field_by_name("authorization_token")
.unwrap();
let reflected_field = reflected_pool
.get_message_by_name("openshell.v1.TcpForwardInit")
.unwrap()
.get_field_by_name("authorization_token")
.unwrap();
assert_eq!(
source_field.options().get_extension(&source_secret),
reflected_field.options().get_extension(&reflected_secret),
"secret-field annotations must retain their original value"
);
let error_response = responses.message().await.unwrap().unwrap();
let Some(MessageResponse::ErrorResponse(error)) = error_response.message_response else {
panic!("expected an in-band reflection error response");
};
assert_eq!(error.error_code, tonic::Code::NotFound as i32);
let response_after_error = responses.message().await.unwrap().unwrap();
assert!(matches!(
response_after_error.message_response,
Some(MessageResponse::ListServicesResponse(_))
));
server.abort();
}
#[tokio::test]
async fn reflection_protocol_resolves_canonical_enum_values_and_extensions() {
use prost::Message;
use tonic_reflection::pb::v1::{
ExtensionRequest, ServerReflectionRequest,
server_reflection_client::ServerReflectionClient,
server_reflection_request::MessageRequest, server_reflection_response::MessageResponse,
};
let reflection =
crate::reflection::build_gateway_reflection_service(&Config::new(None)).unwrap();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let unrouted = tower::service_fn(|_request: Request<BoxBody>| async {
Ok::<_, Infallible>(tonic::Status::unimplemented("test fallback").into_http())
});
let service = MultiplexedService::new(GrpcRouter::new(unrouted, reflection), unrouted);
let server = tokio::spawn(async move {
loop {
let (stream, _) = listener.accept().await.unwrap();
let service = service.clone();
tokio::spawn(async move {
Builder::new(TokioExecutor::new())
.serve_connection(TokioIo::new(stream), service)
.await
.unwrap();
});
}
});
let channel = tonic::transport::Channel::from_shared(format!("http://{addr}"))
.unwrap()
.connect()
.await
.unwrap();
let mut client = ServerReflectionClient::new(channel);
for (symbol, expected_file) in [
("openshell.v1.SANDBOX_PHASE_READY", "openshell.proto"),
(
"google.protobuf.FieldDescriptorProto.TYPE_DOUBLE",
"google/protobuf/descriptor.proto",
),
] {
let request = ServerReflectionRequest {
host: String::new(),
message_request: Some(MessageRequest::FileContainingSymbol(symbol.to_string())),
};
let mut responses = client
.server_reflection_info(tokio_stream::iter([request]))
.await
.unwrap()
.into_inner();
let response = responses.message().await.unwrap().unwrap();
let Some(MessageResponse::FileDescriptorResponse(response)) = response.message_response
else {
panic!("expected a descriptor response for canonical enum value {symbol}");
};
let names: std::collections::BTreeSet<_> = response
.file_descriptor_proto
.iter()
.map(|descriptor| {
prost_types::FileDescriptorProto::decode(descriptor.as_slice())
.unwrap()
.name
.unwrap()
})
.collect();
assert!(names.contains(expected_file));
}
for (containing_type, extension_number) in [
("google.protobuf.MethodOptions", 50_000),
("google.protobuf.FieldOptions", 50_001),
] {
let request = ServerReflectionRequest {
host: String::new(),
message_request: Some(MessageRequest::FileContainingExtension(ExtensionRequest {
containing_type: containing_type.to_string(),
extension_number,
})),
};
let mut responses = client
.server_reflection_info(tokio_stream::iter([request]))
.await
.unwrap()
.into_inner();
let response = responses.message().await.unwrap().unwrap();
let Some(MessageResponse::FileDescriptorResponse(response)) = response.message_response
else {
panic!("expected a descriptor response for extension {extension_number}");
};
let names: std::collections::BTreeSet<_> = response
.file_descriptor_proto
.iter()
.map(|descriptor| {
prost_types::FileDescriptorProto::decode(descriptor.as_slice())
.unwrap()
.name
.unwrap()
})
.collect();
assert!(names.contains("options.proto"));
}
let requests = [
ServerReflectionRequest {
host: String::new(),
message_request: Some(MessageRequest::AllExtensionNumbersOfType(
"google.protobuf.MethodOptions".to_string(),
)),
},
ServerReflectionRequest {
host: String::new(),
message_request: Some(MessageRequest::FileContainingExtension(ExtensionRequest {
containing_type: "google.protobuf.MethodOptions".to_string(),
extension_number: 49_999,
})),
},
ServerReflectionRequest {
host: String::new(),
message_request: Some(MessageRequest::ListServices(String::new())),
},
];
let mut responses = client
.server_reflection_info(tokio_stream::iter(requests))
.await
.unwrap()
.into_inner();
let extension_numbers = responses.message().await.unwrap().unwrap();
let Some(MessageResponse::AllExtensionNumbersResponse(extension_numbers)) =
extension_numbers.message_response
else {
panic!("expected an extension-number response");
};
assert_eq!(
extension_numbers.base_type_name,
"google.protobuf.MethodOptions"
);
assert_eq!(extension_numbers.extension_number, [50_000]);
let unknown_extension = responses.message().await.unwrap().unwrap();
let Some(MessageResponse::ErrorResponse(error)) = unknown_extension.message_response else {
panic!("expected an in-band error for an unknown extension");
};
assert_eq!(error.error_code, tonic::Code::NotFound as i32);
assert!(matches!(
responses.message().await.unwrap().unwrap().message_response,
Some(MessageResponse::ListServicesResponse(_))
));
server.abort();
}
#[tokio::test]
async fn reflection_rate_limit_charges_each_query_on_one_stream() {
use tonic::Code;
use tonic_reflection::pb::v1::{
ServerReflectionRequest, server_reflection_client::ServerReflectionClient,
server_reflection_request::MessageRequest,
};
let config = Config::new(None).with_grpc_rate_limit(Some(1), Some(60));
let reflection = crate::reflection::build_gateway_reflection_service(&config).unwrap();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let unrouted = tower::service_fn(|_request: Request<BoxBody>| async {
Ok::<_, Infallible>(tonic::Status::unimplemented("test fallback").into_http())
});
let service = MultiplexedService::new(GrpcRouter::new(unrouted, reflection), unrouted);
let server = tokio::spawn(async move {
loop {
let (stream, _) = listener.accept().await.unwrap();
let service = service.clone();
tokio::spawn(async move {
Builder::new(TokioExecutor::new())
.serve_connection(TokioIo::new(stream), service)
.await
.unwrap();
});
}
});
let channel = tonic::transport::Channel::from_shared(format!("http://{addr}"))
.unwrap()
.connect()
.await
.unwrap();
let mut client = ServerReflectionClient::new(channel);
let query = ServerReflectionRequest {
host: String::new(),
message_request: Some(MessageRequest::ListServices(String::new())),
};
let mut responses = client
.server_reflection_info(tokio_stream::iter([query.clone(), query]))
.await
.unwrap()
.into_inner();
assert!(responses.message().await.unwrap().is_some());
let status = responses
.message()
.await
.expect_err("second query on the same stream must be rate limited");
assert_eq!(status.code(), Code::ResourceExhausted);
server.abort();
}
#[test]
fn reflection_descriptor_excludes_internal_service_protos() {
let descriptors = crate::reflection::gateway_reflection_descriptors().unwrap();
let names: std::collections::BTreeSet<_> = descriptors
.iter()
.map(prost_reflect::FileDescriptor::name)
.collect();
assert!(names.contains("openshell.proto"));
assert!(names.contains("sandbox.proto"));
assert!(!names.contains("compute_driver.proto"));
assert!(!names.contains("credential_driver.proto"));
assert!(!names.contains("gateway_interceptor.proto"));
assert!(!names.contains("supervisor_middleware.proto"));
}
#[test]
fn normalize_ws_tunnel() {
assert_eq!(normalize_http_path("/_ws_tunnel"), "/_ws_tunnel");
}
#[test]
fn normalize_ws_tunnel_with_trailing() {
assert_eq!(normalize_http_path("/_ws_tunnel/foo"), "/_ws_tunnel");
}
#[test]
fn normalize_auth_path() {
assert_eq!(normalize_http_path("/auth/connect"), "/auth");
}
#[test]
fn normalize_auth_with_query() {
assert_eq!(
normalize_http_path("/auth/connect?callback_port=12345&code=AB7-X9KM"),
"/auth"
);
}
#[test]
fn normalize_unknown_path_collapses_to_unknown() {
assert_eq!(normalize_http_path("/random/scanner/probe"), "unknown");
}
#[test]
fn normalize_empty_path() {
assert_eq!(normalize_http_path(""), "unknown");
}
#[test]
fn normalize_root_path() {
assert_eq!(normalize_http_path("/"), "unknown");
}
mod auth_router {
use super::*;
use crate::auth::authenticator::AuthRejection;
use crate::auth::authenticator::test_support::MockAuthenticator;
use crate::auth::identity::{Identity, IdentityProvider};
use crate::auth::principal::{
Principal, SandboxIdentitySource, SandboxPrincipal, UserPrincipal,
};
use http_body_util::Full;
use jsonwebtoken::{Algorithm, EncodingKey, Header, encode};
use openshell_ocsf::test_support::OcsfCapture;
use openshell_ocsf::validation::schema::{
load_class_schema, validate_known_fields, validate_required_fields,
};
use serde::Serialize;
use std::sync::Arc;
use std::sync::Mutex;
use tower::Service;
type RecordedPrincipal = Arc<Mutex<Option<Principal>>>;
/// Service that snapshots the `Principal` from request extensions
/// and returns 200 OK. Used by router-level tests to assert the
/// chain's effect on the downstream service.
#[derive(Clone)]
struct PrincipalRecorder {
recorded: RecordedPrincipal,
}
impl PrincipalRecorder {
fn new() -> (Self, RecordedPrincipal) {
let recorded = Arc::new(Mutex::new(None));
(
Self {
recorded: recorded.clone(),
},
recorded,
)
}
}
impl<B: Send + 'static> Service<Request<B>> for PrincipalRecorder {
type Response = Response<tonic::body::Body>;
type Error = Infallible;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: Request<B>) -> Self::Future {
let principal = req.extensions().get::<Principal>().cloned();
*self.recorded.lock().unwrap() = principal;
Box::pin(async move { Ok(Response::new(tonic::body::Body::empty())) })
}
}
fn empty_request(path: &str) -> Request<Full<Bytes>> {
Request::builder()
.uri(path)
.body(Full::new(Bytes::new()))
.unwrap()
}
fn grpc_status<B>(res: &Response<B>) -> Option<String> {
res.headers()
.get("grpc-status")
.map(|v| v.to_str().unwrap().to_string())
}
fn user_principal(subject: &str) -> Principal {
Principal::User(UserPrincipal {
identity: Identity {
subject: subject.to_string(),
display_name: None,
roles: vec![],
scopes: vec![],
provider: IdentityProvider::Oidc,
},
})
}
fn provider_writer_principal() -> Principal {
Principal::User(UserPrincipal {
identity: Identity {
subject: "provider-rotation-job".to_string(),
display_name: None,
roles: vec!["openshell-admin".to_string()],
scopes: vec!["provider:write".to_string()],
provider: IdentityProvider::Oidc,
},
})
}
fn mtls_identity(subject: &str) -> Identity {
Identity {
subject: subject.to_string(),
display_name: Some(subject.to_string()),
roles: vec!["openshell-user".to_string()],
scopes: vec![],
provider: IdentityProvider::Mtls,
}
}
fn sandbox_principal() -> Principal {
Principal::Sandbox(SandboxPrincipal {
sandbox_id: "sandbox-a".to_string(),
source: SandboxIdentitySource::BootstrapJwt {
issuer: "openshell-gateway:test".to_string(),
},
trust_domain: Some("openshell".to_string()),
})
}
fn operator_principal() -> OperatorPrincipal {
let mut identity = mtls_identity("operator");
identity.roles = vec!["operator".to_string()];
OperatorPrincipal {
identity,
certificate_sha256: "verified-leaf".to_string(),
}
}
#[tokio::test]
async fn operator_is_admin_but_not_supervisor_peer_or_unknown() {
for (method, permitted) in [
("GetProviderCredentials", true),
("CreateWorkspace", true),
("UpdateProvider", true),
("GetSandboxConfig", true),
("ConnectSupervisor", false),
("PeerRelay", false),
("IssueSandboxToken", false),
("Unknown", false),
] {
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::with_peer_identity(
recorder,
None,
Some(AuthzPolicy {
admin_role: "oidc-admin".into(),
user_role: "oidc-user".into(),
scopes_enabled: true,
}),
None,
false,
false,
)
.with_operator(Some(operator_principal()));
let response = router
.call(empty_request(&format!("/openshell.v1.OpenShell/{method}")))
.await
.unwrap();
if permitted {
assert!(grpc_status(&response).is_none());
assert!(matches!(
*seen.lock().unwrap(),
Some(Principal::Operator(_))
));
} else {
assert_eq!(grpc_status(&response).as_deref(), Some("7"));
}
}
}
#[tokio::test]
async fn user_bearer_claims_and_local_dev_cannot_export_credentials() {
let mut user = mtls_identity("ordinary-user");
user.provider = IdentityProvider::Oidc;
user.roles = vec!["operator".into(), "openshell-admin".into()];
user.scopes = vec!["openshell:all".into()];
for principal in [
Principal::User(UserPrincipal { identity: user }),
sandbox_principal(),
Principal::Peer(crate::auth::principal::PeerPrincipal {
replica_id: "replica".into(),
pod_uid: "pod".into(),
}),
] {
let chain = AuthenticatorChain::new(vec![Arc::new(MockAuthenticator::returning(
Ok(Authentication::Accepted(principal)),
))]);
let (recorder, _) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let response = router
.call(empty_request(
"/openshell.v1.OpenShell/GetProviderCredentials",
))
.await
.unwrap();
assert_eq!(grpc_status(&response).as_deref(), Some("7"));
}
let (recorder, _) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, None, None);
let response = router
.call(empty_request(
"/openshell.v1.OpenShell/GetProviderCredentials",
))
.await
.unwrap();
assert_eq!(grpc_status(&response).as_deref(), Some("7"));
}
#[derive(Serialize)]
struct LegacySandboxClaims {
sub: String,
iss: String,
aud: String,
iat: i64,
exp: i64,
sandbox_id: String,
}
fn legacy_sandbox_token(material: &openshell_bootstrap::jwt::JwtKeyMaterial) -> String {
let claims = LegacySandboxClaims {
sub: "spiffe://openshell/sandbox/sandbox-a".to_string(),
iss: "openshell-gateway:test".to_string(),
aud: "openshell-gateway:test".to_string(),
iat: 1,
exp: 0,
sandbox_id: "sandbox-a".to_string(),
};
let mut header = Header::new(Algorithm::EdDSA);
header.kid = Some(material.kid.clone());
encode(
&header,
&claims,
&EncodingKey::from_ed_pem(material.signing_key_pem.as_bytes())
.expect("signing key"),
)
.expect("legacy token")
}
#[tokio::test]
async fn mtls_peer_identity_fills_missing_principal_when_enabled() {
let mock = Arc::new(MockAuthenticator::returning(Ok(
Authentication::NotApplicable,
)));
let chain = AuthenticatorChain::new(vec![mock]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::with_peer_identity(
recorder,
Some(chain),
None,
Some(mtls_identity("openshell-client")),
true,
false,
);
let res = router
.call(empty_request("/openshell.v1.OpenShell/ListSandboxes"))
.await
.unwrap();
assert_eq!(res.status(), 200);
let principal = seen.lock().unwrap().clone().expect("principal");
match principal {
Principal::User(u) => {
assert_eq!(u.identity.subject, "openshell-client");
assert_eq!(u.identity.provider, IdentityProvider::Mtls);
}
other => panic!("expected mTLS user principal, got {other:?}"),
}
}
#[tokio::test]
async fn mtls_peer_identity_authenticates_without_chain_when_enabled() {
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::with_peer_identity(
recorder,
None,
None,
Some(mtls_identity("openshell-client")),
true,
false,
);
let res = router
.call(empty_request("/openshell.v1.OpenShell/ListSandboxes"))
.await
.unwrap();
assert_eq!(res.status(), 200);
assert!(matches!(
seen.lock().unwrap().as_ref(),
Some(Principal::User(_))
));
}
#[tokio::test]
async fn mtls_auth_enabled_requires_peer_identity() {
let (recorder, seen) = PrincipalRecorder::new();
let mut router =
AuthGrpcRouter::with_peer_identity(recorder, None, None, None, true, false);
let res = router
.call(empty_request("/openshell.v1.OpenShell/ListSandboxes"))
.await
.unwrap();
assert!(seen.lock().unwrap().is_none());
assert_eq!(grpc_status(&res).as_deref(), Some("16"));
}
#[tokio::test]
async fn reflection_bypasses_oidc_and_mtls_user_authentication() {
let oidc = Arc::new(MockAuthenticator::returning(Err(AuthRejection::new(
AuthMechanism::Oidc,
AuthFailureReason::InvalidCredential,
tonic::Status::unauthenticated("OIDC credentials required"),
))));
let chain = AuthenticatorChain::new(vec![oidc]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router =
AuthGrpcRouter::with_peer_identity(recorder, Some(chain), None, None, true, false);
let res = router
.call(empty_request(
"/grpc.reflection.v1.ServerReflection/ServerReflectionInfo",
))
.await
.unwrap();
assert_eq!(res.status(), 200);
assert_eq!(grpc_status(&res), None);
assert!(
seen.lock().unwrap().is_none(),
"reflection must not receive an authenticated user principal"
);
}
#[tokio::test]
async fn unauthenticated_dev_user_fills_missing_principal_when_enabled() {
let mock = Arc::new(MockAuthenticator::returning(Ok(
Authentication::NotApplicable,
)));
let chain = AuthenticatorChain::new(vec![mock]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router =
AuthGrpcRouter::with_peer_identity(recorder, Some(chain), None, None, false, true);
let res = router
.call(empty_request("/openshell.v1.OpenShell/ListSandboxes"))
.await
.unwrap();
assert_eq!(res.status(), 200);
let principal = seen.lock().unwrap().clone().expect("principal");
match principal {
Principal::User(u) => {
assert_eq!(u.identity.subject, "unauthenticated-local-dev");
assert_eq!(u.identity.provider, IdentityProvider::LocalDev);
}
other => panic!("expected dev user principal, got {other:?}"),
}
}
#[tokio::test]
async fn unauthenticated_dev_user_authenticates_without_chain_when_enabled() {
let (recorder, seen) = PrincipalRecorder::new();
let mut router =
AuthGrpcRouter::with_peer_identity(recorder, None, None, None, false, true);
let res = router
.call(empty_request("/openshell.v1.OpenShell/ListSandboxes"))
.await
.unwrap();
assert_eq!(res.status(), 200);
assert!(matches!(
seen.lock().unwrap().as_ref(),
Some(Principal::User(user))
if user.identity.subject == "unauthenticated-local-dev"
));
}
#[tokio::test]
async fn user_principal_lands_in_request_extensions() {
let mock = Arc::new(MockAuthenticator::returning(Ok(Authentication::Accepted(
user_principal("alice"),
))));
let chain = AuthenticatorChain::new(vec![mock]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let _ = router
.call(empty_request("/openshell.v1.OpenShell/ListSandboxes"))
.await
.unwrap();
let principal = seen.lock().unwrap().clone().expect("principal");
match principal {
Principal::User(u) => assert_eq!(u.identity.subject, "alice"),
_ => panic!("expected user principal"),
}
}
#[tokio::test]
async fn sandbox_principal_lands_in_request_extensions() {
let mock = Arc::new(MockAuthenticator::returning(Ok(Authentication::Accepted(
sandbox_principal(),
))));
let chain = AuthenticatorChain::new(vec![mock]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let _ = router
.call(empty_request("/openshell.v1.OpenShell/ReportPolicyStatus"))
.await
.unwrap();
let captured = seen.lock().unwrap().clone();
match captured {
Some(Principal::Sandbox(p)) => assert_eq!(p.sandbox_id, "sandbox-a"),
other => panic!("expected sandbox principal, got {other:?}"),
}
}
#[tokio::test]
async fn sandbox_principal_can_call_allowlisted_method() {
let mock = Arc::new(MockAuthenticator::returning(Ok(Authentication::Accepted(
sandbox_principal(),
))));
let chain = AuthenticatorChain::new(vec![mock]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let res = router
.call(empty_request("/openshell.v1.OpenShell/GetSandboxConfig"))
.await
.unwrap();
assert_eq!(res.status(), 200);
assert!(matches!(
seen.lock().unwrap().as_ref(),
Some(Principal::Sandbox(_))
));
}
/// A user principal — even one carrying `openshell:all` and the
/// admin role — must not reach a `sandbox`-annotated method. The
/// router enforces this from the per-handler auth-mode declarations
/// independent of RBAC.
#[tokio::test]
async fn user_principal_is_denied_on_sandbox_only_methods() {
fn admin_user() -> Principal {
Principal::User(UserPrincipal {
identity: Identity {
subject: "admin".to_string(),
display_name: None,
roles: vec!["openshell-admin".to_string()],
scopes: vec!["openshell:all".to_string()],
provider: IdentityProvider::Oidc,
},
})
}
let policy = AuthzPolicy {
admin_role: "openshell-admin".to_string(),
user_role: "openshell-user".to_string(),
scopes_enabled: true,
};
for path in [
"/openshell.v1.OpenShell/ReportPolicyStatus",
"/openshell.v1.OpenShell/PushSandboxLogs",
"/openshell.v1.OpenShell/SubmitPolicyAnalysis",
"/openshell.v1.OpenShell/GetSandboxProviderEnvironment",
"/openshell.v1.OpenShell/ConnectSupervisor",
"/openshell.v1.OpenShell/RelayStream",
"/openshell.v1.OpenShell/IssueSandboxToken",
"/openshell.v1.OpenShell/RefreshSandboxToken",
] {
let mock = Arc::new(MockAuthenticator::returning(Ok(Authentication::Accepted(
admin_user(),
))));
let chain = AuthenticatorChain::new(vec![mock]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), Some(policy.clone()));
let res = router.call(empty_request(path)).await.unwrap();
assert!(seen.lock().unwrap().is_none(), "{path} reached handler");
// grpc-status=7 (PERMISSION_DENIED).
assert_eq!(grpc_status(&res).as_deref(), Some("7"), "{path}");
}
}
#[tokio::test]
async fn provider_write_scope_allows_update_without_provider_read() {
let policy = AuthzPolicy {
admin_role: "openshell-admin".to_string(),
user_role: "openshell-user".to_string(),
scopes_enabled: true,
};
let mock = Arc::new(MockAuthenticator::returning(Ok(Authentication::Accepted(
provider_writer_principal(),
))));
let chain = AuthenticatorChain::new(vec![mock]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), Some(policy.clone()));
let response = router
.call(empty_request(UPDATE_PROVIDER_PATH))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert!(seen.lock().unwrap().is_some());
let mock = Arc::new(MockAuthenticator::returning(Ok(Authentication::Accepted(
provider_writer_principal(),
))));
let chain = AuthenticatorChain::new(vec![mock]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), Some(policy));
let response = router
.call(empty_request("/openshell.v1.OpenShell/GetProvider"))
.await
.unwrap();
assert_eq!(grpc_status(&response).as_deref(), Some("7"));
assert!(seen.lock().unwrap().is_none());
}
#[tokio::test]
async fn sandbox_principal_is_denied_on_user_and_admin_methods() {
for path in [
"/openshell.v1.OpenShell/ListSandboxes",
"/openshell.v1.OpenShell/DeleteSandbox",
"/openshell.v1.OpenShell/CreateProvider",
"/openshell.v1.OpenShell/ApproveDraftChunk",
] {
let mock = Arc::new(MockAuthenticator::returning(Ok(Authentication::Accepted(
sandbox_principal(),
))));
let chain = AuthenticatorChain::new(vec![mock]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let res = router.call(empty_request(path)).await.unwrap();
assert!(seen.lock().unwrap().is_none(), "{path} reached handler");
assert_eq!(grpc_status(&res).as_deref(), Some("7"), "{path}");
}
}
#[tokio::test]
async fn missing_principal_returns_unauthenticated() {
let mock = Arc::new(MockAuthenticator::returning(Ok(
Authentication::NotApplicable,
)));
let chain = AuthenticatorChain::new(vec![mock]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let res = router
.call(empty_request("/openshell.v1.OpenShell/ListSandboxes"))
.await
.unwrap();
assert!(seen.lock().unwrap().is_none());
// tonic sets grpc-status=16 (UNAUTHENTICATED) in trailers.
assert_eq!(grpc_status(&res).as_deref(), Some("16"));
}
#[tokio::test]
async fn legacy_sandbox_jwt_cannot_reach_sensitive_sandbox_methods() {
let material = openshell_bootstrap::jwt::generate_jwt_key().expect("JWT key");
let authority = Arc::new(
crate::auth::sandbox_jwt::SandboxSessionJwtAuthority::from_pem(
material.signing_key_pem.as_bytes(),
material.public_key_pem.as_bytes(),
material.kid.clone(),
"test",
Some(Duration::from_mins(15)),
)
.expect("session authority"),
);
let store = Arc::new(
crate::persistence::Store::connect("sqlite::memory:")
.await
.expect("store"),
);
let token = legacy_sandbox_token(&material);
for path in [
"/openshell.v1.OpenShell/GetSandboxProviderEnvironment",
"/openshell.v1.OpenShell/ConnectSupervisor",
] {
let authenticator = Arc::new(
crate::auth::sandbox_jwt::SandboxSessionJwtAuthenticator::new(
authority.clone(),
store.clone(),
),
);
let chain = AuthenticatorChain::new(vec![authenticator]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let request = Request::builder()
.uri(path)
.header("authorization", format!("Bearer {token}"))
.body(Full::new(Bytes::new()))
.expect("request");
let response = router.call(request).await.expect("router response");
assert!(seen.lock().unwrap().is_none(), "{path} reached handler");
assert_eq!(grpc_status(&response).as_deref(), Some("16"), "{path}");
}
}
#[tokio::test]
async fn authenticator_error_short_circuits() {
let mock = Arc::new(MockAuthenticator::returning(Err(AuthRejection::new(
AuthMechanism::Oidc,
AuthFailureReason::InvalidCredential,
tonic::Status::unauthenticated("forged"),
))));
let chain = AuthenticatorChain::new(vec![mock]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let res = router
.call(empty_request("/openshell.v1.OpenShell/ListSandboxes"))
.await
.unwrap();
assert!(seen.lock().unwrap().is_none());
assert_eq!(grpc_status(&res).as_deref(), Some("16"));
}
fn captured_json(capture: &OcsfCapture) -> Vec<serde_json::Value> {
capture
.events()
.iter()
.map(|event| event.to_json().unwrap())
.collect()
}
fn assert_authentication_failure(
event: &serde_json::Value,
path: &str,
reason: &str,
severity_id: u8,
) {
let schema = load_class_schema("authentication");
validate_required_fields(event, &schema);
validate_known_fields(event, &schema);
assert_eq!(event["class_uid"], 3002);
assert_eq!(event["activity_id"], 1);
assert_eq!(event["status_id"], 2);
assert_eq!(event["action_id"], 2);
assert_eq!(event["disposition_id"], 2);
assert_eq!(event["severity_id"], severity_id);
assert_eq!(event["status_detail"], reason);
assert_eq!(event["user"]["type_id"], 0);
assert_eq!(event["service"]["name"], "openshell-gateway");
assert_eq!(event["http_request"]["url"]["path"], path);
assert_eq!(event["metadata"]["product"]["name"], "OpenShell Gateway");
}
#[tokio::test]
async fn missing_credential_emits_authentication_failure() {
let capture = OcsfCapture::default();
let _guard = capture.install();
let chain = AuthenticatorChain::new(vec![Arc::new(MockAuthenticator::returning(Ok(
Authentication::NotApplicable,
)))]);
let (recorder, _) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let path = "/openshell.v1.OpenShell/ListSandboxes";
let res = router.call(empty_request(path)).await.unwrap();
assert_eq!(grpc_status(&res).as_deref(), Some("16"));
let events = captured_json(&capture);
assert_eq!(events.len(), 1, "{events:?}");
assert_authentication_failure(&events[0], path, "missing_credential", 2);
assert!(events[0].get("auth_protocol_id").is_none());
}
#[tokio::test]
async fn missing_client_certificate_emits_mtls_authentication_failure() {
let capture = OcsfCapture::default();
let _guard = capture.install();
let (recorder, _) = PrincipalRecorder::new();
let mut router =
AuthGrpcRouter::with_peer_identity(recorder, None, None, None, true, false);
let path = "/openshell.v1.OpenShell/ListSandboxes";
let res = router.call(empty_request(path)).await.unwrap();
assert_eq!(grpc_status(&res).as_deref(), Some("16"));
let events = captured_json(&capture);
assert_eq!(events.len(), 1, "{events:?}");
assert_authentication_failure(&events[0], path, "missing_credential", 2);
assert_eq!(events[0]["auth_protocol_id"], 99);
assert_eq!(events[0]["auth_protocol"], "mtls");
}
#[tokio::test]
async fn operator_with_authorization_header_emits_conflicting_credentials() {
let capture = OcsfCapture::default();
let _guard = capture.install();
let (recorder, seen) = PrincipalRecorder::new();
let mut router =
AuthGrpcRouter::with_peer_identity(recorder, None, None, None, true, false)
.with_operator(Some(operator_principal()));
let path = "/openshell.v1.OpenShell/ListSandboxes";
let canary = "Bearer canary-secret-token-9f2c";
let request = Request::builder()
.uri(path)
.header("authorization", canary)
.body(Full::new(Bytes::new()))
.unwrap();
let res = router.call(request).await.unwrap();
assert!(seen.lock().unwrap().is_none());
assert_eq!(grpc_status(&res).as_deref(), Some("16"));
let events = captured_json(&capture);
assert_eq!(events.len(), 1, "{events:?}");
assert_authentication_failure(&events[0], path, "conflicting_credentials", 3);
assert_eq!(events[0]["auth_protocol"], "mtls");
assert!(!events[0].to_string().contains("canary-secret-token"));
}
async fn call_with_rejection(
rejection: AuthRejection,
path: &str,
) -> (tonic::Status, Vec<serde_json::Value>) {
let capture = OcsfCapture::default();
let _guard = capture.install();
let chain = AuthenticatorChain::new(vec![Arc::new(MockAuthenticator::returning(Err(
rejection,
)))]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let request = Request::builder()
.uri(path)
.header("authorization", "Bearer canary-secret-token-9f2c")
.body(Full::new(Bytes::new()))
.unwrap();
let res = router.call(request).await.unwrap();
assert!(seen.lock().unwrap().is_none());
let status = tonic::Status::from_header_map(res.headers()).expect("grpc status");
(status, captured_json(&capture))
}
#[tokio::test]
async fn declined_credential_emits_its_classification_with_the_missing_credential_status() {
let capture = OcsfCapture::default();
let _guard = capture.install();
let chain = AuthenticatorChain::new(vec![Arc::new(MockAuthenticator::returning(Ok(
Authentication::Declined(crate::auth::authenticator::AuthDecline {
mechanism: AuthMechanism::GatewayPeer,
reason: AuthFailureReason::InvalidCredential,
}),
)))]);
let (recorder, seen) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let path = "/openshell.v1.OpenShell/PeerRelay";
let res = router.call(empty_request(path)).await.unwrap();
assert!(seen.lock().unwrap().is_none());
let status = tonic::Status::from_header_map(res.headers()).expect("grpc status");
assert_eq!(status.code(), tonic::Code::Unauthenticated);
assert_eq!(status.message(), "missing authorization header");
let events = captured_json(&capture);
assert_eq!(events.len(), 1, "{events:?}");
assert_authentication_failure(&events[0], path, "invalid_credential", 3);
assert_eq!(events[0]["auth_protocol"], "k8s_service_account");
}
#[tokio::test]
async fn rejected_oidc_credential_emits_classified_authentication_failure() {
let path = "/openshell.v1.OpenShell/ListSandboxes";
let (status, events) = call_with_rejection(
AuthRejection::new(
AuthMechanism::Oidc,
AuthFailureReason::ExpiredCredential,
tonic::Status::unauthenticated("invalid token: ExpiredSignature"),
),
path,
)
.await;
assert_eq!(status.code(), tonic::Code::Unauthenticated);
assert_eq!(status.message(), "invalid token: ExpiredSignature");
assert_eq!(events.len(), 1, "{events:?}");
assert_authentication_failure(&events[0], path, "expired_credential", 2);
assert_eq!(events[0]["auth_protocol_id"], 4);
assert_eq!(events[0]["auth_protocol"], "OpenID");
assert!(!events[0].to_string().contains("canary-secret-token"));
assert!(!events[0].to_string().contains("ExpiredSignature"));
}
#[tokio::test]
async fn rejected_sandbox_credential_emits_classified_authentication_failure() {
let path = "/openshell.v1.OpenShell/GetSandboxProviderEnvironment";
let (status, events) = call_with_rejection(
AuthRejection::new(
AuthMechanism::SandboxJwt,
AuthFailureReason::StaleIdentity,
tonic::Status::unauthenticated(
"gateway token does not match the active sandbox identity",
),
),
path,
)
.await;
assert_eq!(status.code(), tonic::Code::Unauthenticated);
assert_eq!(events.len(), 1, "{events:?}");
assert_authentication_failure(&events[0], path, "stale_identity", 3);
assert_eq!(events[0]["auth_protocol_id"], 99);
assert_eq!(events[0]["auth_protocol"], "sandbox_jwt");
}
#[tokio::test]
async fn authenticated_and_exempt_requests_emit_no_authentication_event() {
let capture = OcsfCapture::default();
let _guard = capture.install();
let chain = AuthenticatorChain::new(vec![Arc::new(MockAuthenticator::returning(Ok(
Authentication::Accepted(user_principal("alice")),
)))]);
let (recorder, _) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
for path in [
"/openshell.v1.OpenShell/ListSandboxes",
"/openshell.v1.OpenShell/Health",
] {
let res = router.call(empty_request(path)).await.unwrap();
assert!(grpc_status(&res).is_none(), "{path}");
}
let events = captured_json(&capture);
assert!(events.is_empty(), "{events:?}");
}
#[tokio::test]
async fn health_methods_bypass_chain() {
// Authenticator is wired to fail-closed; the request still gets
// through because the path is exempt.
let mock = Arc::new(MockAuthenticator::returning(Err(AuthRejection::new(
AuthMechanism::Oidc,
AuthFailureReason::InvalidCredential,
tonic::Status::unauthenticated("would reject"),
))));
let chain = AuthenticatorChain::new(vec![mock.clone()]);
let (recorder, _) = PrincipalRecorder::new();
let mut router = AuthGrpcRouter::new(recorder, Some(chain), None);
let res = router
.call(empty_request("/openshell.v1.OpenShell/Health"))
.await
.unwrap();
assert_eq!(res.status(), 200);
assert_eq!(mock.call_count(), 0, "health must not consult the chain");
}
}
}