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;