feat(embed)!: in-process upstream, pass-through and your own TLS on one listener

Cargo.toml

@@ -32,8 +32,9 @@ required-features = ["cli"]
 doc = false
 
 [dependencies]
-# HTTP framework
-axum = { version = "0.8", features = ["macros"] }
+# HTTP framework. `http2`: `serve` answers HTTP/1.1 and HTTP/2 on one port,
+# so native gRPC clients share the listener with REST ones.
+axum = { version = "0.8", features = ["macros", "http2"] }
 tower = "0.5"
 tower-http = { version = "0.7", features = ["cors", "trace"] }
 # Foundational HTTP types, used directly by the framework-agnostic embedding
@@ -42,9 +43,19 @@ tower-http = { version = "0.7", features = ["cors", "trace"] }
 # explicit.
 http = "1"
 bytes = "1"
-
-# gRPC client (to upstream service)
-tonic = "0.14"
+# The body bound of an upstream's responses and of requests `ProxyService`
+# accepts; already in the tree through hyper, axum and tonic.
+http-body = "1"
+# The response futures of `ProxyService`, named without boxing each one.
+pin-project-lite = "0.2"
+
+# gRPC client (to upstream service). `tls-connect-info` gives a rustls server
+# stream the connection record (client certificates) a tonic handler reads, so
+# an in-process upstream behind the proxy gets it; tonic exports the type only
+# with its TLS backends, so it is named through that stream (`tokio-rustls`,
+# the same crate tonic uses). Neither links a crypto provider.
+tonic = { version = "0.14", features = ["tls-connect-info"] }
+tokio-rustls = { version = "0.26", default-features = false }
 tonic-health = "0.14"
 # Canonical google.rpc.Status / error_details descriptors (FILE_DESCRIPTOR_SET)
 # and the Status message used to decode `grpc-status-details-bin`, so REST error
@@ -158,7 +169,8 @@ redis = ["dep:redis"]
 cli = ["dep:clap", "dep:tracing-subscriber", "redis"]
 
 [dev-dependencies]
-tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
+# `test-util`: deadline tests run on a paused clock instead of waiting.
+tokio = { version = "1", features = ["macros", "rt-multi-thread", "test-util"] }
 tower = { version = "0.5", features = ["util"] }
 http-body-util = "0.1"
 # Trailer frames for the hand-written upstream response in
@@ -169,6 +181,12 @@ http-body = "1"
 # features would pull in aws-lc.
 tokio-rustls = { version = "0.26", default-features = false, features = ["tls12"] }
 rustls-rustcrypto = "0.0.2-alpha"
+# The embedder-owned TLS server of tests/tls.rs (HTTP/1.1 and HTTP/2 on one
+# connection type), and the TLS stream a tonic client dials through.
+hyper-util = { version = "0.1", features = ["server-auto", "service", "tokio"] }
+# An upstream that speaks gRPC-Web, the way an embedder gives it that
+# protocol (tests/edge.rs).
+tonic-web = "0.14"
 # benches/jwt_verify.rs. Plots and rayon are left out: numbers are enough.
 criterion = { version = "0.8", default-features = false, features = ["async_tokio", "cargo_bench_support"] }
 # The embedding-hooks integration test (tests/hooks.rs) writes hook impls using

README.md

@@ -21,13 +21,15 @@ Works with **any** gRPC service via proto descriptor files. No code generation,
 - **Server-streaming** RPC → NDJSON by default, or Server-Sent Events via `Accept: text/event-stream` negotiation
 - **gRPC → HTTP status mapping** following the standard `google.rpc.Code` table
 - **Typed error details**: the upstream's `google.rpc.Status` details (`ErrorInfo`, `BadRequest`, `RetryInfo`, ...) reach the HTTP client as ProtoJSON, switchable globally and per route (see [Error responses](#error-responses))
