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