Skip to main content

structured_proxy/
lib.rs

1//! Universal gRPC→REST transcoding proxy.
2//!
3//! Config-driven: same binary, different YAML = different product proxy.
4//! Works with ANY gRPC service via proto descriptors as config.
5//!
6//! ## Usage
7//!
8//! ```bash
9//! structured-proxy --config sid-proxy.yaml
10//! structured-proxy --config sflow-proxy.yaml
11//! ```
12//!
13//! ## JWT verification
14//!
15//! The bearer-token check sits behind [`hooks::TokenVerifier`]. A build gets one
16//! of two:
17//!
18//! - the **built-in** verifier (keys from `auth.jwt`), whose crypto backend is
19//!   picked by a feature: `rust_crypto` (default, pure Rust) or `aws_lc_rs`
20//!   (opt-in, constant-time / FIPS-capable, links aws-lc via C FFI). Both may be
21//!   compiled in at once — Cargo features are additive, so a dependency graph
22//!   with two dependents asking for different backends unifies into exactly that
23//!   build. `aws_lc_rs` then wins: it is constant-time and free of the `rsa`
24//!   advisory `rust_crypto` carries. [`ProxyServer::from_config`] settles that
25//!   choice for the process; a process where another crate may reach
26//!   `jsonwebtoken` before any server exists calls
27//!   [`install_default_crypto_provider`] from `main` instead.
28//! - an **injected** one, supplied by the embedder through
29//!   [`ProxyServer::with_token_verifier`]. Since Cargo unifies features across a
30//!   whole dependency graph, a backend feature cannot be chosen per binary —
31//!   injection is how a consumer that needs a different one gets it without
32//!   deciding for everyone else who links this crate. Such a build takes
33//!   `default-features = false` and links no JWT crypto at all.
34//!
35//! ## Outbound TLS
36//!
37//! JWKS fetches and the rate-limit service go over rustls. The crypto provider
38//! is the one the process installed with
39//! `rustls::crypto::CryptoProvider::install_default`, else the one the crypto
40//! backend feature brings (aws-lc for `aws_lc_rs`, RustCrypto for
41//! `rust_crypto`). A build with neither feature links no TLS crypto, and a
42//! config that needs an outbound client then fails at startup unless a
43//! provider is installed first.
44
45// `builtin_jwt` is implied by each backend and never meant to stand alone: on
46// its own it would link jsonwebtoken with no provider, which panics at runtime.
47#[cfg(all(
48    feature = "builtin_jwt",
49    not(any(feature = "rust_crypto", feature = "aws_lc_rs"))
50))]
51compile_error!(
52    "feature `builtin_jwt` needs a crypto backend: enable `rust_crypto` or `aws_lc_rs` \
53     (or neither, and inject a verifier with `ProxyServer::with_token_verifier`)"
54);
55
56/// The README's Rust examples, compiled as doc tests.
57#[cfg(doctest)]
58#[doc = include_str!("../README.md")]
59struct ReadmeDoctests;
60
61pub mod auth;
62pub mod client_address;
63pub mod config;
64mod cors;
65mod embed;
66mod guard;
67mod held;
68pub mod hooks;
69pub mod oidc;
70pub mod openapi;
71pub mod received;
72mod serve;
73pub mod service;
74pub mod shield;
75mod tls;
76pub mod transcode;
77pub mod upstream;
78
79/// Settle the process-wide JWT crypto provider. See
80/// [`install_default_crypto_provider`] for when a call is needed.
81#[cfg(feature = "builtin_jwt")]
82pub use auth::crypto::install_default_crypto_provider;
83pub use client_address::ClientAddress;
84pub use received::ReceivedRequest;
85pub use serve::{serve, serve_with, serve_with_shutdown, ServeOptions};
86pub use service::{ConnectionInfo, ProxyService};
87
88use axum::extract::State;
89use axum::http::StatusCode;
90use axum::response::IntoResponse;
91use axum::routing::get;
92use axum::{Json, Router};
93use cors::CorsLayer;
94use prost_reflect::DescriptorPool;
95use std::net::SocketAddr;
96use tower_http::trace::TraceLayer;
97
98use std::sync::Arc;
99
100use config::{DescriptorSource, ProxyConfig, ScopeConfig};
101use hooks::{AuthDecider, ExtraRoute, OidcBackend, TokenVerifier};
102use upstream::Upstream;
103
104/// What the proxy's handlers share. Every request extracts its own clone, so
105/// it holds the upstream handle and reference-counted settings only.
106#[derive(Clone, Debug)]
107pub(crate) struct ProxyState<U> {
108    /// The gRPC service transcoded calls and readiness probes go to.
109    pub(crate) upstream: U,
110    /// Headers to forward from HTTP to gRPC.
111    pub(crate) forwarded_headers: Arc<[String]>,
112    /// The same, as header names.
113    pub(crate) forwarded_names: Arc<[http::HeaderName]>,
114    /// SSE keep-alive interval (seconds) for server-streaming responses.
115    pub(crate) sse_keep_alive_secs: u64,
116}
117
118/// What [`ProxyServer::routes`] builds.
119struct Built {
120    /// Every route behind its layers, for a router.
121    router: Router,
122    /// The choice among the transcoded bindings, made outside the router's
123    /// layers; `None` without transcoded routes.
124    chooser: Option<transcode::Chooser>,
125    /// The transcoded routes served past any router, and the other routes;
126    /// `None` without transcoded routes, or when an other route's path is
127    /// one they cannot be ranked among.
128    direct: Option<(transcode::Direct, Router)>,
129    /// The CORS policy the routes answer under.
130    cors: CorsLayer,
131    /// The guards, for the traffic outside the routes.
132    guards: Arc<guard::Guards>,
133}
134
135/// Universal proxy server.
136pub struct ProxyServer {
137    config: ProxyConfig,
138    /// Optional pre-loaded descriptor pool (for embedded mode).
139    descriptor_pool: Option<DescriptorPool>,
140    /// Optional in-process forward-auth/PDP gate (embedded Tier-2 hook).
141    auth_decider: Option<Arc<dyn AuthDecider>>,
142    /// The traffic the injected decider gates; `transcoded` when unset.
143    auth_decider_scope: Option<ScopeConfig>,
144    /// Optional stateless OIDC surface backing (embedded Tier-2 hook).
145    oidc_backend: Option<Arc<dyn OidcBackend>>,
146    /// Embedder-supplied extra stateless routes (embedded Tier-2 hook).
147    extra_routes: Vec<ExtraRoute>,
148    /// Override for the `/verify` forward-auth path of an injected AuthDecider.
149    verify_path: Option<String>,
150    /// Embedder-supplied JWT verifier, replacing the built-in one.
151    token_verifier: Option<Arc<dyn TokenVerifier>>,
152    /// How the transcoded routes render errors and frame NDJSON streams.
153    transcode: transcode::TranscodeOptions,
154}
155
156impl Default for ProxyServer {
157    /// [`ProxyServer::new`]: every capability off.
158    fn default() -> Self {
159        Self::new()
160    }
161}
162
163impl ProxyServer {
164    /// Create from YAML config file.
165    pub fn from_config(config: ProxyConfig) -> Self {
166        // Earliest point this crate owns: settle the JWT crypto provider here,
167        // long before the first token arrives. A process whose other crates
168        // reach jsonwebtoken before any server exists calls
169        // `install_default_crypto_provider` from `main` instead.
170        #[cfg(feature = "builtin_jwt")]
171        auth::crypto::install_default_crypto_provider();
172
173        Self {
174            config,
175            descriptor_pool: None,
176            auth_decider: None,
177            auth_decider_scope: None,
178            oidc_backend: None,
179            extra_routes: Vec::new(),
180            verify_path: None,
181            token_verifier: None,
182            transcode: transcode::TranscodeOptions::default(),
183        }
184    }
185
186    /// Create from a YAML document: the [`ProxyConfig`] plus the transcoding
187    /// settings it does not hold (`error_details:`,
188    /// `streaming.ndjson_envelope` and `response_headers:`), applied as
189    /// [`with_error_details`], [`with_ndjson_envelope`] and
190    /// [`with_denied_response_headers`] would. A top-level or `streaming:` key
191    /// no setting reads is logged as a warning.
192    ///
193    /// # Errors
194    ///
195    /// Invalid YAML, a [`ProxyConfig`] that fails
196    /// [`validate`](ProxyConfig::validate), an `error_details` route pattern
197    /// that is relative or not a valid glob, or a `response_headers.deny`
198    /// entry that is not a header name.
199    ///
200    /// [`with_error_details`]: Self::with_error_details
201    /// [`with_ndjson_envelope`]: Self::with_ndjson_envelope
202    /// [`with_denied_response_headers`]: Self::with_denied_response_headers
203    pub fn from_yaml_str(yaml: &str) -> anyhow::Result<Self> {
204        let config = ProxyConfig::from_yaml_str(yaml)?;
205        for key in config::unknown_config_keys(yaml) {
206            tracing::warn!(%key, "unknown config key is ignored");
207        }
208        let settings: config::TranscodeFileConfig = serde_yaml::from_str(yaml)?;
209        let options = settings
210            .options()
211            .map_err(|e| anyhow::anyhow!("invalid transcoding config: {e}"))?;
212        let mut server = Self::from_config(config);
213        server.transcode = options;
214        Ok(server)
215    }
216
217    /// [`from_yaml_str`](Self::from_yaml_str) on the contents of a file.
218    ///
219    /// # Errors
220    ///
221    /// The file cannot be read, or its contents are rejected by
222    /// [`from_yaml_str`](Self::from_yaml_str).
223    pub fn from_file(path: &std::path::Path) -> anyhow::Result<Self> {
224        Self::from_yaml_str(&std::fs::read_to_string(path)?)
225    }
226
227    /// The configuration this server was created with.
228    pub fn config(&self) -> &ProxyConfig {
229        &self.config
230    }
231
232    /// A proxy with every capability off: native gRPC and gRPC-Web pass to
233    /// the upstream, every other request to the fallback (`404` by default).
234    /// Turn on what you need with the methods below; each sets the config
235    /// section of the same name, so code and YAML describe one proxy.
236    ///
237    /// # Examples
238    ///
239    /// ```
240    /// use structured_proxy::config::{ConcurrencyConfig, ScopeConfig, Traffic};
241    /// use structured_proxy::ProxyServer;
242    ///
243    /// # fn build() -> anyhow::Result<()> {
244    /// // Your gRPC API, with at most 1000 calls in flight and nothing else.
245    /// let grpc = tonic::service::Routes::default();
246    /// let service = ProxyServer::new()
247    ///     .with_concurrency_limit(
248    ///         ConcurrencyConfig::new(1000).with_scope(ScopeConfig::traffic([Traffic::Grpc])),
249    ///     )
250    ///     .service(grpc)?;
251    /// # let _ = service;
252    /// # Ok(())
253    /// # }
254    /// # build().unwrap();
255    /// ```
256    pub fn new() -> Self {
257        let mut config = ProxyConfig::default();
258        config.health.enabled = false;
259        config.metrics.enabled = false;
260        Self::from_config(config)
261    }
262
263    /// The remote gRPC upstream (`upstream.default`), for
264    /// [`upstream`](Self::upstream) and [`serve`](Self::serve).
265    pub fn with_upstream_address(mut self, address: impl Into<String>) -> Self {
266        self.config.upstream = Some(config::UpstreamConfig {
267            default: address.into(),
268        });
269        self
270    }
271
272    /// The listener of [`serve`](Self::serve): address, TLS, connection
273    /// limit (`listen:`).
274    pub fn with_listen(mut self, listen: config::ListenConfig) -> Self {
275        self.config.listen = listen;
276        self
277    }
278
279    /// Transcode only these services (`package.Service`) and methods
280    /// (`package.Service/Method`) of the descriptors (`transcode.only`); every
281    /// annotated RPC by default. A name the descriptors do not hold fails the
282    /// build.
283    pub fn with_transcoded_rpcs(
284        mut self,
285        names: impl IntoIterator<Item = impl Into<String>>,
286    ) -> Self {
287        self.config.transcode.only = names.into_iter().map(Into::into).collect();
288        self
289    }
290
291    /// Path aliases rewritten before routing (`aliases:`).
292    pub fn with_aliases(mut self, aliases: impl IntoIterator<Item = config::AliasConfig>) -> Self {
293        self.config.aliases = aliases.into_iter().collect();
294        self
295    }
296
297    /// The request headers transcoded calls forward as gRPC metadata
298    /// (`forwarded_headers:`), replacing the default list.
299    pub fn with_forwarded_headers(
300        mut self,
301        names: impl IntoIterator<Item = impl Into<String>>,
302    ) -> Self {
303        self.config.forwarded_headers = names.into_iter().map(Into::into).collect();
304        self
305    }
306
307    /// Health probe endpoints (`health:`).
308    pub fn with_health(mut self, health: config::HealthConfig) -> Self {
309        self.config.health = health;
310        self
311    }
312
313    /// The Prometheus metrics endpoint (`metrics:`).
314    pub fn with_metrics(mut self, metrics: config::MetricsConfig) -> Self {
315        self.config.metrics = metrics;
316        self
317    }
318
319    /// The OpenAPI spec and docs endpoints (`openapi:`).
320    pub fn with_openapi(mut self, openapi: config::OpenApiConfig) -> Self {
321        self.config.openapi = Some(openapi);
322        self
323    }
324
325    /// Static OIDC discovery and JWKS endpoints (`oidc_discovery:`).
326    pub fn with_oidc_discovery(mut self, oidc: config::OidcDiscoveryConfig) -> Self {
327        self.config.oidc_discovery = Some(oidc);
328        self
329    }
330
331    /// The CORS policy (`cors:`).
332    pub fn with_cors(mut self, cors: config::CorsConfig) -> Self {
333        self.config.cors = cors;
334        self
335    }
336
337    /// Translate gRPC-Web to gRPC for an upstream that speaks only gRPC
338    /// (`grpc_web.translate`).
339    pub fn with_grpc_web_translation(mut self, translate: bool) -> Self {
340        self.config.grpc_web.translate = translate;
341        self
342    }
343
344    /// Server-streaming behavior (`streaming:`).
345    pub fn with_streaming(mut self, streaming: config::StreamingConfig) -> Self {
346        self.config.streaming = streaming;
347        self
348    }
349
350    /// Maintenance mode (`maintenance:`).
351    pub fn with_maintenance(mut self, maintenance: config::MaintenanceConfig) -> Self {
352        self.config.maintenance = maintenance;
353        self
354    }
355
356    /// The limit on requests in flight (`concurrency:`).
357    pub fn with_concurrency_limit(mut self, concurrency: config::ConcurrencyConfig) -> Self {
358        self.config.concurrency = Some(concurrency);
359        self
360    }
361
362    /// How the client's address is resolved (`client_address:`): the proxies
363    /// trusted to report it and the header they use. Every consumer reads the
364    /// result, the rate limits as much as the upstream.
365    ///
366    /// # Examples
367    ///
368    /// ```
369    /// use structured_proxy::config::{ClientAddressConfig, ForwardingHeader};
370    /// use structured_proxy::ProxyServer;
371    ///
372    /// # fn build() -> anyhow::Result<()> {
373    /// // Behind a load balancer in 10.0.0.0/8 that appends to X-Forwarded-For.
374    /// let mut client_address = ClientAddressConfig::default();
375    /// client_address.trusted_proxies = vec!["10.0.0.0/8".into()];
376    /// client_address.header = ForwardingHeader::XForwardedFor;
377    /// client_address.required = true;
378    /// let service = ProxyServer::new()
379    ///     .with_client_address(client_address)
380    ///     .service(tonic::service::Routes::default())?;
381    /// # let _ = service;
382    /// # Ok(())
383    /// # }
384    /// # build().unwrap();
385    /// ```
386    pub fn with_client_address(mut self, client_address: config::ClientAddressConfig) -> Self {
387        self.config.client_address = client_address;
388        self
389    }
390
391    /// Rate limits (`shield:`); build the section with [`config::from_yaml`].
392    pub fn with_rate_limits(mut self, shield: config::ShieldConfig) -> Self {
393        self.config.shield = Some(shield);
394        self
395    }
396
397    /// JWT auth, route policies, forward-auth and ext_authz (`auth:`); build
398    /// the section with [`config::from_yaml`].
399    pub fn with_auth(mut self, auth: config::AuthConfig) -> Self {
400        self.config.auth = Some(auth);
401        self
402    }
403
404    /// Create with an embedded descriptor pool (for sid-proxy backward compat).
405    pub fn with_descriptors(mut self, pool: DescriptorPool) -> Self {
406        self.descriptor_pool = Some(pool);
407        self
408    }
409
410    /// Inject an in-process forward-auth / PDP decision (embedded Tier-2 hook).
411    ///
412    /// The decider gates the transcoded requests inline (other traffic with
413    /// [`with_auth_decider_scope`](Self::with_auth_decider_scope)) and also
414    /// backs the `/verify` forward-auth endpoint. Its signature is `axum`-free
415    /// (see [`hooks::AuthDecider`]), so the embedder never names an HTTP
416    /// framework.
417    pub fn with_auth_decider(mut self, decider: Arc<dyn AuthDecider>) -> Self {
418        self.auth_decider = Some(decider);
419        self
420    }
421
422    /// Choose the traffic the injected [`AuthDecider`] gates, `transcoded`
423    /// by default; the `/verify` endpoint is never behind it, since it answers
424    /// for the decider.
425    ///
426    /// # Examples
427    ///
428    /// ```
429    /// use structured_proxy::config::{ScopeConfig, Traffic};
430    /// use structured_proxy::ProxyServer;
431    ///
432    /// # fn build() -> anyhow::Result<()> {
433    /// // Gate native gRPC calls as well as the transcoded ones.
434    /// let server = ProxyServer::from_yaml_str("service:\n  name: demo\n")?
435    ///     .with_auth_decider_scope(ScopeConfig::traffic([Traffic::Transcoded, Traffic::Grpc]));
436    /// # let _ = server;
437    /// # Ok(())
438    /// # }
439    /// # build().unwrap();
440    /// ```
441    pub fn with_auth_decider_scope(mut self, scope: ScopeConfig) -> Self {
442        self.auth_decider_scope = Some(scope);
443        self
444    }
445
446    /// Back the stateless OIDC surface (discovery, JWKS, userinfo) with the
447    /// embedder's key/client metadata (embedded Tier-2 hook).
448    ///
449    /// When set, this supersedes the config-driven static `oidc_discovery`
450    /// routes. See [`hooks::OidcBackend`].
451    pub fn with_oidc_backend(mut self, backend: Arc<dyn OidcBackend>) -> Self {
452        self.oidc_backend = Some(backend);
453        self
454    }
455
456    /// Register extra stateless routes through an `axum`-free adapter (embedded
457    /// Tier-2 hook). See [`hooks::ExtraRoute`] / [`hooks::ExtraRouteHandler`].
458    pub fn with_extra_routes(mut self, routes: impl IntoIterator<Item = ExtraRoute>) -> Self {
459        self.extra_routes.extend(routes);
460        self
461    }
462
463    /// Set the path at which the injected [`AuthDecider`] answers forward-auth
464    /// sub-requests (`/verify`). Independent of any JWT `forward_auth` config, so
465    /// a decider-only embedder can place it without a JWT block.
466    ///
467    /// Resolution order for the path: this override, then
468    /// `auth.forward_auth.path` from config, then the default `/auth/verify`.
469    pub fn with_verify_path(mut self, path: impl Into<String>) -> Self {
470        self.verify_path = Some(path.into());
471        self
472    }
473
474    /// Verify bearer tokens with the embedder's own verifier instead of the
475    /// built-in one (embedded Tier-2 hook).
476    ///
477    /// The JWT middleware keeps everything around the signature check — route
478    /// policies, the roles claim, claim→header forwarding — and takes the
479    /// verdict from [`hooks::TokenVerifier`]. Use this when the built-in crypto
480    /// backend is not the one this binary needs: a validated / FIPS module, an
481    /// HSM, or a verifier the embedder already owns. It is also the way out of
482    /// Cargo's feature unification, which makes `rust_crypto` / `aws_lc_rs` a
483    /// property of the whole dependency graph rather than of one binary.
484    ///
485    /// `auth.mode` must still be `"jwt"` for the middleware to run; `auth.jwt`
486    /// then only configures claim forwarding, and any key source in it is
487    /// ignored (with a warning), since the verifier owns its own keys.
488    pub fn with_token_verifier(mut self, verifier: Arc<dyn TokenVerifier>) -> Self {
489        self.token_verifier = Some(verifier);
490        self
491    }
492
493    /// Choose which transcoded routes return the upstream's
494    /// `google.rpc.Status` details in their error bodies. Every route does by
495    /// default; see [`transcode::error::ErrorDetailsPolicy`] to switch them
496    /// off globally or per route.
497    pub fn with_error_details(mut self, policy: transcode::error::ErrorDetailsPolicy) -> Self {
498        self.transcode = self.transcode.with_error_details(policy);
499        self
500    }
501
502    /// Wrap NDJSON stream lines in `{"result"}` / `{"error"}` envelopes; see
503    /// [`transcode::TranscodeOptions::with_ndjson_envelope`].
504    pub fn with_ndjson_envelope(mut self, enabled: bool) -> Self {
505        self.transcode = self.transcode.with_ndjson_envelope(enabled);
506        self
507    }
508
509    /// Keep these upstream response metadata keys off the HTTP responses of
510    /// the transcoded routes; see
511    /// [`transcode::TranscodeOptions::with_denied_response_headers`].
512    pub fn with_denied_response_headers(
513        mut self,
514        names: impl IntoIterator<Item = http::HeaderName>,
515    ) -> Self {
516        self.transcode = self.transcode.with_denied_response_headers(names);
517        self
518    }
519
520    /// Load descriptor pool from configured sources.
521    ///
522    /// Multiple descriptor files are merged into a single pool,
523    /// enabling multi-service proxying from one binary.
524    fn load_descriptors(&self) -> anyhow::Result<DescriptorPool> {
525        if let Some(pool) = &self.descriptor_pool {
526            return Ok(pool.clone());
527        }
528
529        let mut pool = DescriptorPool::new();
530
531        for source in &self.config.descriptors {
532            match source {
533                DescriptorSource::File { file } => {
534                    let bytes = std::fs::read(file).map_err(|e| {
535                        anyhow::anyhow!("Failed to read descriptor file {:?}: {}", file, e)
536                    })?;
537                    pool.decode_file_descriptor_set(bytes.as_slice())
538                        .map_err(|e| {
539                            anyhow::anyhow!("Failed to decode descriptor file {:?}: {}", file, e)
540                        })?;
541                    tracing::info!("Loaded descriptor from {:?}", file);
542                }
543                DescriptorSource::Reflection { reflection } => {
544                    tracing::warn!(
545                        "gRPC reflection client not supported — use descriptor files instead (reflection endpoint: {})",
546                        reflection
547                    );
548                }
549                DescriptorSource::Embedded { bytes } => {
550                    pool.decode_file_descriptor_set(*bytes).map_err(|e| {
551                        anyhow::anyhow!("Failed to decode embedded descriptors: {}", e)
552                    })?;
553                }
554            }
555        }
556
557        Ok(pool)
558    }
559
560    /// The path an injected [`AuthDecider`] answers `/verify` at: the
561    /// `with_verify_path` override, then `auth.forward_auth.path`, then the
562    /// default `/auth/verify`. Only meaningful when a decider is set (the
563    /// override does not apply to config-driven JWT forward-auth).
564    fn decider_verify_path(&self) -> String {
565        self.verify_path.clone().unwrap_or_else(|| {
566            self.config
567                .auth
568                .as_ref()
569                .and_then(|a| a.forward_auth.as_ref())
570                .map(|fa| fa.path.clone())
571                .unwrap_or_else(|| "/auth/verify".to_string())
572        })
573    }
574
575    /// The verify path that is ACTUALLY mounted, or `None` when no verify route
576    /// is mounted. This is what the collision guard and maintenance-exempt list
577    /// must use, since the two mount sites use different paths:
578    /// - an injected decider mounts at [`decider_verify_path`](Self::decider_verify_path)
579    ///   (the `with_verify_path` override applies), whereas
580    /// - config-driven JWT forward-auth mounts `forward_auth.routes()` at
581    ///   `auth.forward_auth.path` (the override does NOT apply, and it mounts
582    ///   only when `auth.mode == "jwt"`, since the endpoint shares the built JWT
583    ///   `Auth`).
584    fn mounted_verify_path(&self) -> Option<String> {
585        if self.auth_decider.is_some() {
586            return Some(self.decider_verify_path());
587        }
588        self.config.auth.as_ref().and_then(|a| {
589            if a.mode != "jwt" {
590                return None;
591            }
592            a.forward_auth
593                .as_ref()
594                .filter(|fa| fa.enabled)
595                .map(|fa| fa.path.clone())
596        })
597    }
598
599    /// Every `(method, path)` route of the proxy's own endpoints and the
600    /// embedder's extra routes, the verify endpoint aside. With the
601    /// transcoded routes, it checks the mounted edge for a real collision,
602    /// reported as a clear error instead of an axum duplicate-route panic,
603    /// and ranks the transcoded paths among the others for the proxy
604    /// service. `method` is the uppercase HTTP token; same-path routes with
605    /// different methods do NOT collide (the extra-route adapter and axum
606    /// merge them), so the key is the pair, not the path alone.
607    ///
608    /// Must stay exhaustive: health probes, metrics, OpenAPI spec/docs, the OIDC
609    /// surface (injected backend or config-driven static discovery) and
610    /// embedder extra routes. All built-in surfaces here are `GET`.
611    fn endpoint_routes(&self) -> anyhow::Result<Vec<(String, String)>> {
612        let mut routes = Vec::new();
613        let mut get = |path: String| routes.push(("GET".to_string(), path));
614        if self.config.health.enabled {
615            get(self.config.health.path.clone());
616            get(self.config.health.live_path.clone());
617            get(self.config.health.ready_path.clone());
618            get(self.config.health.startup_path.clone());
619        }
620        if self.config.metrics.enabled {
621            get(self.config.metrics.path.clone());
622        }
623        if let Some(openapi) = self.config.openapi.as_ref().filter(|o| o.enabled) {
624            get(openapi.path.clone());
625            get(openapi.docs_path.clone());
626        }
627        // OIDC: an injected backend supersedes config-driven static discovery.
628        if let Some(backend) = &self.oidc_backend {
629            for doc in backend.metadata_documents() {
630                get(doc.path);
631            }
632            get(backend.jwks().path);
633            get(backend.userinfo_path());
634        } else if let Some(cfg) = &self.config.oidc_discovery {
635            if let Some(oidc) = oidc::Oidc::build(cfg)
636                .map_err(|e| anyhow::anyhow!("invalid oidc_discovery config: {e}"))?
637            {
638                for path in oidc.paths() {
639                    get(path);
640                }
641            }
642        }
643        // Only the extra routes mounted: one the router skips answers nothing.
644        for route in self
645            .extra_routes
646            .iter()
647            .filter(|route| embed::mountable(route))
648        {
649            routes.push((route.method.as_str().to_string(), route.path.clone()));
650        }
651        Ok(routes)
652    }
653
654    /// A lazy channel to the configured upstream address (`upstream.default`),
655    /// the upstream of the standalone proxy. It connects on first use, giving
656    /// up after five seconds.
657    ///
658    /// # Errors
659    ///
660    /// No upstream address is configured, or it is not a valid URI.
661    pub fn upstream(&self) -> anyhow::Result<tonic::transport::Channel> {
662        let Some(upstream) = &self.config.upstream else {
663            anyhow::bail!("no gRPC upstream address is configured (upstream.default)");
664        };
665        Ok(
666            tonic::transport::Channel::from_shared(upstream.default.clone())
667                .map_err(|e| anyhow::anyhow!("invalid gRPC upstream URL: {e}"))?
668                .connect_timeout(std::time::Duration::from_secs(5))
669                .connect_lazy(),
670        )
671    }
672
673    /// The proxy's HTTP routes in front of the configured upstream address
674    /// (see [`upstream`](Self::upstream)), as an axum `Router` to serve or to
675    /// merge into an axum application. It answers HTTP only; use
676    /// [`service`](Self::service) for native gRPC on the same listener, or for
677    /// an upstream in process.
678    ///
679    /// A URL whose path a transcoded route matches but which no binding
680    /// answers (its field template or custom verb does not fit) goes to the
681    /// proxy's other routes, extra routes included, and is answered `404`
682    /// when none of them answers it. An axum router routes a request to one
683    /// route, so routes merged after this one are not tried for it: serve the
684    /// application's own routes through
685    /// [`with_extra_routes`](Self::with_extra_routes), or use
686    /// [`service`](Self::service) with
687    /// [`with_fallback`](ProxyService::with_fallback).
688    ///
689    /// # Errors
690    ///
691    /// No valid upstream address, or a configuration [`service`](Self::service)
692    /// rejects.
693    pub fn router(&self) -> anyhow::Result<Router> {
694        let Built {
695            router, chooser, ..
696        } = self.routes(self.upstream()?)?;
697        Ok(match chooser {
698            Some(chooser) => chooser.layer(router),
699            None => router,
700        })
701    }
702
703    /// The whole proxy as one tower service in front of `upstream`: native
704    /// gRPC requests reach `upstream` unchanged, every other request the
705    /// proxy's routes, whose transcoded calls go to `upstream` too. See
706    /// [`ProxyService`].
707    ///
708    /// `upstream` is any gRPC service ([`upstream::Upstream`]): the embedder's
709    /// own tonic services in process, with no socket between them and the
710    /// proxy, or a remote [`Channel`](tonic::transport::Channel) such as
711    /// [`upstream`](Self::upstream).
712    ///
713    /// # Errors
714    ///
715    /// An invalid configuration (see [`ProxyConfig::validate`]), descriptors
716    /// that cannot be loaded, a route mounted twice, or a malformed auth,
717    /// authz, shield or OIDC section.
718    ///
719    /// # Examples
720    ///
721    /// ```
722    /// use structured_proxy::ProxyServer;
723    ///
724    /// # fn build() -> anyhow::Result<()> {
725    /// let grpc = tonic::service::Routes::default(); // add your services here
726    /// let service = ProxyServer::from_yaml_str("service:\n  name: demo\n")?.service(grpc)?;
727    /// # let _ = service;
728    /// # Ok(())
729    /// # }
730    /// # build().unwrap();
731    /// ```
732    pub fn service<U: Upstream>(&self, upstream: U) -> anyhow::Result<ProxyService<U>> {
733        let Built {
734            router,
735            chooser,
736            direct,
737            cors,
738            guards,
739        } = self.routes(upstream.clone())?;
740        // The routes answer a browser's preflight for gRPC-Web too, so its
741        // call carries the same policy unless the upstream sets its own.
742        let grpc_web_cors = self.config.cors.grpc_web.then_some(cors);
743        let routing = match direct {
744            Some((direct, others)) => service::Routing::Direct { direct, others },
745            None => service::Routing::routed(router, chooser),
746        };
747        Ok(ProxyService::new(
748            upstream,
749            routing,
750            grpc_web_cors,
751            guards,
752            self.config.grpc_web.translate,
753        ))
754    }
755
756    /// Build the axum router with all endpoints, calling `upstream`; the
757    /// choice among its transcoded bindings, made before its layers; the CORS
758    /// policy it answers under; and the guards, for the traffic outside it.
759    fn routes<U: Upstream>(&self, upstream: U) -> anyhow::Result<Built> {
760        // Enforce cross-field invariants on the embedded path too, where the
761        // config is built directly instead of through `from_yaml_str`.
762        self.config.validate()?;
763        let pool = self.load_descriptors()?;
764        let selection = transcode::RpcSelection::new(&pool, &self.config.transcode.only)
765            .map_err(|e| anyhow::anyhow!("invalid transcode.only: {e}"))?;
766
767        let service_name = self.config.service.name.clone();
768
769        // The verify path that is actually mounted (branch-correct), if any.
770        let verify_path = self.mounted_verify_path();
771
772        // Validate the WHOLE mounted edge BEFORE any router is built, so a
773        // malformed path (missing leading '/') or a collision (between built-in
774        // routes, the OIDC surface, embedder extra routes, transcoded paths, or
775        // the verify endpoint) is a clear error instead of an axum panic at
776        // `.route`/`.merge`. Collisions are keyed by (method, path): same-path
777        // routes with different methods are legal (they merge), so only a
778        // repeated (method, path) — or any overlap with the verify endpoint,
779        // which answers ALL methods (`*`) — is a real conflict.
780        let endpoints_mounted = self.endpoint_routes()?;
781        let mut mounted = endpoints_mounted.clone();
782        mounted.extend(transcode::route_paths(
783            &pool,
784            &self.config.aliases,
785            &selection,
786        ));
787        if let Some(vp) = &verify_path {
788            mounted.push(("*".to_string(), vp.clone()));
789        }
790        // Key by NORMALIZED shape, not raw text: axum/matchit treats two dynamic
791        // routes with the same structure but different param names (e.g.
792        // `/v1/x/{a}` and `/v1/x/{b}`) as a conflict, so they must collide here.
793        let mut methods_by_shape: std::collections::HashMap<
794            String,
795            std::collections::HashSet<&str>,
796        > = std::collections::HashMap::new();
797        for (method, path) in &mounted {
798            if !path.starts_with('/') {
799                anyhow::bail!("route path {path:?} must start with '/'");
800            }
801            let methods = methods_by_shape
802                .entry(normalize_route_shape(path))
803                .or_default();
804            // `*` (the verify endpoint) claims every method, so it conflicts with
805            // any other route on the same shape, and vice versa.
806            let conflict = if method == "*" {
807                !methods.is_empty()
808            } else {
809                methods.contains("*") || methods.contains(method.as_str())
810            };
811            if conflict {
812                anyhow::bail!("route path {path:?} is registered by more than one endpoint");
813            }
814            methods.insert(method.as_str());
815        }
816        // Paths the router cannot hold together are refused here, with the
817        // router axum routes with: a variable and a catch-all at one position,
818        // or one shape spelled with other variable names (routes of other
819        // methods merge into one path only under the same spelling).
820        let mut router = matchit::Router::new();
821        let mut paths = std::collections::HashSet::new();
822        for (_, path) in &mounted {
823            if paths.insert(path.as_str()) {
824                if let Err(error) = router.insert(path.as_str(), ()) {
825                    anyhow::bail!("route path {path:?} conflicts with another route: {error}");
826                }
827            }
828        }
829
830        // Keep the actually-configured probe / metrics / verify paths reachable
831        // under maintenance mode. The default exempt list names the default
832        // paths; once those are relocated via config, the relocated paths must
833        // be exempted too, or maintenance would 503 probe and forward-auth
834        // traffic that was intentionally exempt before.
835        let mut maintenance_exempt = self.config.maintenance.exempt_paths.clone();
836        if self.config.health.enabled {
837            maintenance_exempt.push(self.config.health.path.clone());
838            maintenance_exempt.push(self.config.health.live_path.clone());
839            maintenance_exempt.push(self.config.health.ready_path.clone());
840            maintenance_exempt.push(self.config.health.startup_path.clone());
841        }
842        if self.config.metrics.enabled {
843            maintenance_exempt.push(self.config.metrics.path.clone());
844        }
845        if let Some(vp) = &verify_path {
846            maintenance_exempt.push(vp.clone());
847        }
848
849        // A forwarded name gRPC metadata cannot carry would be rejected by a
850        // conforming upstream on every request; refuse it here instead.
851        if let Some(name) = self
852            .config
853            .forwarded_headers
854            .iter()
855            .find(|name| !transcode::metadata::is_grpc_key(name))
856        {
857            anyhow::bail!(
858                "forwarded_headers entry {name:?} is not a gRPC metadata key \
859                 (letters, digits, '_', '-' and '.')"
860            );
861        }
862
863        let resolver = Arc::new(
864            client_address::Resolver::build(&self.config.client_address)
865                .map_err(|e| anyhow::anyhow!("invalid client_address config: {e}"))?,
866        );
867        // The client-address headers reach a transcoded call as the
868        // forwarding policy wrote them on the request, whatever the list
869        // says; the rest of the list is forwarded as configured.
870        let forwarded_headers: Arc<[String]> = self
871            .config
872            .forwarded_headers
873            .iter()
874            .filter(|name| !resolver.reserves(name))
875            .cloned()
876            .chain(resolver.forwarded_headers().map(str::to_owned))
877            .collect();
878        // Parsed once rather than for every request; every entry is a gRPC
879        // metadata key, so a header name.
880        let forwarded_names = forwarded_headers
881            .iter()
882            .filter_map(|name| http::HeaderName::from_bytes(name.as_bytes()).ok())
883            .collect();
884        let state = ProxyState {
885            upstream,
886            forwarded_headers,
887            forwarded_names,
888            sse_keep_alive_secs: self.config.streaming.sse_keep_alive_secs,
889        };
890
891        let cors = self.build_cors()?;
892
893        // Build transcoding routes from descriptor pool.
894        let (transcode_routes, choices) = transcode::routes_and_choices(
895            &pool,
896            &self.config.aliases,
897            &self.transcode.clone().with_selection(selection.clone()),
898        );
899
900        // JWT auth, if configured (auth.mode == "jwt").
901        let auth = match &self.config.auth {
902            Some(cfg) => auth::Auth::build(cfg, self.token_verifier.clone())
903                .map_err(|e| anyhow::anyhow!("invalid auth config: {e}"))?,
904            None => None,
905        };
906        // Forward-auth verification endpoint, sharing the built Auth.
907        let forward_auth = auth.as_ref().and_then(|built| {
908            auth::forward::ForwardAuth::build(self.config.auth.as_ref()?, built.clone())
909        });
910        // The methods a route answers: `*` is not a method. It marks the
911        // verify endpoint and a `custom` `*` rule, which answer every method,
912        // and a transcoded path with verbs, whose bindings still name theirs.
913        let mut methods: Vec<http::Method> = mounted
914            .iter()
915            .filter(|(method, _)| method != "*")
916            .filter_map(|(method, _)| http::Method::from_bytes(method.as_bytes()).ok())
917            .collect();
918        let bound = transcode::bound_methods(&pool, &self.config.aliases, &selection);
919        methods.extend(bound.methods);
920        let mut routed = guard::Routed::new(&methods);
921        if bound.every {
922            routed = routed.every(guard::Class::Transcoded);
923        }
924        if verify_path.is_some() {
925            routed = routed.every(guard::Class::Verify);
926        }
927        let guards = Arc::new(self.guards(resolver, auth, maintenance_exempt, routed)?);
928
929        // Health routes. Paths are configurable; the whole group is skippable.
930        let health_routes = if self.config.health.enabled {
931            let health = &self.config.health;
932            let health_service_name = service_name.clone();
933            Router::new()
934                .route(
935                    &health.path,
936                    get({
937                        let name = health_service_name.clone();
938                        move || async move {
939                            Json(serde_json::json!({
940                                "status": "ok",
941                                "service": name,
942                            }))
943                        }
944                    }),
945                )
946                .route(&health.live_path, get(|| async { StatusCode::OK }))
947                .route(
948                    &health.ready_path,
949                    get(|State(state): State<ProxyState<U>>| async move {
950                        let mut client =
951                            tonic_health::pb::health_client::HealthClient::new(state.upstream);
952                        let check = client.check(tonic_health::pb::HealthCheckRequest {
953                            service: String::new(),
954                        });
955                        // An upstream that does not answer in time is not ready.
956                        match tokio::time::timeout(transcode::UPSTREAM_DEADLINE, check)
957                            .await
958                            .unwrap_or_else(|_| {
959                                Err(tonic::Status::deadline_exceeded("health check timed out"))
960                            }) {
961                            Ok(resp) => {
962                                let status = resp.into_inner().status;
963                                if status
964                                    == tonic_health::pb::health_check_response::ServingStatus::Serving
965                                        as i32
966                                {
967                                    StatusCode::OK
968                                } else {
969                                    StatusCode::SERVICE_UNAVAILABLE
970                                }
971                            }
972                            Err(_) => StatusCode::SERVICE_UNAVAILABLE,
973                        }
974                    }),
975                )
976                .route(&health.startup_path, get(|| async { StatusCode::OK }))
977        } else {
978            Router::new()
979        };
980
981        // Metrics route. Path is configurable; the endpoint is skippable.
982        let metrics_routes = if self.config.metrics.enabled {
983            Router::new().route(
984                &self.config.metrics.path,
985                get(|| async {
986                    let encoder = prometheus::TextEncoder::new();
987                    let metric_families = prometheus::default_registry().gather();
988                    match encoder.encode_to_string(&metric_families) {
989                        Ok(text) => (
990                            StatusCode::OK,
991                            [(
992                                axum::http::header::CONTENT_TYPE,
993                                "text/plain; version=0.0.4; charset=utf-8",
994                            )],
995                            text,
996                        )
997                            .into_response(),
998                        Err(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
999                    }
1000                }),
1001            )
1002        } else {
1003            Router::new()
1004        };
1005
1006        // OpenAPI + docs routes (if enabled).
1007        let openapi_routes = self.build_openapi_routes(&pool, &selection);
1008
1009        // OIDC routes (public, like the health endpoints). An injected
1010        // OidcBackend supersedes the config-driven static discovery: the proxy
1011        // hosts the HTTP surface, the embedder supplies the content.
1012        let oidc_routes = match &self.oidc_backend {
1013            Some(backend) => embed::oidc_backend_routes(backend.clone()),
1014            None => match &self.config.oidc_discovery {
1015                Some(cfg) => oidc::Oidc::build(cfg)
1016                    .map_err(|e| anyhow::anyhow!("invalid oidc_discovery config: {e}"))?
1017                    .map(|o| o.routes())
1018                    .unwrap_or_default(),
1019                None => Router::new(),
1020            },
1021        };
1022
1023        let endpoints = Router::new()
1024            .merge(health_routes)
1025            .merge(metrics_routes)
1026            .merge(openapi_routes)
1027            .merge(oidc_routes)
1028            .merge(embed::extra_routes_router(&self.extra_routes));
1029
1030        // Forward-auth `/verify` endpoint. An injected AuthDecider owns it when
1031        // present (in-process PDP); otherwise the config-driven JWT ForwardAuth
1032        // backs it. Its path was validated above.
1033        let verify = if let Some(decider) = &self.auth_decider {
1034            let decider = decider.clone();
1035            Router::new().route(
1036                &self.decider_verify_path(),
1037                axum::routing::any(move |req: axum::extract::Request| {
1038                    let decider = decider.clone();
1039                    async move { embed::verify_via_decider(decider, req).await }
1040                }),
1041            )
1042        } else if let Some(forward_auth) = &forward_auth {
1043            forward_auth.routes()
1044        } else {
1045            Router::new()
1046        };
1047
1048        // Each class of traffic behind the guards that cover it, so a request
1049        // runs only the guards of its own class. A path no route answers
1050        // reaches the plain 404 (or the fallback's own guards).
1051        let endpoints = guards.router(endpoints, guard::Class::Endpoints);
1052        let verify = guards.router(verify, guard::Class::Verify);
1053        // One stack, outermost first, applied at once: each layer applied on
1054        // its own would box every route again, and the router clones a
1055        // route's boxes for every request.
1056        let layered = |router: Router<ProxyState<U>>| {
1057            router.layer((
1058                // Every request, preflights included.
1059                TraceLayer::new_for_http(),
1060                // Wraps every enforcement layer so short-circuited responses
1061                // keep CORS headers, and answers preflight before auth.
1062                cors.clone(),
1063                // Before every guard: they and the handlers read its result.
1064                client_address::ClientAddressLayer::with(guards.client_address.clone()),
1065            ))
1066        };
1067        let router = layered(
1068            Router::new()
1069                .merge(guards.router(transcode_routes, guard::Class::Transcoded))
1070                .merge(endpoints.clone())
1071                .merge(verify.clone()),
1072        );
1073        // The binding is chosen before every layer: a URL no binding answers
1074        // (its field template, its custom verb) is no transcoded request. It
1075        // goes to the proxy's other routes, which a router without the
1076        // transcoded ones ranks for it, and on to the fallback (or the plain
1077        // 404) when none answers it either, past the layers like any path no
1078        // route answers.
1079        let elsewhere =
1080            layered(Router::new().merge(endpoints).merge(verify)).with_state(state.clone());
1081        // The transcoded paths are ranked among the paths of every other
1082        // route: a binding refused at a better-ranked path passes the request
1083        // on only to a transcoded path above them all.
1084        let mut others: Vec<(Option<http::Method>, String)> = endpoints_mounted
1085            .iter()
1086            .map(|(method, path)| {
1087                (
1088                    http::Method::from_bytes(method.as_bytes()).ok(),
1089                    path.clone(),
1090                )
1091            })
1092            .collect();
1093        if let Some(path) = &verify_path {
1094            others.push((None, path.clone()));
1095        }
1096        let choices = choices.map(|choices| choices.among(&others));
1097        // Served by the proxy service, the transcoded routes are matched by
1098        // the proxy itself and reached behind the same layers past any
1099        // router.
1100        let direct = choices.as_ref().and_then(|choices| {
1101            let layered = |service: guard::BoxedService| {
1102                guard::BoxedService::new(
1103                    tower::ServiceBuilder::new()
1104                        .map_response(|response: http::Response<_>| {
1105                            response.map(axum::body::Body::new)
1106                        })
1107                        .layer((
1108                            TraceLayer::new_for_http(),
1109                            cors.clone(),
1110                            client_address::ClientAddressLayer::with(guards.client_address.clone()),
1111                        ))
1112                        .service(guards.service(service, guard::Class::Transcoded)),
1113                )
1114            };
1115            let direct = choices.direct(state.clone(), layered)?;
1116            Some((direct, elsewhere.clone()))
1117        });
1118        let chooser = choices.map(|choices| choices.with_elsewhere(elsewhere.clone()));
1119        let router = router.with_state(state);
1120
1121        Ok(Built {
1122            router,
1123            chooser,
1124            direct,
1125            cors,
1126            guards,
1127        })
1128    }
1129
1130    /// The guards the configuration and the hooks turn on, each with its
1131    /// scope, after the client-address resolution `resolver`;
1132    /// `maintenance_exempt` lists the paths maintenance mode leaves
1133    /// reachable, `routed` the methods the routes answer.
1134    ///
1135    /// # Errors
1136    ///
1137    /// A malformed shield, authz or concurrency section, a JWT claim mapped
1138    /// onto a header the client-address forwarding writes, or a scope that
1139    /// covers no traffic, names an invalid path glob or an invalid method.
1140    fn guards(
1141        &self,
1142        resolver: Arc<client_address::Resolver>,
1143        auth: Option<Arc<auth::Auth>>,
1144        maintenance_exempt: Vec<String>,
1145        routed: guard::Routed<'_>,
1146    ) -> anyhow::Result<guard::Guards> {
1147        use config::Traffic::{Endpoints, Grpc, Transcoded};
1148        let scope = |config: Option<&ScopeConfig>, default: &[config::Traffic], what: &str| {
1149            guard::Scope::compile(config, default, what, routed).map_err(anyhow::Error::msg)
1150        };
1151        // A guard never sets what the client-address forwarding writes.
1152        let reserved: Arc<[http::HeaderName]> = resolver.configured_headers().cloned().collect();
1153        if let Some(header) = self
1154            .config
1155            .auth
1156            .as_ref()
1157            .and_then(|auth| auth.jwt.as_ref())
1158            .and_then(|jwt| jwt.claims_headers.values().find(|h| resolver.reserves(h)))
1159        {
1160            anyhow::bail!(
1161                "auth.jwt.claims_headers maps a claim onto {header:?}, which the \
1162                 client_address forwarding writes"
1163            );
1164        }
1165        let mut guards = guard::Guards {
1166            client_address: resolver,
1167            ..Default::default()
1168        };
1169        if guards.client_address.required() {
1170            guards.require_client = Some(scope(None, &[config::Traffic::All], "client_address")?);
1171        }
1172        // Mounted only while maintenance is on, so normal traffic pays nothing
1173        // for it.
1174        let maintenance = &self.config.maintenance;
1175        if maintenance.enabled {
1176            guards.maintenance = Some((
1177                Arc::new(guard::Maintenance {
1178                    exempt: maintenance_exempt,
1179                    message: maintenance.message.clone(),
1180                }),
1181                scope(
1182                    maintenance.scope.as_ref(),
1183                    &[Transcoded, Endpoints],
1184                    "maintenance",
1185                )?,
1186            ));
1187        }
1188        if let Some(cfg) = &self.config.concurrency {
1189            guards.concurrency = Some((
1190                guard::Concurrency::build(cfg).map_err(anyhow::Error::msg)?,
1191                // The proxy's own endpoints stay out by default: a health
1192                // probe refused under load gets a busy instance restarted.
1193                scope(cfg.scope.as_ref(), &[Transcoded, Grpc], "concurrency")?,
1194            ));
1195        }
1196        if let Some(cfg) = &self.config.shield {
1197            if let Some(shield) = shield::Shield::build(cfg)
1198                .map_err(|e| anyhow::anyhow!("invalid shield config: {e}"))?
1199            {
1200                guards.shield = Some((
1201                    shield,
1202                    scope(cfg.scope.as_ref(), &[Transcoded, Endpoints], "shield")?,
1203                ));
1204            }
1205        }
1206        let auth_config = self.config.auth.as_ref();
1207        if let Some(auth) = auth {
1208            guards.auth = Some((
1209                auth,
1210                scope(
1211                    auth_config.and_then(|a| a.scope.as_ref()),
1212                    &[Transcoded, Endpoints],
1213                    "auth",
1214                )?,
1215            ));
1216        }
1217        if let Some(cfg) = auth_config.and_then(|a| a.authz.as_ref()) {
1218            if let Some(authz) = auth::authz::Authz::build_reserving(cfg, reserved.clone())
1219                .map_err(|e| anyhow::anyhow!("invalid authz config: {e}"))?
1220            {
1221                guards.authz = Some((
1222                    authz,
1223                    scope(cfg.scope.as_ref(), &[Transcoded], "auth.authz")?,
1224                ));
1225            }
1226        }
1227        if let Some(decider) = &self.auth_decider {
1228            guards.decider = Some((
1229                Arc::new(embed::DeciderGate {
1230                    decider: decider.clone(),
1231                    reserved,
1232                }),
1233                scope(
1234                    self.auth_decider_scope.as_ref(),
1235                    &[Transcoded],
1236                    "auth_decider",
1237                )?,
1238            ));
1239        }
1240        Ok(guards)
1241    }
1242
1243    fn build_openapi_routes<S>(
1244        &self,
1245        pool: &DescriptorPool,
1246        selection: &transcode::RpcSelection,
1247    ) -> Router<S>
1248    where
1249        S: Clone + Send + Sync + 'static,
1250    {
1251        let openapi_config = match &self.config.openapi {
1252            Some(cfg) if cfg.enabled => cfg,
1253            _ => return Router::new(),
1254        };
1255
1256        let spec = openapi::generate(pool, openapi_config, &self.config.aliases, selection);
1257        let spec_json = serde_json::to_string_pretty(&spec).unwrap_or_default();
1258        let openapi_path = openapi_config.path.clone();
1259        let docs_path = openapi_config.docs_path.clone();
1260        let title = openapi_config
1261            .title
1262            .clone()
1263            .unwrap_or_else(|| self.config.service.name.clone());
1264        let openapi_path_for_docs = openapi_path.clone();
1265
1266        tracing::info!("OpenAPI spec at {}, docs at {}", openapi_path, docs_path,);
1267
1268        Router::new()
1269            .route(
1270                &openapi_path,
1271                get(move || async move {
1272                    (
1273                        StatusCode::OK,
1274                        [(
1275                            axum::http::header::CONTENT_TYPE,
1276                            "application/json; charset=utf-8",
1277                        )],
1278                        spec_json,
1279                    )
1280                }),
1281            )
1282            .route(
1283                &docs_path,
1284                get(move || async move {
1285                    let html = openapi::docs_html(&openapi_path_for_docs, &title);
1286                    (
1287                        StatusCode::OK,
1288                        [(axum::http::header::CONTENT_TYPE, "text/html; charset=utf-8")],
1289                        html,
1290                    )
1291                }),
1292            )
1293    }
1294
1295    /// The CORS policy of `cors:`.
1296    ///
1297    /// # Errors
1298    ///
1299    /// An origin that is not a header value, or an `expose_headers` entry that
1300    /// is not a header name: dropping either would quietly narrow the policy.
1301    fn build_cors(&self) -> anyhow::Result<CorsLayer> {
1302        let config = &self.config.cors;
1303        // Checked in both modes, so a typo fails the same way whether or not
1304        // origins are set.
1305        let configured = config
1306            .expose_headers
1307            .iter()
1308            .map(|name| {
1309                http::HeaderName::from_bytes(name.as_bytes()).map_err(|_| {
1310                    anyhow::anyhow!("cors.expose_headers entry {name:?} is not a header name")
1311                })
1312            })
1313            .collect::<anyhow::Result<Vec<_>>>()?;
1314        if config.origins.is_empty() {
1315            tracing::warn!("CORS origins not set — using permissive CORS (dev mode)");
1316            // Exposes every header already.
1317            Ok(CorsLayer::any(config.max_age_secs))
1318        } else {
1319            let origins = config
1320                .origins
1321                .iter()
1322                .map(|origin| {
1323                    http::HeaderValue::from_str(origin)
1324                        .map_err(|_| anyhow::anyhow!("cors.origins entry {origin:?} is not valid"))
1325                })
1326                .collect::<anyhow::Result<Vec<_>>>()?;
1327            let exposed = [
1328                // What a gRPC-Web client reads its status and details from.
1329                http::HeaderName::from_static("grpc-status"),
1330                http::HeaderName::from_static("grpc-message"),
1331                http::HeaderName::from_static("grpc-status-details-bin"),
1332                // Let browser clients read the rate-limit budget and back off.
1333                http::HeaderName::from_static("ratelimit-limit"),
1334                http::HeaderName::from_static("ratelimit-remaining"),
1335                http::HeaderName::from_static("ratelimit-reset"),
1336                http::HeaderName::from_static("retry-after"),
1337            ]
1338            .into_iter()
1339            .chain(configured);
1340            // With credentials the Fetch standard (§3.2.5) forbids `*` for
1341            // methods and headers, so the preflight's own request is echoed
1342            // back instead: what the browser asked for, from an allowed origin.
1343            Ok(CorsLayer::listed(
1344                origins,
1345                &exposed.collect::<Vec<_>>(),
1346                config.max_age_secs,
1347            ))
1348        }
1349    }
1350
1351    /// The [`ServeOptions`] of `listen:`: TLS from `listen.tls` (mTLS with its
1352    /// `client_ca_file`), the `listen.max_connections` cap and the connection
1353    /// timeouts, for [`serve_with`] on a listener of your own.
1354    ///
1355    /// # Errors
1356    ///
1357    /// A `max_connections`, `header_read_timeout_secs` or
1358    /// `tls.handshake_timeout_secs` of zero, TLS files that cannot be loaded,
1359    /// or no rustls crypto provider for TLS.
1360    pub fn serve_options(&self) -> anyhow::Result<ServeOptions> {
1361        use std::time::Duration;
1362        let listen = &self.config.listen;
1363        anyhow::ensure!(
1364            listen.header_read_timeout_secs > 0,
1365            "listen.header_read_timeout_secs must be at least 1"
1366        );
1367        let mut options = ServeOptions::new()
1368            .idle_timeout(
1369                (listen.idle_timeout_secs > 0)
1370                    .then(|| Duration::from_secs(listen.idle_timeout_secs)),
1371            )
1372            .header_read_timeout(Duration::from_secs(listen.header_read_timeout_secs))
1373            .drain_timeout(
1374                (listen.drain_timeout_secs > 0)
1375                    .then(|| Duration::from_secs(listen.drain_timeout_secs)),
1376            );
1377        if let Some(max) = listen.max_connections {
1378            anyhow::ensure!(max > 0, "listen.max_connections must be at least 1");
1379            options = options.max_connections(max);
1380        }
1381        if let Some(tls) = &listen.tls {
1382            anyhow::ensure!(
1383                tls.handshake_timeout_secs > 0,
1384                "listen.tls.handshake_timeout_secs must be at least 1"
1385            );
1386            options = options
1387                .tls(
1388                    tls::server_config(tls)
1389                        .map_err(|e| anyhow::anyhow!("invalid listen.tls: {e}"))?,
1390                )
1391                .tls_handshake_timeout(Duration::from_secs(tls.handshake_timeout_secs));
1392        }
1393        Ok(options)
1394    }
1395
1396    /// Serve the proxy on the configured listen address in front of the
1397    /// configured upstream address: REST and native gRPC on one port, with
1398    /// the TLS and connection limit of `listen:` (see
1399    /// [`serve_options`](Self::serve_options) and [`serve_with`]).
1400    ///
1401    /// # Errors
1402    ///
1403    /// What [`upstream`](Self::upstream), [`service`](Self::service) and
1404    /// [`serve_options`](Self::serve_options) reject, an invalid listen
1405    /// address, or a listener that fails.
1406    pub async fn serve(&self) -> anyhow::Result<()> {
1407        self.serve_with_shutdown(std::future::pending()).await
1408    }
1409
1410    /// [`serve`](Self::serve) until `signal` completes, then shut down with
1411    /// the drain of `listen.drain_timeout_secs` (see [`serve_with_shutdown`]).
1412    ///
1413    /// # Errors
1414    ///
1415    /// What [`serve`](Self::serve) returns.
1416    ///
1417    /// # Examples
1418    ///
1419    /// ```no_run
1420    /// # async fn run(server: structured_proxy::ProxyServer) -> anyhow::Result<()> {
1421    /// server
1422    ///     .serve_with_shutdown(async {
1423    ///         tokio::signal::ctrl_c().await.ok();
1424    ///     })
1425    ///     .await
1426    /// # }
1427    /// ```
1428    pub async fn serve_with_shutdown(
1429        &self,
1430        signal: impl std::future::Future<Output = ()>,
1431    ) -> anyhow::Result<()> {
1432        let service = self.service(self.upstream()?)?;
1433        let options = self.serve_options()?;
1434        let addr: SocketAddr = self.config.listen.http.parse()?;
1435        let listener = tokio::net::TcpListener::bind(addr).await?;
1436
1437        tracing::info!(
1438            tls = self.config.listen.tls.is_some(),
1439            "{} listening on {}",
1440            self.config.service.name,
1441            addr
1442        );
1443        serve_with_shutdown(listener, service, options, signal).await?;
1444        tracing::info!("{} stopped", self.config.service.name);
1445        Ok(())
1446    }
1447}
1448
1449/// Canonical shape of an axum route path for collision detection: every dynamic
1450/// segment (`{name}` capture or `{*name}` wildcard) is replaced by a
1451/// name-independent placeholder, so structurally identical routes that differ
1452/// only in parameter name (which axum/matchit rejects as a conflict) map to the
1453/// same key. Literal segments are unchanged.
1454fn normalize_route_shape(path: &str) -> String {
1455    path.split('/')
1456        .map(|seg| {
1457            if seg.starts_with("{*") && seg.ends_with('}') {
1458                "{*}"
1459            } else if seg.starts_with('{') && seg.ends_with('}') {
1460                "{}"
1461            } else {
1462                seg
1463            }
1464        })
1465        .collect::<Vec<_>>()
1466        .join("/")
1467}
1468
1469/// A [`ProxyState`] for tests whose routers never call the upstream: a lazy
1470/// channel to a port nothing listens on.
1471#[cfg(test)]
1472pub(crate) fn test_state() -> ProxyState<tonic::transport::Channel> {
1473    ProxyState {
1474        upstream: tonic::transport::Channel::from_static("http://127.0.0.1:1")
1475            .connect_timeout(std::time::Duration::from_millis(100))
1476            .connect_lazy(),
1477        forwarded_headers: Arc::from([]),
1478        forwarded_names: Arc::from([]),
1479        sse_keep_alive_secs: 15,
1480    }
1481}
1482
1483#[cfg(test)]
1484mod tests;