+- **One port for REST and native gRPC**: HTTP/1.1 and HTTP/2 on the same listener, gRPC and gRPC-Web requests pass through to the upstream unchanged (an upstream that speaks gRPC-Web, e.g. behind tonic-web, answers it); behind your own TLS too, with the client's address and certificate reaching the upstream
+- **In-process upstream** for embedders: transcoded calls reach your own tonic services with no socket or loopback hop (see [Library Usage](#library-usage))
 - **Header forwarding** from HTTP requests to gRPC metadata (configurable allow-list)
 - **Context propagation**: W3C trace-context (`traceparent` forwarded or synthesized) and client deadlines (`grpc-timeout`) carried across the REST↔gRPC boundary
 - **Path aliasing** for route remapping (e.g. `/oauth2/*` → `/v1/oauth2/*`)
 - **Maintenance mode** returning 503 with a configurable exempt-path list
 - **Health endpoints** `/health/live`, `/health/ready` (upstream gRPC health probe), `/health/startup`
 - **Prometheus metrics** at `/metrics`
-- **CORS** with a configurable origin allow-list
+- **CORS** with a configurable origin allow-list, exposed headers and preflight cache, applied to gRPC-Web pass-through too
 - **Rate limiting (Shield)**: local GCRA shaper (no blocking latency) keyed by client IP, header, or validated JWT claim; named limit tiers as config data; optional async cross-instance reconciliation for an approximate fleet-wide limit (requires both the `redis` feature and a configured `sync` block)
 - **JWT auth**: validate `Bearer` tokens via an Ed25519 PEM key or JWKS auto-discovery, enforce per-route `require_auth` / `required_roles`, and forward claims as headers — or hand the signature check to your own verifier (a validated / FIPS module, an HSM) without changing anything else
 - **OIDC discovery**: serve `/.well-known/openid-configuration` and a JWKS endpoint (Ed25519) built from config, to front an identity provider
@@ -66,6 +68,8 @@ log line (`RUST_LOG=info`) states the count and where it came from.
 listen:
   http: "0.0.0.0:8080"
 
+# The gRPC service behind the proxy. Required by the standalone binary; an
+# embedder with an in-process upstream leaves it out.
 upstream:
   default: "http://127.0.0.1:50051"
 
@@ -79,10 +83,24 @@ service:
 
 cors:
   # Empty list = permissive CORS (dev mode, reflects any Origin).
-  # A non-empty list allows those exact origins; there is no "*" wildcard
-  # (browsers never send `Origin: *`, so listing "*" would block everything).
+  # A non-empty list allows those exact origins, with credentials; the
+  # preflight echoes the methods and headers the browser asks for. There is
+  # no "*" wildcard (browsers never send `Origin: *`, so listing "*" would
+  # block everything).
   origins: []
   # e.g. origins: ["https://app.example.com", "https://admin.example.com"]
+  # Response headers a browser script may read, on top of grpc-status,
+  # grpc-message, grpc-status-details-bin and the rate-limit headers:
+  # typically upstream metadata forwarded as a header.
+  expose_headers: []
+  # e.g. expose_headers: ["x-request-id"]
+  # How long a browser caches a preflight answer (seconds). Unset: the
+  # browser's default.
+  # max_age_secs: 600
+  # Apply this policy to gRPC-Web calls passed through to the upstream and to
+  # their preflights. Turn off only when the upstream sets CORS on gRPC-Web
+  # itself: its preflights then reach the upstream too.
+  grpc_web: true
 
 # Optional: path aliases (rewrite before routing)
 aliases:
@@ -530,7 +548,7 @@ details are malformed is dropped along with it (see
 [Error responses](#error-responses)). Browsers read only
 [CORS-safelisted](https://fetch.spec.whatwg.org/#cors-safelisted-response-header-name)
 response headers plus the exposed ones, so a browser client that must read a
-forwarded header needs a CORS setup that exposes it.
+forwarded header needs it listed in `cors.expose_headers`.
 
 **Status from `x-http-code`.** On a successful unary call, the response
 metadata `x-http-code` (grpc-gateway's convention) sets the HTTP status: one
@@ -613,7 +631,117 @@ async fn main() -> anyhow::Result<()> {
 }
 ```
 
-Or build the axum `Router` yourself for custom serving / embedding:
+`serve` answers HTTP/1.1 and HTTP/2 on one port: REST requests go to the
+proxy's routes, and requests with a gRPC or gRPC-Web content type go to the
+upstream as they arrived, so native gRPC clients can use the same address.
+
+### Your own gRPC services as the upstream
+
+A gRPC service that embeds the proxy to add REST (a forward-auth decision
+service, an API that also speaks gRPC) hands its own services to the proxy
+instead of an address. Transcoded calls then reach them in process: no
+socket, no loopback connection, no second HTTP/2 round, and they pass through
+the service's whole tonic stack (interceptors, layers) like a native gRPC
+call. `Request::remote_addr` in a handler gives the HTTP client's address.
+
+```rust
+use structured_proxy::ProxyServer;
+
+# async fn run() -> anyhow::Result<()> {
+// Your services, exactly as you would give them to tonic's server.
+let grpc = tonic::service::Routes::default(); // .add_service(MyServer::new(...))
+
+// No `upstream:` in the config: the upstream is `grpc`.
+let proxy = ProxyServer::from_file(std::path::Path::new("my-service.yaml"))?
+    .service(grpc)?;
+
+// REST and native gRPC on one port.
+let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?;
+structured_proxy::serve(listener, proxy).await?;
+# Ok(())
+# }
+```
+
+`ProxyServer::service` takes any gRPC tower service
+(`structured_proxy::upstream::Upstream`): `tonic::service::Routes`, a remote
+`tonic::transport::Channel` (what `ProxyServer::upstream` builds from the
+config), or anything else that speaks gRPC over `http` types. The result is a
+tower service, so it can also run on a server of your own.
+
+### Behind your own TLS
+
+`serve` speaks cleartext. For TLS, run the service on your own acceptor: one
+`ProxyService::for_connection` call per accepted connection tells the proxy
+who is on the other end, so its middleware sees the client's address and a
+tonic handler in process reads it with `Request::remote_addr`, and the client
+certificate with `Request::peer_certs` (mTLS), for native and transcoded calls
+alike. Advertise `h2` next to `http/1.1` in ALPN so gRPC clients get HTTP/2.
+
+```rust
+use std::sync::Arc;
+use hyper_util::rt::{TokioExecutor, TokioIo};
+use hyper_util::server::conn::auto::Builder;
+use hyper_util::service::TowerToHyperService;
+use structured_proxy::{ConnectionInfo, ProxyServer};
+use tonic::transport::server::Connected;
+
+# async fn run(mut tls: rustls::ServerConfig, grpc: tonic::service::Routes) -> anyhow::Result<()> {
+let proxy = ProxyServer::from_file(std::path::Path::new("my-service.yaml"))?.service(grpc)?;
+tls.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
+let acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(tls));
+let listener = tokio::net::TcpListener::bind("0.0.0.0:8443").await?;
+loop {
+    let (tcp, _) = listener.accept().await?;
+    let (acceptor, proxy) = (acceptor.clone(), proxy.clone());
+    tokio::spawn(async move {
+        // The handshake runs in the connection's task, so a slow client
+        // does not hold up the others.
+        let Ok(stream) = acceptor.accept(tcp).await else { return };
+        let service = proxy.for_connection(ConnectionInfo::tls(stream.connect_info()));
+        let served = Builder::new(TokioExecutor::new())
+            .serve_connection(TokioIo::new(stream), TowerToHyperService::new(service))
+            .await;
+        if let Err(error) = served {
+            tracing::debug!(%error, "connection ended");
+        }
+    });
+}
+# }
+```
+
+A plain TCP connection passes `stream.connect_info()` directly
+(`ConnectionInfo` converts from tonic's `TcpConnectInfo`).
+
+What the proxy does not serve is not its business: a request no route matches
+gets `404`, or goes to a service of yours with
+`ProxyService::with_fallback(my_axum_app)`. The proxy's middleware (CORS,
+maintenance, rate limits, auth) sees neither those requests nor native gRPC
+ones; they reach your service untouched.
+
+gRPC-Web requests pass through unchanged as well, so the upstream answers
+them in that protocol: wrap your services in tonic-web's layer
+(`tower::ServiceBuilder::new().layer(tonic_web::GrpcWebLayer::new()).service(grpc)`),
+binary and text gRPC-Web alike. When the upstream cannot take a call at all,
+the proxy's own error answer keeps the request's protocol. Browsers get the
+proxy's CORS policy on these calls, the same one their preflight got
+(`cors.grpc_web`, on by default); a gRPC-Web client reads `grpc-status`,
+`grpc-message` and `grpc-status-details-bin`, which are always exposed. A
+browser's preflight for a gRPC-Web call (one announcing `x-grpc-web`) goes
+where the call goes: the proxy answers it under that policy, a fallback never
+sees it, and with `cors.grpc_web: false` it reaches the upstream, whose own
+CORS policy then covers preflight and call alike.
+
+**Deadlines.** Every call waits at most five seconds for the upstream's
+response headers, or less when the client's `grpc-timeout` says so; after that
+the client gets `504` `DEADLINE_EXCEEDED`. The proxy enforces this itself, in
+process and remote alike. The client's `grpc-timeout` travels to the upstream;
+the five-second default does not, so an upstream that applies `grpc-timeout`
+to a whole call does not cut a long server stream short.
+
+### Merging into an axum application
+
+`ProxyServer::router` returns the proxy's HTTP routes in front of the
+configured upstream address, to serve or to merge into your own axum `Router`:
 
 ```rust
 use std::path::Path;

src/config.rs

@@ -15,8 +15,12 @@ use std::path::PathBuf;
 /// `#[non_exhaustive]` instead, since those are deserialized, not hand-built.
 #[derive(Debug, Clone, Deserialize)]
 pub struct ProxyConfig {
-    /// Upstream gRPC service(s).
-    pub upstream: UpstreamConfig,
+    /// The remote gRPC upstream. Required by the standalone proxy and by
+    /// [`ProxyServer::upstream`](crate::ProxyServer::upstream); an embedder
+    /// whose upstream is in process
+    /// ([`ProxyServer::service`](crate::ProxyServer::service)) leaves it out.
+    #[serde(default)]
+    pub upstream: Option<UpstreamConfig>,
 
     /// Proto descriptor sources.
     #[serde(default, deserialize_with = "deserialize_descriptor_sources")]
@@ -998,12 +1002,39 @@ impl Default for MaintenanceConfig {
 }
 
 /// CORS configuration.
-#[derive(Debug, Clone, Default, Deserialize)]
+#[derive(Debug, Clone, Deserialize)]
 #[non_exhaustive]
 pub struct CorsConfig {
     /// Allowed origins. Empty = permissive (dev mode).
     #[serde(default)]
     pub origins: Vec<String>,
+    /// Response headers a browser script may read on top of the ones the
+    /// proxy always exposes (`grpc-status`, `grpc-message`,
+    /// `grpc-status-details-bin` and the rate-limit headers): typically
+    /// upstream metadata forwarded as a header, such as `x-request-id`.
+    #[serde(default)]
+    pub expose_headers: Vec<String>,
+    /// How long a browser may cache a preflight answer, in seconds. Unset,
+    /// the browser's own default applies.
+    #[serde(default)]
+    pub max_age_secs: Option<u64>,
+    /// Apply this policy to gRPC-Web calls that pass through to the upstream
+    /// and to their preflights, so a browser gets one policy for both (on by
+    /// default). Off only for an upstream that sets CORS on gRPC-Web itself:
+    /// its preflights then reach the upstream as well.
+    #[serde(default = "default_true")]
+    pub grpc_web: bool,
+}
+
+impl Default for CorsConfig {
+    fn default() -> Self {
+        Self {
+            origins: Vec::new(),
+            expose_headers: Vec::new(),
+            max_age_secs: None,
+            grpc_web: true,
+        }
+    }
 }
 
 /// Logging configuration.

src/config/tests.rs

@@ -7,7 +7,7 @@ upstream:
   default: "grpc://localhost:4180"
 "#;
     let config: ProxyConfig = serde_yaml::from_str(yaml).unwrap();
-    assert_eq!(config.upstream.default, "grpc://localhost:4180");
+    assert_eq!(config.upstream.unwrap().default, "grpc://localhost:4180");
     assert_eq!(config.listen.http, "0.0.0.0:8080");
     assert_eq!(config.service.name, "structured-proxy");
     assert_eq!(config.streaming.sse_keep_alive_secs, 15);
@@ -16,6 +16,44 @@ upstream:
     assert!(config.shield.is_none());
 }
 
+#[test]
+fn cors_defaults_cover_grpc_web_and_nothing_extra() {
+    // Absent `cors:` and an empty one agree: permissive origins, gRPC-Web
+    // included, no extra exposed headers, the browser's own preflight cache.
+    for yaml in ["service:\n  name: demo\n", "cors: {}\n"] {
+        let cors = serde_yaml::from_str::<ProxyConfig>(yaml).unwrap().cors;
+        assert!(cors.origins.is_empty(), "{yaml}");
+        assert!(cors.expose_headers.is_empty(), "{yaml}");
+        assert_eq!(cors.max_age_secs, None, "{yaml}");
+        assert!(cors.grpc_web, "{yaml}");
+    }
+}
+
+#[test]
+fn cors_settings_are_read() {
+    let yaml = "cors:\n  origins: [\"https://app.example\"]\n  expose_headers: [\"x-request-id\"]\n  max_age_secs: 600\n  grpc_web: false\n";
+    let cors = serde_yaml::from_str::<ProxyConfig>(yaml).unwrap().cors;
+    assert_eq!(cors.origins, ["https://app.example"]);
+    assert_eq!(cors.expose_headers, ["x-request-id"]);
+    assert_eq!(cors.max_age_secs, Some(600));
+    assert!(!cors.grpc_web);
+}
+
+#[test]
+fn upstream_is_optional() {
+    // An embedder whose upstream is in process names no address.
+    let config: ProxyConfig = serde_yaml::from_str("service:\n  name: demo\n").unwrap();
+    assert!(config.upstream.is_none());
+}
+
+#[test]
+fn upstream_without_its_address_is_rejected() {
+    // A present `upstream:` block must say where: an empty one is a mistake,
+    // not a request for an in-process upstream.
+    let err = serde_yaml::from_str::<ProxyConfig>("upstream: {}\n").unwrap_err();
+    assert!(err.to_string().contains("default"), "{err}");
+}
+
 #[test]
 fn jwks_max_age_defaults_and_overrides() {
     let yaml = "upstream:\n  default: \"grpc://x:1\"\nauth:\n  mode: jwt\n  jwt:\n    jwks_uri: \"https://idp/jwks\"\n";
@@ -231,7 +269,10 @@ forwarded_headers:
   - "x-request-id"
 "#;
     let config: ProxyConfig = serde_yaml::from_str(yaml).unwrap();
-    assert_eq!(config.upstream.default, "grpc://sid-identity:4180");
+    assert_eq!(
+        config.upstream.as_ref().unwrap().default,
+        "grpc://sid-identity:4180"
+    );
     assert_eq!(config.listen.http, "0.0.0.0:9090");
     assert_eq!(config.service.name, "sid-proxy");
     assert_eq!(config.aliases.len(), 1);

src/embed.rs

@@ -20,7 +20,6 @@ use axum::routing::{get, on, MethodFilter, MethodRouter};
 use axum::{Json, Router};
 
 use crate::hooks::{AuthDecider, Decision, ExtraRoute, OidcBackend, RequestParts, RouteRequest};
-use crate::ProxyState;
 
 /// Cap on the body an extra-route handler will buffer (16 MiB). Extra routes are
 /// a stateless escape hatch, not a bulk-upload path; a bounded buffer keeps a
@@ -115,7 +114,10 @@ pub(crate) async fn verify_via_decider(
 }
 
 /// Routes for the stateless OIDC surface supplied by an [`OidcBackend`].
-pub(crate) fn oidc_backend_routes(backend: Arc<dyn OidcBackend>) -> Router<ProxyState> {
+pub(crate) fn oidc_backend_routes<S>(backend: Arc<dyn OidcBackend>) -> Router<S>
+where
+    S: Clone + Send + Sync + 'static,
+{
     let mut router = Router::new();
 
     // Static metadata documents (openid-configuration, provider-specific docs).
@@ -180,10 +182,13 @@ pub(crate) fn oidc_backend_routes(backend: Arc<dyn OidcBackend>) -> Router<Proxy
 ///
 /// Routes that share a path but differ in method are merged into one
 /// [`MethodRouter`], so registering `GET /x` and `POST /x` does not panic.
-pub(crate) fn extra_routes_router(routes: &[ExtraRoute]) -> Router<ProxyState> {
+pub(crate) fn extra_routes_router<S>(routes: &[ExtraRoute]) -> Router<S>
+where
+    S: Clone + Send + Sync + 'static,
+{
     use std::collections::HashMap;
 
-    let mut by_path: HashMap<String, MethodRouter<ProxyState>> = HashMap::new();
+    let mut by_path: HashMap<String, MethodRouter<S>> = HashMap::new();
     for route in routes {
         let Ok(filter) = MethodFilter::try_from(route.method.clone()) else {
             tracing::warn!(