postrust_server/build.rs
1//! Building the application a host serves.
2//!
3//! Everything between a loaded configuration and a router answering requests:
4//! the pool, the schema cache, the REST surface, GraphQL and the layers around
5//! them. The `postrust` binary serves the result on a socket; the Lambda
6//! adapter hands the same router to the Lambda runtime. One path, so a request
7//! is answered the same way wherever it arrives -- the Lambda adapter used to
8//! be a second copy of the request path, and that copy bound no parameters and
9//! switched to no role.
10
11use crate::app::{self, handle_request};
12use crate::custom;
13use crate::state::AppState;
14use anyhow::Result;
15#[cfg(feature = "graphql")]
16use axum::routing::{get, post};
17use axum::{
18 extract::{DefaultBodyLimit, Request},
19 http::{HeaderValue, Method},
20 response::{IntoResponse, Json},
21 routing::{any, Route},
22 Router,
23};
24use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
25use std::convert::Infallible;
26use std::sync::Arc;
27use tower::{Layer, Service};
28use tower_http::cors::{AllowOrigin, Any as CorsAny, CorsLayer};
29use tracing::{error, info};
30
31/// What the host running the application can do beyond answering requests.
32///
33/// A server process lives as long as it likes and owns its sockets. A Lambda
34/// sandbox is frozen between invocations, so a task listening for `NOTIFY` is
35/// asleep exactly when the notification arrives, and API Gateway does not
36/// upgrade a connection to a WebSocket at all.
37#[derive(Clone, Copy, Debug)]
38#[non_exhaustive]
39pub struct Host {
40 /// Long-lived background work: the `NOTIFY` schema reloader
41 /// (`db_channel_enabled`) and the subscription broker's listener.
42 pub background_tasks: bool,
43 /// Upgrading a connection to a WebSocket, which GraphQL subscriptions are
44 /// served over.
45 pub websockets: bool,
46}
47
48impl Host {
49 /// A long-running server process.
50 pub const SERVER: Host = Host {
51 background_tasks: true,
52 websockets: true,
53 };
54
55 /// An AWS Lambda function: no background work, no WebSockets.
56 pub const LAMBDA: Host = Host {
57 background_tasks: false,
58 websockets: false,
59 };
60}
61
62/// The application, ready to serve.
63#[non_exhaustive]
64pub struct App {
65 /// Every route the configuration enables, with its layers and state.
66 pub router: Router,
67 /// The state the routes share, for anything served beside them -- the
68 /// binary's admin port answers health checks from it.
69 pub state: Arc<AppState>,
70}
71
72/// Connect, load the schema cache and build the router.
73///
74/// Fails rather than serving something partial when the database cannot be
75/// reached or a setting cannot be honoured. GraphQL is the exception: a schema
76/// that cannot be built is logged and the rest is served without it.
77///
78/// The application with nothing of the caller's added. [`Builder`] builds the
79/// same with routes of the caller's own beside it.
80pub async fn build(config: postrust_core::AppConfig, host: Host) -> Result<App> {
81 Builder::new(config, host).build().await
82}
83
84/// Builds the application with routes of the caller's own beside the API.
85///
86/// For a binary of your own that depends on this crate rather than a fork of
87/// it: webhooks, an endpoint that does several things in one transaction,
88/// anything that is not a table or a function.
89///
90/// ```no_run
91/// # async fn run() -> anyhow::Result<()> {
92/// use axum::{routing::post, Router};
93/// use postrust_server::{AppState, Builder, Host};
94/// use std::sync::Arc;
95///
96/// let hooks: Router<Arc<AppState>> = Router::new().route("/stripe", post(|| async { "ok" }));
97/// let config = postrust_core::AppConfig::try_from_env()?;
98/// let app = Builder::new(config, Host::SERVER)
99/// .nest("/hooks", hooks)
100/// .build()
101/// .await?;
102/// # let _ = app;
103/// # Ok(())
104/// # }
105/// ```
106///
107/// A nested router is served with everything the API is served with: the
108/// CORS policy, the body limit, the `OPTIONS` handling and the shared
109/// [`AppState`]. Its routes are explicit routes, so in compatibility mode they
110/// are matched before the REST surface served at the root -- a table that
111/// shares a prefix's name is shadowed there, and stays reachable under `/api`.
112///
113/// [`Builder::layer`] is for what has to see every request rather than only
114/// the caller's own: an authentication gate, a rate limit, a header every
115/// response carries.
116#[must_use = "a Builder does nothing until `build` is called"]
117pub struct Builder {
118 config: postrust_core::AppConfig,
119 host: Host,
120 nests: Vec<(String, Router<Arc<AppState>>)>,
121 layers: Vec<Wrap>,
122}
123
124/// A layer given to [`Builder::layer`], held until there is a router for it.
125///
126/// Boxed as the step that applies it rather than as the layer, since each
127/// layer is its own type and only `Router::layer` knows what it needs of one.
128type Wrap = Box<dyn FnOnce(Router) -> Router + Send>;
129
130impl Builder {
131 /// The application `config` asks for, on `host`, with nothing added yet.
132 pub fn new(config: postrust_core::AppConfig, host: Host) -> Self {
133 Builder {
134 config,
135 host,
136 nests: Vec::new(),
137 layers: Vec::new(),
138 }
139 }
140
141 /// Serve `router` under `prefix`. May be called more than once.
142 ///
143 /// The prefix is taken as a path: the leading slash is added if missing
144 /// and a trailing one dropped, so `"hooks"`, `"/hooks"` and `"/hooks/"`
145 /// are one prefix. Nothing is checked until [`Builder::build`], which
146 /// refuses a prefix that is the root, that overlaps a path the server
147 /// serves itself ([`RESERVED_PREFIXES`], and `/_` itself), or that
148 /// overlaps another prefix given here. In compatibility mode, where REST
149 /// is served at `/` as well, a prefix whose first segment is `rpc` or the
150 /// name of a table or view in an exposed schema takes those paths from
151 /// REST, and a warning says so at start-up. A prefix beneath `/_`, such as
152 /// `/_/hooks`, is served beside the built-in `/_/health` and `/_/ready`.
153 pub fn nest(mut self, prefix: impl Into<String>, router: Router<Arc<AppState>>) -> Self {
154 self.nests.push((prefix.into(), router));
155 self
156 }
157
158 /// Wrap the whole application in `layer`. May be called more than once.
159 ///
160 /// Outermost, around every route the server serves and around its own
161 /// layers too, so it sees each request first and each response last --
162 /// including a preflight the CORS policy answers itself, and a refusal
163 /// that never reaches a handler. Of several, the one given last is the
164 /// outermost, as with `Router::layer`.
165 ///
166 /// For something only the caller's own routes need, put the layer on the
167 /// router given to [`Builder::nest`] instead.
168 pub fn layer<L>(mut self, layer: L) -> Self
169 where
170 L: Layer<Route> + Clone + Send + Sync + 'static,
171 L::Service: Service<Request> + Clone + Send + Sync + 'static,
172 <L::Service as Service<Request>>::Response: IntoResponse + 'static,
173 <L::Service as Service<Request>>::Error: Into<Infallible> + 'static,
174 <L::Service as Service<Request>>::Future: Send + 'static,
175 {
176 self.layers
177 .push(Box::new(move |router: Router| router.layer(layer)));
178 self
179 }
180
181 /// Connect, load the schema cache and build the router.
182 ///
183 /// The prefixes given to [`Builder::nest`] are checked first, before
184 /// anything is connected to, so a clash is reported as an error rather
185 /// than as a panic from the router -- and without a database to find it.
186 /// Otherwise as [`build`].
187 pub async fn build(self) -> Result<App> {
188 let nests = mount_points(self.nests)?;
189 // The JWT keys too, for the same reason: a key set that does not
190 // parse, or a JWKS URL that would not be fetched, is a configuration
191 // error, and is better reported before the database is connected to
192 // than after. Made once, here, and handed to the state.
193 let jwt = jwt_config(&self.config)?;
194 let mut app = assemble(self.config, self.host, nests, jwt).await?;
195 for wrap in self.layers {
196 app.router = wrap(app.router);
197 }
198 Ok(app)
199 }
200}
201
202/// The paths the server serves itself, which a prefix given to
203/// [`Builder::nest`] may neither equal nor sit above or beneath.
204///
205/// Reserved whether or not the feature that serves them is compiled in, so
206/// that a binary which builds today does not start failing when GraphQL or the
207/// admin UI is turned on. `/_` is not in the list: the built-in routes under
208/// it are, and the rest of it is the caller's.
209pub const RESERVED_PREFIXES: &[&str] = &[
210 "/api",
211 "/healthz",
212 "/v1/version",
213 "/v1/graphql",
214 "/v1alpha1/graphql",
215 "/admin",
216 "/_/health",
217 "/_/ready",
218];
219
220/// Where the built-in custom router is mounted.
221const CUSTOM_PREFIX: &str = "/_";
222
223/// The caller's routers, each under its prefix once the prefix is checked.
224type Nests = Vec<(String, Router<Arc<AppState>>)>;
225
226/// Check the prefixes given to [`Builder::nest`], normalizing each.
227fn mount_points(nests: Nests) -> Result<Nests> {
228 let mut checked: Nests = Vec::with_capacity(nests.len());
229 for (given, router) in nests {
230 let taken: Vec<&str> = checked.iter().map(|(prefix, _)| prefix.as_str()).collect();
231 let prefix = check_prefix(&given, &taken)?;
232 checked.push((prefix, router));
233 }
234 Ok(checked)
235}
236
237/// `given` as a prefix to nest at -- one leading slash, no trailing one -- or
238/// why it cannot be nested at beside the server's own paths and the prefixes
239/// already `taken`.
240fn check_prefix(given: &str, taken: &[&str]) -> Result<String> {
241 let trimmed = given.trim().trim_matches('/');
242 if trimmed.is_empty() {
243 // The root is the API's in compatibility mode and the server's own
244 // directory otherwise; a router merged there would take one or the
245 // other.
246 anyhow::bail!("cannot nest routes at {given:?}: a prefix must name a path below /");
247 }
248 // An empty segment matches nothing a client sends, and a parameter or a
249 // wildcard in a nest path is something the router refuses by panicking.
250 // Both are refused here instead, where it can be said why.
251 if trimmed
252 .split('/')
253 .any(|segment| segment.is_empty() || segment.contains(['{', '}', '*']))
254 {
255 anyhow::bail!(
256 "cannot nest routes at {given:?}: a prefix is literal path segments, \
257 without parameters, wildcards or empty segments"
258 );
259 }
260 let prefix = format!("/{trimmed}");
261
262 // Overlap rather than equality. `/v1` is not `/v1/version`, but nesting
263 // at `/v1` claims everything beneath it, and which of the two then
264 // answers `/v1/version` would be settled by the router's matching
265 // priority rather than by anything the caller wrote. A nest at exactly
266 // the same path as another is one the router refuses by panicking.
267 if prefix == CUSTOM_PREFIX {
268 anyhow::bail!(
269 "cannot nest routes at {given:?}: the server serves /_ itself; nest beneath it, \
270 at /_/<name>, instead"
271 );
272 }
273 if let Some(reserved) = RESERVED_PREFIXES
274 .iter()
275 .find(|reserved| overlaps(&prefix, reserved))
276 {
277 anyhow::bail!(
278 "cannot nest routes at {given:?}: it overlaps {reserved}, which the server serves itself"
279 );
280 }
281 if let Some(other) = taken.iter().find(|other| overlaps(&prefix, other)) {
282 anyhow::bail!("cannot nest routes at {given:?}: it overlaps {other}, nested earlier");
283 }
284 Ok(prefix)
285}
286
287/// Whether `path` is `base` or beneath it, segment by segment: `/apis` is not
288/// beneath `/api`.
289/// In compatibility mode REST is served at `/` too, beneath every path no
290/// route claims, so a prefix naming a table, a view or `/rpc` takes those
291/// requests from REST. That is allowed -- the route was asked for, and the
292/// table is still served under `/api` -- but said, so it is not found out
293/// from a response. Checked against the schema cache at start-up.
294fn warn_of_shadowing(nests: &Nests, schema_cache: &postrust_core::SchemaCache, schemas: &[String]) {
295 for (prefix, _) in nests {
296 let first = prefix
297 .trim_start_matches('/')
298 .split('/')
299 .next()
300 .unwrap_or_default();
301 if first == "rpc" {
302 tracing::warn!(
303 "compatibility mode: routes nested at {prefix} answer in place of the \
304 database functions REST serves at /rpc; they are still served at /api/rpc"
305 );
306 }
307 if let Some(table) = schema_cache
308 .tables
309 .keys()
310 .find(|qi| qi.name == first && schemas.contains(&qi.schema))
311 {
312 tracing::warn!(
313 "compatibility mode: routes nested at {prefix} answer in place of {}.{first}, \
314 which REST would serve at /{first}; it is still served at /api/{first}",
315 table.schema
316 );
317 }
318 }
319}
320
321fn beneath(path: &str, base: &str) -> bool {
322 path.strip_prefix(base)
323 .is_some_and(|rest| rest.is_empty() || rest.starts_with('/'))
324}
325
326/// Whether nesting at both `a` and `b` would claim a path twice.
327fn overlaps(a: &str, b: &str) -> bool {
328 beneath(a, b) || beneath(b, a)
329}
330
331/// The keys and claims tokens are verified with: `PGRST_JWT_SECRET`, and the
332/// key set at `PGRST_JWT_JWKS_URL` when there is one (not yet fetched).
333fn jwt_config(config: &postrust_core::AppConfig) -> Result<postrust_auth::JwtConfig> {
334 let jwt = postrust_engine::jwt_config(config)
335 .map_err(|e| anyhow::anyhow!("PGRST_JWT_SECRET: {e}"))?;
336 // Its errors name `PGRST_JWT_JWKS_URL` themselves.
337 crate::jwks::attach(jwt, config).map_err(|e| anyhow::anyhow!("{e}"))
338}
339
340/// Everything [`Builder::build`] does once the caller's prefixes and the JWT
341/// configuration are checked.
342async fn assemble(
343 config: postrust_core::AppConfig,
344 host: Host,
345 nests: Nests,
346 jwt: postrust_auth::JwtConfig,
347) -> Result<App> {
348 info!("Database: {}", mask_db_uri(&config.db_uri));
349
350 // Create database pool
351 let connect_options: PgConnectOptions = config
352 .db_uri
353 .parse()
354 .map_err(|e| anyhow::anyhow!("invalid database URI: {e}"))?;
355
356 // Prepared statements are sqlx's default and stay on unless asked. Turning
357 // them off is for a connection pooler in transaction mode -- PgBouncer and
358 // the like -- where the server connection a statement was prepared on is
359 // not the one the next query lands on, and every reuse fails with
360 // "prepared statement does not exist".
361 let connect_options = if config.db_prepared_statements {
362 connect_options
363 } else {
364 info!("Prepared statements disabled");
365 connect_options.statement_cache_capacity(0)
366 };
367
368 let pool = PgPoolOptions::new()
369 .max_connections(config.db_pool_size)
370 .acquire_timeout(std::time::Duration::from_secs(config.db_pool_timeout))
371 .connect_with(connect_options)
372 .await?;
373
374 info!("Connected to database");
375
376 // Load schema cache
377 let schema_cache = postrust_core::SchemaCache::load_with_search_path(
378 &pool,
379 &config.db_schemas,
380 &config.db_extra_search_path,
381 )
382 .await?;
383 info!("{}", schema_cache.summary());
384 if config.compat_mode {
385 warn_of_shadowing(&nests, &schema_cache, &config.db_schemas);
386 }
387
388 // Create app state
389 let state = Arc::new(AppState::with_jwt(pool, schema_cache, config.clone(), jwt));
390
391 // The pool and the token cache, for the metrics read when collected.
392 #[cfg(feature = "otel")]
393 crate::otel::observe(&state);
394
395 // `db_channel_enabled`: reload the schema cache when PostgreSQL says the
396 // schema changed, instead of on a restart. A migration ends with
397 // `NOTIFY pgrst, 'reload schema'` and the API picks the new shape up.
398 //
399 // The REST cache is swapped here; each reload then bumps `reloaded`, and
400 // the GraphQL schema, built from its own snapshot of the cache, is
401 // rebuilt from the new one when it sees the bump.
402 // Keys from `PGRST_JWT_JWKS_URL`: fetched once before serving, then kept
403 // fresh in the background where the host runs background work. A fetch
404 // that fails does not stop the start; see `crate::jwks`.
405 crate::jwks::start(&state, host).await;
406
407 let (reloaded, schema_reloads) = tokio::sync::watch::channel(0u64);
408 if config.db_channel_enabled && host.background_tasks {
409 spawn_schema_reloader(state.clone(), config.db_channel.clone(), reloaded);
410 }
411 #[cfg(not(feature = "graphql"))]
412 drop(schema_reloads);
413
414 // Build REST API router (under /api prefix)
415 let api_router: Router<Arc<AppState>> = Router::new()
416 .route("/", any(handle_request))
417 .route("/{*path}", any(handle_request));
418
419 // Build main router
420 let mut app: Router<Arc<AppState>> = Router::new().nest("/api", api_router);
421
422 // Two endpoints every Hasura deployment has, and that the things around a
423 // deployment reach for without being asked to: `/healthz` is what a load
424 // balancer, a Kubernetes probe and `docker-compose` healthchecks are
425 // already configured to poll, and `/v1/version` is what a client library
426 // calls to decide which features to use. Serving them costs nothing and
427 // not serving them makes a drop-in replacement fail its first health
428 // check.
429 app = app
430 .route("/healthz", axum::routing::get(|| async { "OK" }))
431 .route(
432 "/v1/version",
433 axum::routing::get(|| async {
434 Json(serde_json::json!({ "version": env!("CARGO_PKG_VERSION") }))
435 }),
436 );
437
438 // Add custom routes (health checks, webhooks, etc.)
439 //
440 // The caller's own routes go in here too, before any layer is applied, so
441 // they are wrapped in the same CORS policy, body limit and `OPTIONS`
442 // handling as the API. One beneath `/_`, such as `/_/hooks`, sits beside
443 // the nest at `/_`: the more specific path is matched first.
444 app = app.nest(CUSTOM_PREFIX, custom::custom_router());
445 info!("Custom routes enabled at /_");
446 for (prefix, router) in nests {
447 info!("Custom routes enabled at {}", prefix);
448 // In compatibility mode a database error is answered in the
449 // database's own words, by REST and so by a route's
450 // `postrust_server::Error` too. The error cannot see the
451 // configuration when `?` makes it, so this renders it again on the
452 // way out; see `crate::error`.
453 let router = match config.compat_mode {
454 true => router.layer(axum::middleware::map_response_with_state(
455 state.clone(),
456 crate::error::render_verbatim,
457 )),
458 false => router,
459 };
460 app = app.nest(&prefix, router);
461 }
462
463 // Add admin routes and GraphQL endpoint if feature is enabled
464 #[cfg(feature = "graphql")]
465 {
466 use async_graphql_axum::GraphQLBatchRequest as GqlRequest;
467 use axum::extract::State as AxumState;
468 use axum::http::HeaderMap;
469 use postrust_graphql::handler::GraphQLState;
470 use postrust_graphql::schema::SchemaConfig;
471
472 #[cfg(feature = "admin-ui")]
473 {
474 info!("Admin UI enabled at /admin");
475 app = app.nest("/admin", crate::admin::admin_router());
476 }
477
478 // Create GraphQL state with subscriptions enabled
479 let schema_cache_snapshot = state.schema_cache.read().await.clone();
480 let schema_cache_arc = Arc::new(schema_cache_snapshot);
481 // What Hasura keeps in metadata and a schema cannot carry: the names,
482 // and what each role may do. Absent, every name is derived exactly as
483 // before and there is no permission layer at all.
484 //
485 // Two spellings because the document outgrew the first one. `_NAMES`
486 // is what it was called when names were all it held, and a deployment
487 // already setting it should not have to change anything.
488 let names_var = ["PGRST_GRAPHQL_METADATA", "PGRST_GRAPHQL_NAMES"]
489 .into_iter()
490 .find_map(|name| std::env::var(name).ok().map(|value| (name, value)));
491
492 let graphql_names = match names_var {
493 Some((var, value)) => match postrust_graphql::names::NameOverrides::parse(&value) {
494 Ok(names) => {
495 if !names.is_empty() {
496 info!("GraphQL names given for {} tables", names.len());
497 }
498 if names.placed_functions() > 0 {
499 info!(
500 "GraphQL roots given for {} functions",
501 names.placed_functions()
502 );
503 }
504 if names.has_permissions() {
505 info!(
506 "GraphQL permissions given for {} roles, {} grants",
507 names.roles().len(),
508 names.granted()
509 );
510 }
511 names
512 }
513 Err(e) => {
514 // Serving the derived names instead would answer every
515 // request under a name the client does not send, which
516 // reads as a broken server rather than a bad setting. A
517 // permission read wrong is worse still: it would serve
518 // rows a rule was written to withhold.
519 tracing::error!("{}: {}", var, e);
520 return Err(anyhow::anyhow!("{}: {}", var, e));
521 }
522 },
523 None => postrust_graphql::names::NameOverrides::default(),
524 };
525
526 // An enum table's members are rows, not schema, so they are read here
527 // rather than reflected, and again each time the schema is rebuilt:
528 // a migration that adds a value notifies like one that adds a column.
529 let graphql_enum_values = read_enum_values(
530 &state.pool,
531 &*state.schema_cache.read().await,
532 &graphql_names,
533 )
534 .await;
535
536 // How often a live query re-reads itself with nothing having
537 // notified it. The notifications do the work; this is the floor
538 // under what a trigger cannot see -- a view, an embedded row on a
539 // table with no trigger, a predicate written against the clock.
540 let subscription_refresh_seconds = std::env::var("PGRST_SUBSCRIPTION_REFRESH")
541 .ok()
542 .and_then(|value| value.trim().parse::<u64>().ok())
543 .unwrap_or(30);
544
545 let mut graphql_config = SchemaConfig::default();
546 // A subscription is a WebSocket, so a host that cannot hold one
547 // does not advertise a `Subscription` root it cannot serve.
548 graphql_config.enable_subscriptions = host.websockets;
549 graphql_config.subscription_refresh_seconds = subscription_refresh_seconds;
550 graphql_config.max_rows = config.db_max_rows;
551 graphql_config.enable_federation = config.graphql_federation;
552 graphql_config.type_prefix = config.graphql_type_prefix.clone();
553 // The GraphQL schema was built for `public` whatever the server
554 // was told to expose, so a table in any other schema of
555 // `PGRST_DB_SCHEMAS` was reachable over REST and invisible over
556 // GraphQL.
557 graphql_config.exposed_schemas = config.db_schemas.clone();
558 graphql_config.enum_values = graphql_enum_values;
559 graphql_config.names = graphql_names;
560 // A schema that cannot be built is a reason to serve without GraphQL,
561 // not a reason to exit. The REST surface, the admin UI and the health
562 // endpoint do not depend on it, and taking the process down with it
563 // turns one unrepresentable table into a server that will not start.
564 // The endpoint is mounted all the same, answering that there is no
565 // schema, so the reload after a migration that fixes the table brings
566 // GraphQL up without a restart.
567 let graphql_state = match GraphQLState::new(
568 state.pool.clone(),
569 schema_cache_arc.clone(),
570 graphql_config.clone(),
571 ) {
572 Ok(graphql_state) => {
573 // Initialize subscription broker
574 if !host.websockets {
575 info!("GraphQL subscriptions disabled: this host cannot hold a WebSocket");
576 } else if let Err(e) = graphql_state.init_subscriptions().await {
577 tracing::warn!("Failed to initialize subscription broker: {}. Subscriptions may not work until triggers are created.", e);
578 } else {
579 info!("GraphQL subscriptions enabled");
580 }
581 info!("GraphQL endpoint enabled at /api/graphql");
582 Some(Arc::new(graphql_state))
583 }
584 Err(e) => {
585 tracing::error!(
586 "GraphQL schema could not be built, serving without it: {}",
587 e
588 );
589 None
590 }
591 };
592 {
593 // The schema in use, replaced whole when the database's changes.
594 // A request takes the current one once and keeps it, so an
595 // operation is never answered half from one schema and half from
596 // the next.
597 let current = Arc::new(std::sync::RwLock::new(graphql_state));
598 spawn_graphql_rebuilder(
599 state.clone(),
600 current.clone(),
601 graphql_config,
602 schema_reloads,
603 );
604
605 // Combined state for GraphQL routes (includes JWT config for auth)
606 #[derive(Clone)]
607 struct GraphQLAppState {
608 gql_state: Arc<std::sync::RwLock<Option<Arc<GraphQLState>>>>,
609 /// REST's state, for the JWT configuration and the cache of
610 /// verified tokens: one token is verified once, whichever
611 /// surface it is presented to.
612 rest: Arc<AppState>,
613 hasura_auth: postrust_auth::HasuraAuthConfig,
614 /// REST's configuration, shared by every operation's context:
615 /// a GraphQL transaction is set up from it as a REST one is.
616 /// One copy for the server's life, since the configuration is
617 /// read once at start-up and never changes under it.
618 config: Arc<postrust_core::AppConfig>,
619 }
620
621 impl GraphQLAppState {
622 /// The GraphQL state to answer this operation from, if there
623 /// is one yet.
624 fn gql(&self) -> Option<Arc<GraphQLState>> {
625 self.gql_state
626 .read()
627 .unwrap_or_else(|e| e.into_inner())
628 .clone()
629 }
630 }
631
632 let graphql_app_state = GraphQLAppState {
633 gql_state: current,
634 rest: state.clone(),
635 hasura_auth: postrust_auth::HasuraAuthConfig {
636 admin_secret: config.hasura_admin_secret.clone(),
637 unauthorized_role: config.hasura_unauthorized_role.clone(),
638 },
639 config: Arc::new(state.config.clone()),
640 };
641
642 // Wrapper handler that creates context from request with proper auth
643 /// A body that is a JSON *array* is a batch: several operations
644 /// in one request, answered with an array of responses in the
645 /// order they were sent. Each is its own operation with its own
646 /// transaction -- batching is a way to save round trips, not a
647 /// way to make several mutations atomic, which is what naming
648 /// several root fields in one mutation is for.
649 async fn handle_graphql(
650 AxumState(app_state): AxumState<GraphQLAppState>,
651 // Which of the three addresses this arrived at. The routes are
652 // nested, so the request's own path is `/` by the time it gets
653 // here and the original is the only thing that still says.
654 axum::extract::OriginalUri(uri): axum::extract::OriginalUri,
655 headers: HeaderMap,
656 req: GqlRequest,
657 ) -> (axum::http::StatusCode, Json<serde_json::Value>) {
658 // The status a refused write is answered with belongs to the
659 // endpoint rather than to the refusal: `/v1alpha1/graphql`
660 // reports it in the transport and `/v1/graphql` does not.
661 let endpoint = match uri.path().starts_with("/v1alpha1/") {
662 true => postrust_graphql::hasura::Endpoint::Legacy,
663 false => postrust_graphql::hasura::Endpoint::Current,
664 };
665 match req.0 {
666 async_graphql::BatchRequest::Single(request) => {
667 let body = one_graphql(app_state, headers, request).await;
668 let status = axum::http::StatusCode::from_u16(
669 postrust_graphql::hasura::status_for(&body, endpoint),
670 )
671 .unwrap_or(axum::http::StatusCode::OK);
672 (status, Json(body))
673 }
674 async_graphql::BatchRequest::Batch(requests) => {
675 let mut answers = Vec::with_capacity(requests.len());
676 for request in requests {
677 answers.push(
678 one_graphql(app_state.clone(), headers.clone(), request).await,
679 );
680 }
681 // A batch is answered 200 whatever its parts say: the
682 // transport carried every one of them, and which of
683 // them failed is in the array.
684 (
685 axum::http::StatusCode::OK,
686 Json(serde_json::Value::Array(answers)),
687 )
688 }
689 }
690 }
691
692 /// Who is asking, settled once: per request over HTTP, and per
693 /// connection over a WebSocket, where the token arrives in the
694 /// `connection_init` payload rather than on every operation.
695 #[derive(Clone)]
696 struct GraphQLIdentity {
697 auth_result: postrust_auth::AuthResult,
698 hasura_role: Option<String>,
699 session: std::collections::HashMap<String, String>,
700 elevated: bool,
701 headers: HeaderMap,
702 }
703
704 /// An operation ready to run against the schema its role answers
705 /// from, and the transaction its writes will share.
706 struct PreparedOperation {
707 schema: async_graphql::dynamic::Schema,
708 request: async_graphql::Request,
709 write: postrust_graphql::context::SharedWrite,
710 }
711
712 /// What [`postrust_auth::ensure_keys`] is to REST: keys from a
713 /// JWKS URL fetched, if the token needs them, before the
714 /// synchronous [`graphql_identity`] verifies it.
715 async fn ensure_keys(app_state: &GraphQLAppState, headers: &HeaderMap) {
716 let auth_header = headers.get("authorization").and_then(|v| v.to_str().ok());
717 postrust_auth::ensure_keys(auth_header, &app_state.rest.jwt_config).await;
718 }
719
720 /// The caller's identity, or the Hasura-shaped body refusing it.
721 fn graphql_identity(
722 app_state: &GraphQLAppState,
723 headers: HeaderMap,
724 ) -> Result<GraphQLIdentity, serde_json::Value> {
725 // Extract auth header and authenticate
726 let auth_header = headers.get("authorization").and_then(|v| v.to_str().ok());
727
728 // Through the token cache where one is configured, as REST
729 // authenticates.
730 let jwt_config = &app_state.rest.jwt_config;
731 let token = match &app_state.rest.jwt_cache {
732 Some(cache) => cache.authenticate(auth_header, jwt_config),
733 None => postrust_auth::authenticate(auth_header, jwt_config),
734 };
735 // A token that was offered and verified. Not the same as
736 // `token.is_ok()`, which is also true of the anonymous
737 // fallback -- and an anonymous request is precisely the one
738 // that has not been authenticated.
739 let token_verified = auth_header.is_some() && token.is_ok();
740 // A token that was offered and is refused: REST answers each of
741 // these 401. Without a secret there is nothing to verify
742 // against, and the header is ignored as Hasura ignores it when
743 // JWT mode is off -- a router forwarding `Authorization` to
744 // every subgraph relies on that. With no header at all, and no
745 // anonymous role, the request goes on as it always has.
746 let token_refusal = match (&token, auth_header) {
747 (Ok(_) | Err(postrust_auth::JwtError::SecretMissing), _) | (Err(_), None) => {
748 None
749 }
750 // Verified, and naming no role, with no anonymous role to
751 // fall back to. Served as anyone, it would be served as a
752 // role nobody chose.
753 (Err(postrust_auth::JwtError::NoIdentity), Some(_)) => Some(
754 postrust_graphql::hasura::missing_role_claim(&jwt_config.role_claim_key),
755 ),
756 (Err(e), Some(_)) => {
757 Some(postrust_graphql::hasura::invalid_jwt(&e.to_string()))
758 }
759 };
760
761 let auth_result = match token {
762 Ok(auth) => auth,
763 Err(e) => {
764 tracing::debug!("GraphQL auth failed: {}, using anon role", e);
765 postrust_auth::AuthResult {
766 role: jwt_config
767 .anon_role
768 .clone()
769 .unwrap_or_else(|| "anon".to_string()),
770 claims: std::collections::HashMap::new(),
771 }
772 }
773 };
774
775 tracing::debug!(
776 "GraphQL request authenticated as role: {}",
777 auth_result.role
778 );
779
780 // Who the caller is in Hasura's sense, which is a different
781 // question from which database user the transaction runs as.
782 // The secret goes first: a caller that tried to be an
783 // administrator and got the secret wrong is refused rather
784 // than asked for a token.
785 let pairs: Vec<(&str, &str)> = headers
786 .iter()
787 .filter_map(|(name, value)| {
788 value.to_str().ok().map(|value| (name.as_str(), value))
789 })
790 .collect();
791
792 let (hasura_role, session, elevated) = match (
793 postrust_auth::hasura::from_admin_secret(&app_state.hasura_auth, &pairs[..]),
794 token_refusal,
795 ) {
796 (postrust_auth::SecretOutcome::Accepted(identity), _) => {
797 (Some(identity.role), identity.session, identity.elevated)
798 }
799 (postrust_auth::SecretOutcome::Rejected, _) => {
800 return Err(postrust_graphql::hasura::denied(
801 postrust_auth::hasura::SECRET_INVALID,
802 ));
803 }
804 // The secret settled nothing, so the token is refused
805 // -- and an administrator's secret beside it wins, as
806 // it does in Hasura.
807 (_, Some(refusal)) => return Err(refusal),
808 // Nothing settled by the secret. A verified token speaks
809 // next, and session variables come from its claims.
810 //
811 // One header is read here and only here: `X-Hasura-Role`,
812 // against the `x-hasura-allowed-roles` the token carries.
813 // That list is inside the signature, so it lets a caller
814 // choose among the identities it was issued and not add
815 // one. Every other `x-hasura-*` header is still ignored --
816 // on a server with no secret to gate them they would let
817 // any caller name its own identity.
818 _ if token_verified => {
819 match postrust_auth::hasura::role_for_token(&auth_result.claims, &pairs[..])
820 {
821 postrust_auth::hasura::TokenRole::Is(role) => (
822 role,
823 postrust_auth::hasura::session_from_claims(&auth_result.claims),
824 false,
825 ),
826 postrust_auth::hasura::TokenRole::NotAllowed => {
827 return Err(postrust_graphql::hasura::denied(
828 postrust_auth::hasura::ROLE_NOT_ALLOWED,
829 ));
830 }
831 }
832 }
833 // A secret is configured and nothing authenticated this
834 // request. Answering it as a stranger is a choice the
835 // deployment makes; refusing is the default.
836 _ if app_state.hasura_auth.is_configured() => {
837 match postrust_auth::hasura::unauthenticated(&app_state.hasura_auth) {
838 Some(identity) => (Some(identity.role), identity.session, false),
839 None => {
840 return Err(postrust_graphql::hasura::denied(
841 postrust_auth::hasura::SECRET_MISSING,
842 ));
843 }
844 }
845 }
846 // No Hasura auth configured at all: unchanged from before
847 // any of this existed.
848 _ => (
849 postrust_auth::hasura::role_from_claims(&auth_result.claims),
850 postrust_auth::hasura::session_from_claims(&auth_result.claims),
851 false,
852 ),
853 };
854
855 Ok(GraphQLIdentity {
856 auth_result,
857 hasura_role,
858 session,
859 elevated,
860 headers,
861 })
862 }
863
864 /// Everything between knowing who is asking and running what they
865 /// asked: the checks Hasura makes before reading the document, the
866 /// schema the role answers from, and the context every resolver
867 /// reads the identity from.
868 fn prepare_operation(
869 app_state: &GraphQLAppState,
870 identity: &GraphQLIdentity,
871 req: async_graphql::Request,
872 ) -> Result<PreparedOperation, serde_json::Value> {
873 let GraphQLIdentity {
874 auth_result,
875 hasura_role,
876 session,
877 elevated,
878 headers,
879 } = identity.clone();
880 let Some(gql) = app_state.gql() else {
881 return Err(postrust_graphql::hasura::unavailable(
882 "the GraphQL schema could not be built; see the server log",
883 ));
884 };
885 let pairs: Vec<(&str, &str)> = headers
886 .iter()
887 .filter_map(|(name, value)| {
888 value.to_str().ok().map(|value| (name.as_str(), value))
889 })
890 .collect();
891
892 // A protocol neither server implements, which only one of
893 // them says. See `hasura::persisted_query`.
894 if postrust_graphql::hasura::persisted_query(&req) {
895 return Err(postrust_graphql::hasura::not_supported(
896 "PersistedQueryNotSupported",
897 ));
898 }
899
900 // A header Hasura reads itself, before it reads the
901 // document. `x-hasura-use-backend-only-permissions: random` is
902 // not false, and treating it as false would send a
903 // backend-only write down the path meant for everyone else on
904 // a header the client thought it had set.
905 if let Some(message) = postrust_auth::hasura::unreadable_boolean_header(&pairs[..])
906 {
907 return Err(postrust_graphql::hasura::malformed(&message));
908 }
909
910 // Which schema answers. A permission is a statement about what
911 // exists, so the role decides the shape of the API before it
912 // decides anything about a row: a role the document does not
913 // name has no schema at all, and is refused here rather than
914 // being answered from someone else's.
915 let Some(schema) = gql.schema_for(
916 hasura_role.as_deref(),
917 postrust_auth::hasura::backend_only_requested(&pairs[..], elevated),
918 ) else {
919 return Err(postrust_graphql::hasura::denied(&format!(
920 "role \"{}\" is not defined in the permissions",
921 hasura_role.as_deref().unwrap_or_default()
922 )));
923 };
924
925 // Create SchemaCacheRef from the static Arc<SchemaCache>
926 //
927 // The unreduced cache, deliberately. The schema above decides
928 // what may be asked for; this is what a resolver looks a type
929 // up in once something has been asked, and a superset can only
930 // answer the same questions.
931 let schema_cache_ref = postrust_core::schema_cache::SchemaCacheRef::from_static(
932 (*gql.schema_cache).clone(),
933 );
934
935 // Every write in this operation goes into one transaction, and
936 // this is the half that decides its fate: a mutation naming
937 // several root fields is all-or-nothing, so the second one
938 // failing has to take the first one's rows with it. Nothing to
939 // settle for a query, which never opens it.
940 let write: postrust_graphql::context::SharedWrite = Arc::default();
941
942 let gql_ctx = postrust_graphql::context::GraphQLContext::new(
943 gql.pool.clone(),
944 schema_cache_ref,
945 auth_result,
946 )
947 .with_session(session)
948 .with_identity(hasura_role.clone(), elevated)
949 .with_write(Arc::clone(&write))
950 .with_config(Arc::clone(&app_state.config));
951
952 // A role may be told it cannot read the schema as data. Not a
953 // permission on a table, so it is not settled by which schema
954 // answers -- the schema is the thing being withheld.
955 //
956 // Not withheld from a caller holding the admin secret, whatever
957 // role it then names. That is measured rather than assumed: a
958 // v2.50.1 reference given `set_graphql_schema_introspection_
959 // options` answers `__schema` for the named role when the
960 // secret is sent beside the role header, and answers it from
961 // that role's own restricted schema. Reading the schema is an
962 // administrator's to do; the permissions still apply to it.
963 let introspection_disabled_for = hasura_role
964 .as_deref()
965 .filter(|_| !elevated)
966 .filter(|role| gql.config.names.introspection_disabled(role));
967 let prepared = postrust_graphql::hasura::prepare(
968 Some(schema),
969 req,
970 introspection_disabled_for,
971 );
972 let request = match prepared {
973 Ok(request) => request,
974 // Refused before it ran: nothing was written, so there is
975 // nothing to settle.
976 Err((_, errors)) => {
977 let mut response = async_graphql::Response::new(async_graphql::Value::Null);
978 response.errors = errors;
979 return Err(postrust_graphql::hasura::envelope(response));
980 }
981 };
982 let request = request
983 .data(gql_ctx)
984 .data(gql.pool.clone())
985 .data(Arc::clone(&gql.broker));
986 Ok(PreparedOperation {
987 schema: schema.clone(),
988 request,
989 write,
990 })
991 }
992
993 /// Settle an operation's writes before its answer goes out.
994 async fn settle_write(
995 write: &postrust_graphql::context::SharedWrite,
996 response: &mut async_graphql::Response,
997 ) {
998 if let Some(tx) = write.lock().await.take() {
999 let settled = if response.errors.is_empty() {
1000 tx.commit().await
1001 } else {
1002 tx.rollback().await
1003 };
1004 // A commit that fails is a mutation that did not happen,
1005 // however well every statement in it went. Saying so is
1006 // the whole point of running them together.
1007 if let Err(e) = settled {
1008 tracing::error!("GraphQL mutation could not be settled: {}", e);
1009 response
1010 .errors
1011 .push(async_graphql::ServerError::new(e.to_string(), None));
1012 response.data = async_graphql::Value::Null;
1013 }
1014 }
1015 }
1016
1017 async fn one_graphql(
1018 app_state: GraphQLAppState,
1019 headers: HeaderMap,
1020 req: async_graphql::Request,
1021 ) -> serde_json::Value {
1022 ensure_keys(&app_state, &headers).await;
1023 let identity = match graphql_identity(&app_state, headers) {
1024 Ok(identity) => identity,
1025 Err(refusal) => return refusal,
1026 };
1027 let PreparedOperation {
1028 schema,
1029 request,
1030 write,
1031 } = match prepare_operation(&app_state, &identity, req) {
1032 Ok(operation) => operation,
1033 Err(refusal) => return refusal,
1034 };
1035 // One span for the operation, beneath the request's, named by
1036 // the operation as the GraphQL conventions have it. A span
1037 // and not async-graphql's OpenTelemetry extension: that is
1038 // built against opentelemetry 0.27, and this against 0.33.
1039 use tracing::Instrument as _;
1040 let span = tracing::debug_span!(
1041 "graphql",
1042 "graphql.operation.name" = request.operation_name.as_deref(),
1043 );
1044 let mut response = schema.execute(request).instrument(span).await;
1045 settle_write(&write, &mut response).await;
1046 postrust_graphql::hasura::envelope(response)
1047 }
1048
1049 /// A Hasura-shaped body as the response a WebSocket carries, so a
1050 /// subscription is refused, and its errors read, exactly as the
1051 /// same operation over HTTP would be.
1052 fn hasura_response(body: serde_json::Value) -> async_graphql::Response {
1053 let data = body
1054 .get("data")
1055 .cloned()
1056 .and_then(|data| async_graphql::Value::from_json(data).ok())
1057 .unwrap_or(async_graphql::Value::Null);
1058 let mut response = async_graphql::Response::new(data);
1059 for error in body
1060 .get("errors")
1061 .and_then(serde_json::Value::as_array)
1062 .into_iter()
1063 .flatten()
1064 {
1065 let message = error
1066 .get("message")
1067 .and_then(serde_json::Value::as_str)
1068 .unwrap_or_default();
1069 let mut server_error = async_graphql::ServerError::new(message, None);
1070 if let Some(fields) = error
1071 .get("extensions")
1072 .and_then(serde_json::Value::as_object)
1073 {
1074 let mut extensions = async_graphql::ErrorExtensionValues::default();
1075 for (name, value) in fields {
1076 if let Ok(value) = async_graphql::Value::from_json(value.clone()) {
1077 extensions.set(name, value);
1078 }
1079 }
1080 server_error.extensions = Some(extensions);
1081 }
1082 response.errors.push(server_error);
1083 }
1084 response
1085 }
1086
1087 /// The headers a WebSocket connection is authenticated with: the
1088 /// upgrade request's, overlaid with those the `connection_init`
1089 /// payload carries. A browser cannot set headers on a WebSocket,
1090 /// so a client puts `Authorization` in the payload -- Hasura's own
1091 /// clients under `headers`, others at the top level.
1092 fn connection_headers(
1093 mut headers: HeaderMap,
1094 payload: &serde_json::Value,
1095 ) -> HeaderMap {
1096 let nested = payload
1097 .get("headers")
1098 .and_then(serde_json::Value::as_object);
1099 for fields in [payload.as_object(), nested].into_iter().flatten() {
1100 for (name, value) in fields {
1101 let Some(value) = value.as_str() else {
1102 continue;
1103 };
1104 if let (Ok(name), Ok(value)) = (
1105 axum::http::HeaderName::from_bytes(name.as_bytes()),
1106 axum::http::HeaderValue::from_str(value),
1107 ) {
1108 headers.insert(name, value);
1109 }
1110 }
1111 }
1112 headers
1113 }
1114
1115 /// Runs every WebSocket operation -- query, mutation or
1116 /// subscription -- as the identity its connection was initialised
1117 /// with, through the same preparation an HTTP request gets.
1118 #[derive(Clone)]
1119 struct WebSocketExecutor(GraphQLAppState);
1120
1121 impl async_graphql::Executor for WebSocketExecutor {
1122 async fn execute(
1123 &self,
1124 request: async_graphql::Request,
1125 ) -> async_graphql::Response {
1126 use async_graphql::futures_util::StreamExt;
1127 // Not reached over a WebSocket, which runs everything as a
1128 // stream; answered from the stream so the two cannot differ.
1129 self.execute_stream(request, None)
1130 .next()
1131 .await
1132 .unwrap_or_else(|| async_graphql::Response::new(async_graphql::Value::Null))
1133 }
1134
1135 fn execute_stream(
1136 &self,
1137 request: async_graphql::Request,
1138 session_data: Option<Arc<async_graphql::Data>>,
1139 ) -> async_graphql::futures_util::stream::BoxStream<'static, async_graphql::Response>
1140 {
1141 use async_graphql::futures_util::{stream, StreamExt};
1142 let identity = session_data
1143 .as_deref()
1144 .and_then(|data| data.get(&std::any::TypeId::of::<GraphQLIdentity>()))
1145 .and_then(|identity| identity.downcast_ref::<GraphQLIdentity>())
1146 .cloned();
1147 // `connection_init` refuses a connection it cannot
1148 // identify, so an operation without an identity is one the
1149 // protocol let through before initialising. Refused rather
1150 // than run as nobody in particular.
1151 let Some(identity) = identity else {
1152 let refusal =
1153 postrust_graphql::hasura::denied("the connection was not initialised");
1154 return stream::once(async move { hasura_response(refusal) }).boxed();
1155 };
1156 match prepare_operation(&self.0, &identity, request) {
1157 Err(refusal) => {
1158 stream::once(async move { hasura_response(refusal) }).boxed()
1159 }
1160 Ok(PreparedOperation {
1161 schema,
1162 request,
1163 write,
1164 }) => {
1165 async_graphql::Executor::execute_stream(&schema, request, session_data)
1166 .then(move |mut response| {
1167 let write = Arc::clone(&write);
1168 async move {
1169 settle_write(&write, &mut response).await;
1170 hasura_response(postrust_graphql::hasura::envelope(
1171 response,
1172 ))
1173 }
1174 })
1175 .boxed()
1176 }
1177 }
1178 }
1179 }
1180
1181 /// GraphQL over a WebSocket, authenticated as HTTP is. The
1182 /// identity is settled once, from `connection_init`, and a
1183 /// connection that would have been refused over HTTP -- a wrong
1184 /// admin secret, a role the token does not allow -- is refused
1185 /// there rather than served as the anonymous role.
1186 async fn graphql_ws(
1187 AxumState(app_state): AxumState<GraphQLAppState>,
1188 protocol: async_graphql_axum::GraphQLProtocol,
1189 headers: HeaderMap,
1190 ws: axum::extract::ws::WebSocketUpgrade,
1191 ) -> axum::response::Response {
1192 ws.protocols(async_graphql::http::ALL_WEBSOCKET_PROTOCOLS)
1193 .on_upgrade(move |socket| async move {
1194 let init_state = app_state.clone();
1195 async_graphql_axum::GraphQLWebSocket::new(
1196 socket,
1197 WebSocketExecutor(app_state),
1198 protocol,
1199 )
1200 .on_connection_init(move |payload| async move {
1201 let headers = connection_headers(headers, &payload);
1202 ensure_keys(&init_state, &headers).await;
1203 match graphql_identity(&init_state, headers) {
1204 Ok(identity) => {
1205 let mut data = async_graphql::Data::default();
1206 data.insert(identity);
1207 Ok(data)
1208 }
1209 Err(refusal) => Err(async_graphql::Error::new(
1210 refusal
1211 .pointer("/errors/0/message")
1212 .and_then(serde_json::Value::as_str)
1213 .unwrap_or("connection refused"),
1214 )),
1215 }
1216 })
1217 .serve()
1218 .await
1219 })
1220 }
1221
1222 // Add GraphQL routes with WebSocket support for subscriptions
1223 let graphql_router = Router::new()
1224 .route("/", post(handle_graphql))
1225 .route("/", get(postrust_graphql::handler::graphql_playground))
1226 .with_state(graphql_app_state.clone());
1227
1228 // The same state as HTTP, because a WebSocket is authenticated the
1229 // same way: `postrust_graphql::handler::graphql_ws_handler` carries
1230 // no caller identity, so every subscription it served ran without
1231 // a role, a session or claims.
1232 let ws_router = Router::new()
1233 .route("/ws", get(graphql_ws))
1234 .with_state(graphql_app_state);
1235
1236 let graphql_app = match host.websockets {
1237 true => graphql_router.merge(ws_router),
1238 false => graphql_router,
1239 };
1240 // `/v1/graphql` is where a Hasura client sends its queries, and it is
1241 // the only address most of them can be told about: the endpoint is
1242 // baked into generated clients and codegen configs. `/api/graphql`
1243 // keeps working for anything already pointed at it, and
1244 // `/v1alpha1/graphql` is the address Hasura served before `/v1`
1245 // and still answers on -- a client old enough to have been
1246 // pointed there is exactly the one that cannot be repointed.
1247 app = app
1248 .nest("/v1/graphql", graphql_app.clone())
1249 .nest("/v1alpha1/graphql", graphql_app.clone())
1250 .nest("/api/graphql", graphql_app);
1251 }
1252 }
1253
1254 // PostgREST compatibility mode: also serve the REST surface at the root so
1255 // canonical PostgREST paths (`/rpc/<name>`, `/<table>`) work in addition to
1256 // the `/api`-prefixed paths. Explicit routes (`/`, `/_`, `/admin`, `/api`,
1257 // and whatever was given to `Builder::nest`) still take precedence; only
1258 // otherwise-unmatched paths hit this fallback.
1259 if config.compat_mode {
1260 app = app.fallback(handle_request);
1261 info!("PostgREST compatibility mode enabled: REST surface also served at /");
1262
1263 // Key ordering is fixed when the binary is compiled, so asking for
1264 // compatibility at runtime cannot turn it on. Say so, rather than
1265 // leaving someone to discover the difference by diffing responses.
1266 if !cfg!(feature = "compat-key-order") {
1267 tracing::warn!(
1268 "compatibility mode: object keys will be returned in alphabetical order, \
1269 not in select order as PostgREST returns them. Build with \
1270 --features compat-key-order to match, at a cost of up to 15% throughput \
1271 on wide rows."
1272 );
1273 }
1274 }
1275
1276 // Add root info endpoint.
1277 //
1278 // Not in compatibility mode: there `/` is part of the API surface -- it is
1279 // where PostgREST serves the schema description, and where it reports an
1280 // `Accept-Profile` naming a schema that is not exposed. A directory of
1281 // this server's own endpoints in its place answers a different question
1282 // from the one asked.
1283 if !config.compat_mode {
1284 app = app.route(
1285 "/",
1286 axum::routing::get(|| async {
1287 Json(serde_json::json!({
1288 "name": "postrust",
1289 "version": env!("CARGO_PKG_VERSION"),
1290 "api": "/api",
1291 "custom": "/_",
1292 "health": "/_/health",
1293 "admin": "/admin",
1294 "docs": "/admin/swagger"
1295 }))
1296 }),
1297 );
1298 }
1299
1300 // An empty list is "any origin", which is the default and what
1301 // `PGRST_SERVER_CORS_ORIGINS="*"` asks for. Anything that is not a valid
1302 // header value is dropped with a warning rather than quietly widening the
1303 // policy back to `*` -- a typo in an origin should cost that origin, not
1304 // the restriction.
1305 let allow_origin = if config.server_cors_origins.is_empty() {
1306 AllowOrigin::any()
1307 } else {
1308 let origins: Vec<HeaderValue> = config
1309 .server_cors_origins
1310 .iter()
1311 .filter_map(|o| match HeaderValue::from_str(o) {
1312 Ok(v) => Some(v),
1313 Err(_) => {
1314 tracing::warn!("Ignoring CORS origin {:?}: not a valid header value", o);
1315 None
1316 }
1317 })
1318 .collect();
1319 info!("CORS restricted to {} origin(s)", origins.len());
1320 AllowOrigin::list(origins)
1321 };
1322
1323 // Apply CORS and state
1324 let app = app
1325 .layer(
1326 CorsLayer::new()
1327 .allow_origin(allow_origin)
1328 .allow_methods([
1329 Method::GET,
1330 Method::POST,
1331 Method::PUT,
1332 Method::PATCH,
1333 Method::DELETE,
1334 Method::OPTIONS,
1335 Method::HEAD,
1336 ])
1337 .allow_headers(CorsAny)
1338 .expose_headers(CorsAny),
1339 )
1340 // For the routes whose handlers take a body *extractor* -- GraphQL,
1341 // the admin surface -- which is what this layer reaches. The REST
1342 // handler takes the whole `Request` and reads the body itself, so it
1343 // applies `max_body_size` directly; see `postrust_engine::handle`.
1344 .layer(DefaultBodyLimit::max(config.max_body_size))
1345 // Outermost, so it runs before the CORS layer -- which answers every
1346 // OPTIONS itself and never calls what it wraps, so nothing downstream
1347 // of it can say what a resource allows.
1348 .layer(axum::middleware::from_fn_with_state(
1349 state.clone(),
1350 app::options_allow,
1351 ))
1352 // Outermost of all, so the request span covers everything above --
1353 // an OPTIONS answered by `options_allow`, a preflight answered by
1354 // CORS -- and every response, whoever made it, carries the request
1355 // id. A `Router::layer`, so the router has matched the route before
1356 // this runs and `MatchedPath` is there to name the span's
1357 // `http.route`. See `crate::telemetry`.
1358 .layer(crate::telemetry::RequestTraceLayer)
1359 .with_state(state.clone());
1360
1361 Ok(App { router: app, state })
1362}
1363
1364/// Reload the REST schema cache on `NOTIFY <channel>`.
1365///
1366/// Reconnects on its own: a listening connection is dropped by a restart of
1367/// the database or anything between, and a reloader that gives up on the first
1368/// error is one that stops working silently partway through a deployment.
1369fn spawn_schema_reloader(
1370 state: Arc<AppState>,
1371 channel: String,
1372 reloaded: tokio::sync::watch::Sender<u64>,
1373) {
1374 const RETRY: std::time::Duration = std::time::Duration::from_secs(5);
1375
1376 tokio::spawn(async move {
1377 loop {
1378 let mut listener = match sqlx::postgres::PgListener::connect_with(&state.pool).await {
1379 Ok(listener) => listener,
1380 Err(e) => {
1381 error!("Schema reloader could not connect: {}; retrying", e);
1382 tokio::time::sleep(RETRY).await;
1383 continue;
1384 }
1385 };
1386
1387 if let Err(e) = listener.listen(&channel).await {
1388 error!("Schema reloader could not LISTEN on {}: {}", channel, e);
1389 tokio::time::sleep(RETRY).await;
1390 continue;
1391 }
1392 info!("Listening on channel {} for schema reloads", channel);
1393
1394 // Inner loop until the connection fails, then reconnect.
1395 loop {
1396 match listener.recv().await {
1397 Ok(notification) => {
1398 info!("Schema reload requested: {:?}", notification.payload());
1399 match state.reload_schema().await {
1400 Ok(()) => {
1401 info!(
1402 "Schema cache reloaded: {}",
1403 state.schema_cache().await.summary()
1404 );
1405 reloaded.send_modify(|n| *n += 1);
1406 #[cfg(feature = "otel")]
1407 crate::otel::schema_reloaded(true);
1408 }
1409 // The old cache is kept. Serving the previous
1410 // schema is better than serving none, and the next
1411 // notification tries again.
1412 Err(e) => {
1413 error!("Schema cache reload failed: {}", e);
1414 #[cfg(feature = "otel")]
1415 crate::otel::schema_reloaded(false);
1416 }
1417 }
1418 }
1419 Err(e) => {
1420 error!("Schema reloader lost its connection: {}; reconnecting", e);
1421 break;
1422 }
1423 }
1424 }
1425 }
1426 });
1427}
1428
1429/// Rebuild the GraphQL schema each time the REST schema cache is reloaded.
1430///
1431/// The new state is built beside the old one and swapped in only once it is
1432/// whole: a schema that cannot be built leaves the previous one serving, as a
1433/// failed reload of the REST cache does. Enum tables are read again, since a
1434/// migration may add a value; the permission document is kept from start-up,
1435/// since it is not part of what a migration notifies about. The subscription
1436/// broker carries over to the new state, so a subscription opened on the old
1437/// schema keeps receiving notifications, and listens on any table the new
1438/// one adds.
1439#[cfg(feature = "graphql")]
1440fn spawn_graphql_rebuilder(
1441 state: Arc<AppState>,
1442 current: Arc<std::sync::RwLock<Option<Arc<postrust_graphql::handler::GraphQLState>>>>,
1443 mut config: postrust_graphql::schema::SchemaConfig,
1444 mut reloads: tokio::sync::watch::Receiver<u64>,
1445) {
1446 use postrust_graphql::handler::GraphQLState;
1447
1448 tokio::spawn(async move {
1449 while reloads.changed().await.is_ok() {
1450 let cache = Arc::new(state.schema_cache().await.clone());
1451 config.enum_values = read_enum_values(&state.pool, &cache, &config.names).await;
1452 let mut next = match GraphQLState::new(state.pool.clone(), cache, config.clone()) {
1453 Ok(next) => next,
1454 Err(e) => {
1455 error!(
1456 "GraphQL schema rebuild failed, keeping the previous one: {}",
1457 e
1458 );
1459 continue;
1460 }
1461 };
1462 let previous = current.read().unwrap_or_else(|e| e.into_inner()).clone();
1463 if let Err(e) = next.continue_subscriptions(previous.as_deref()).await {
1464 // The tables already listened to still are; a new one is
1465 // answered by the refresh until the next reload retries it.
1466 tracing::warn!("Subscriptions on the rebuilt schema: {}", e);
1467 }
1468 *current.write().unwrap_or_else(|e| e.into_inner()) = Some(Arc::new(next));
1469 info!("GraphQL schema rebuilt");
1470 }
1471 });
1472}
1473
1474/// The members of every table the GraphQL metadata marks as an enumeration,
1475/// keyed "schema.table". A table that cannot be read is left out, with a
1476/// warning, and becomes an ordinary table type.
1477#[cfg(feature = "graphql")]
1478async fn read_enum_values(
1479 pool: &sqlx::PgPool,
1480 cache: &postrust_core::schema_cache::SchemaCache,
1481 names: &postrust_graphql::names::NameOverrides,
1482) -> std::collections::HashMap<String, Vec<(String, Option<String>)>> {
1483 let mut values_of = std::collections::HashMap::new();
1484 for (schema, table) in names.enum_tables() {
1485 let qi = postrust_core::api_request::QualifiedIdentifier::new(&schema, &table);
1486 let Some(definition) = cache.get_table(&qi) else {
1487 tracing::warn!(
1488 "{}.{} is marked as an enumeration but was not found",
1489 schema,
1490 table
1491 );
1492 continue;
1493 };
1494 // One column identifies a member. A composite key names no
1495 // single value, so there is nothing to call the member.
1496 let [key_column] = definition.pk_cols.as_slice() else {
1497 tracing::warn!(
1498 "{}.{} is marked as an enumeration but its primary key is not one column",
1499 schema,
1500 table
1501 );
1502 continue;
1503 };
1504 // Hasura's convention, and a useful one: a `comment` column
1505 // describes each value. Read as a role that may read the
1506 // table, which a NOINHERIT authenticator is not by itself.
1507 match postrust_core::schema_cache::enum_values(
1508 pool,
1509 &schema,
1510 &table,
1511 key_column,
1512 definition.get_column("comment").is_some(),
1513 )
1514 .await
1515 {
1516 Ok(values) => {
1517 values_of.insert(format!("{}.{}", schema, table), values);
1518 }
1519 Err(e) => {
1520 tracing::warn!("cannot read the values of {}.{}: {}", schema, table, e)
1521 }
1522 }
1523 }
1524 values_of
1525}
1526
1527/// Mask database URI for logging.
1528fn mask_db_uri(uri: &str) -> String {
1529 if let Some(at_pos) = uri.find('@') {
1530 if let Some(proto_end) = uri.find("://") {
1531 return format!("{}://***@{}", &uri[..proto_end], &uri[at_pos + 1..]);
1532 }
1533 }
1534 uri.to_string()
1535}
1536
1537#[cfg(test)]
1538mod tests {
1539 use super::*;
1540
1541 fn routes(prefixes: &[&str]) -> Vec<(String, Router<Arc<AppState>>)> {
1542 prefixes
1543 .iter()
1544 .map(|prefix| (prefix.to_string(), Router::new()))
1545 .collect()
1546 }
1547
1548 fn refusal(prefixes: &[&str]) -> String {
1549 match mount_points(routes(prefixes)) {
1550 Ok(_) => panic!("{prefixes:?} was accepted"),
1551 Err(e) => e.to_string(),
1552 }
1553 }
1554
1555 #[test]
1556 fn a_prefix_is_normalized() {
1557 for given in ["hooks", "/hooks", "/hooks/", " /hooks// "] {
1558 let nests = mount_points(routes(&[given])).unwrap();
1559 assert_eq!(nests[0].0, "/hooks", "{given:?}");
1560 }
1561 }
1562
1563 #[test]
1564 fn the_root_and_malformed_prefixes_are_refused() {
1565 for given in ["", "/", "//", "/a//b", "/{id}", "/files/{*rest}", "/*"] {
1566 refusal(&[given]);
1567 }
1568 }
1569
1570 #[test]
1571 fn a_prefix_overlapping_the_servers_own_paths_is_refused() {
1572 for given in [
1573 "/api",
1574 "/api/hooks",
1575 "/healthz",
1576 "/v1",
1577 "/v1/version",
1578 "/v1/graphql/extra",
1579 "/v1alpha1",
1580 "/admin",
1581 "/_",
1582 "/_/health",
1583 "/_/ready/deep",
1584 ] {
1585 let refused = refusal(&[given]);
1586 assert!(refused.contains("serves"), "{given:?}: {refused}");
1587 }
1588 }
1589
1590 #[test]
1591 fn a_prefix_sharing_only_characters_with_a_reserved_one_is_accepted() {
1592 let nests = mount_points(routes(&[
1593 "/apis",
1594 "/v1/other",
1595 "/administration",
1596 "/_/hooks",
1597 "/__",
1598 ]))
1599 .unwrap();
1600 assert_eq!(nests.len(), 5);
1601 }
1602
1603 #[test]
1604 fn two_prefixes_that_overlap_are_refused() {
1605 assert!(refusal(&["/hooks", "hooks/"]).contains("nested earlier"));
1606 assert!(refusal(&["/hooks", "/hooks/stripe"]).contains("nested earlier"));
1607 assert!(refusal(&["/hooks/stripe", "/hooks"]).contains("nested earlier"));
1608 assert!(refusal(&["/_/hooks", "/_/hooks"]).contains("nested earlier"));
1609 }
1610
1611 /// Whatever `mount_points` accepts, the router takes without panicking.
1612 /// The built-in routes are laid out as `assemble` lays them out, with
1613 /// every feature's paths present whether or not the feature is compiled
1614 /// in, and each nested router serving something, since an empty one
1615 /// claims no path.
1616 #[test]
1617 fn what_is_accepted_does_not_panic_the_router() {
1618 let stub = || Router::new().route("/", any(|| async {}));
1619 let nests = mount_points(
1620 [
1621 "/_/hooks",
1622 "/_/jobs/nightly",
1623 "/hooks",
1624 "/v1/other",
1625 "/apis",
1626 ]
1627 .into_iter()
1628 .map(|prefix| (prefix.to_string(), stub()))
1629 .collect(),
1630 )
1631 .unwrap();
1632 let mut app: Router<Arc<AppState>> = Router::new()
1633 .nest("/api", Router::new().route("/{*path}", any(|| async {})))
1634 .route("/healthz", any(|| async {}))
1635 .route("/v1/version", any(|| async {}))
1636 .nest(CUSTOM_PREFIX, custom::custom_router());
1637 for (prefix, router) in nests {
1638 app = app.nest(&prefix, router);
1639 }
1640 let _ = app
1641 .nest("/admin", stub())
1642 .nest("/v1/graphql", stub())
1643 .nest("/v1alpha1/graphql", stub())
1644 .nest("/api/graphql", stub())
1645 .route("/", any(|| async {}))
1646 .fallback(|| async {});
1647 }
1648}