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;