Skip to main content

dynamic_config_server/
routes.rs

1//! The HTTP surface.
2//!
3//! Spring Cloud Config Server's shape, because a deployment's mental model
4//! transfers for free: the first path segment is the application, the second
5//! is the profile, and what hangs off them is this crate's own vocabulary.
6//!
7//! # One endpoint returns values
8//!
9//! `GET /{application}/{profile}` is the handover — the resolved document,
10//! secrets included, which is what a config server is *for*. Every other
11//! endpoint returns shape, provenance or counts: paths without what is at
12//! them, an explanation with every value replaced by `***`, a check report
13//! that names keys and origins, a status that is timestamps and numbers,
14//! and a metrics scrape that is the same numbers with a label naming the
15//! section they belong to.
16//!
17//! That line is drawn once, here, and it is drawn wider than the library
18//! draws it. `explain` in the library deliberately *does* carry values —
19//! you asked, at a terminal, for one path. Over a socket the same answer is
20//! a value that has left the process for a reason nobody weighed, so the
21//! server pushes every explanation through
22//! [`Explanation::redacted`](dynamic_config::Explanation::redacted) rather
23//! than only the paths it believes are secret. Reusing the library's
24//! redaction rather than writing a second one is the point; applying it
25//! unconditionally is the server's own decision.
26//!
27//! # It will not be an oracle
28//!
29//! A caller that may not read `billing` and a caller asking for an
30//! application nobody serves get the same 404, with the same body, having
31//! done the same work: authorisation is decided from the caller's grants
32//! alone, and the section map is never consulted for an application the
33//! caller was not granted. There is nothing to time and nothing to read.
34
35use std::future::Future;
36use std::pin::Pin;
37use std::sync::Arc;
38use std::task::{Context, Poll};
39use std::time::Duration;
40
41use axum::extract::{Path, State};
42use axum::http::{header, HeaderMap, StatusCode};
43use axum::response::sse::{Event, KeepAlive, Sse};
44use axum::response::{IntoResponse, Response};
45use axum::routing::get;
46use axum::{Json, Router};
47use dynamic_config::telemetry::Exposition;
48use dynamic_config::Changes;
49use futures_core::Stream;
50use serde::Serialize;
51
52use crate::audit::{AuditEntry, Outcome};
53use crate::auth::Principal;
54use crate::document::Document;
55use crate::server::{Section, Server, StreamPermit};
56
57/// The router, over a started [`Server`].
58///
59/// Everything is a `GET`: this server serves configuration and changes
60/// nothing, so there is no verb here that could.
61pub fn router(server: Arc<Server>) -> Router {
62    Router::new()
63        .route("/healthz", get(healthz))
64        .route("/readyz", get(readyz))
65        .route("/metrics", get(metrics))
66        .route("/{application}/{profile}", get(document))
67        .route("/{application}/{profile}/paths", get(paths))
68        .route("/{application}/{profile}/check", get(check))
69        .route("/{application}/{profile}/status", get(status))
70        .route("/{application}/{profile}/stream", get(stream))
71        .route("/{application}/{profile}/explain/{path}", get(explain))
72        // So that a path this server does not route answers exactly like a
73        // section it will not serve: same status, same body.
74        .fallback(|| async { not_found() })
75        .with_state(server)
76}
77
78// ---------------------------------------------------------------------------
79// Liveness and readiness. Unauthenticated, and so they say nothing: not how
80// many sections there are, not which one is unhappy. An operator reads
81// `/{application}/{profile}/status` for that, with a credential.
82// ---------------------------------------------------------------------------
83
84async fn healthz() -> Response {
85    (StatusCode::OK, Json(serde_json::json!({ "status": "ok" }))).into_response()
86}
87
88async fn readyz(State(server): State<Arc<Server>>) -> Response {
89    let ready = server.is_ready();
90    let code = if ready {
91        StatusCode::OK
92    } else {
93        StatusCode::SERVICE_UNAVAILABLE
94    };
95
96    (code, Json(serde_json::json!({ "ready": ready }))).into_response()
97}
98
99// ---------------------------------------------------------------------------
100// Metrics. Authenticated, and scoped to the caller's grants — see the
101// handler.
102// ---------------------------------------------------------------------------
103
104/// The Prometheus exposition format's content type. Version 0.0.4 is the
105/// text format every scraper reads, and naming it is what makes a scraper
106/// parse the body rather than store it.
107const EXPOSITION: &str = "text/plain; version=0.0.4; charset=utf-8";
108
109/// `GET /metrics` — one scrape, covering **the sections this caller may
110/// read** and no others.
111///
112/// # Why this one is authenticated when `/healthz` and `/readyz` are not
113///
114/// Those two answer a boolean and say nothing else: not how many sections
115/// there are, not which one is unhappy. That is precisely what lets them be
116/// open. A useful metrics endpoint cannot be that — a series that cannot
117/// name the section it describes is a series nobody can alert on — so
118/// `/metrics` carries application and profile labels, and an application
119/// name is exactly what the not-an-oracle property exists to withhold. An
120/// open `/metrics` would enumerate every service the fleet configures to
121/// anyone who could reach the port.
122///
123/// So it takes the same bearer token as everything else and, having taken
124/// it, reports only what that principal is already entitled to ask for one
125/// section at a time through `/status`. A scraper is a client like any
126/// other: give it a token and grant it the applications it should see.
127/// Prometheus has read `authorization` and `bearer_token_file` from its
128/// scrape configuration for years, so this costs a deployment two lines.
129///
130/// The alternative — an open endpoint with no labels, counting sections in
131/// aggregate — was rejected: it says less than `/readyz` already does and
132/// still cannot be alerted on.
133///
134/// # Cardinality
135///
136/// Bounded by the served set, not by the documents. `6 × sections` series
137/// at a scrape, and `19 × sections` over a process's life once the two
138/// fixed enums behind the `reason` and `kind` labels are counted. **No key
139/// path, file name or value can become a label**: every sample here comes
140/// from [`ConfigStatus`](dynamic_config::ConfigStatus), which holds none of
141/// them, and the two labels this crate adds are an application and a
142/// profile that the server's own configuration named and that `is_name`
143/// has already bounded.
144async fn metrics(State(server): State<Arc<Server>>, headers: HeaderMap) -> Response {
145    let authorization = headers
146        .get(header::AUTHORIZATION)
147        .and_then(|value| value.to_str().ok());
148
149    let Some(principal) = server.authenticate(authorization) else {
150        return refuse(&server, "metrics", Outcome::Unauthenticated, None, None);
151    };
152
153    let mut exposition = Exposition::new();
154
155    for section in server.sections() {
156        // The same grant check every other endpoint makes, and for the same
157        // reason: a caller learns the shape of its own sections and of
158        // nothing else. A principal granted nothing gets a well-formed
159        // empty scrape rather than a refusal — it is somebody, and there is
160        // nothing to tell it.
161        if !principal.may_read(section.application()) {
162            continue;
163        }
164
165        exposition.add_with(
166            &[
167                ("application", section.application()),
168                ("profile", section.profile()),
169            ],
170            &section.status(),
171        );
172    }
173
174    server.record(&AuditEntry {
175        caller: Some(principal.name().to_owned()),
176        // A scrape is about the server, not about one section: naming a
177        // section here would be naming several.
178        application: None,
179        profile: None,
180        endpoint: "metrics",
181        outcome: Outcome::Served,
182        generation: None,
183    });
184
185    ([(header::CONTENT_TYPE, EXPOSITION)], exposition.render()).into_response()
186}
187
188// ---------------------------------------------------------------------------
189// The served endpoints.
190// ---------------------------------------------------------------------------
191
192#[derive(Serialize)]
193struct DocumentBody<'a> {
194    application: &'a str,
195    profile: &'a str,
196    generation: u64,
197    config: &'a serde_json::Value,
198}
199
200/// The resolved document. **The one endpoint that returns values.**
201async fn document(
202    State(server): State<Arc<Server>>,
203    headers: HeaderMap,
204    Path((application, profile)): Path<(String, String)>,
205) -> Response {
206    let admitted = match admit(&server, &headers, &application, &profile, "document") {
207        Ok(admitted) => admitted,
208        Err(response) => return *response,
209    };
210
211    // One coherent pair: the number never describes an install this
212    // document is not already at least at. See `Section::installed`.
213    let Some((generation, document)) = admitted.section.installed() else {
214        return unready(&server, &admitted);
215    };
216
217    served(&server, &admitted, generation);
218
219    Json(DocumentBody {
220        application: admitted.section.application(),
221        profile: admitted.section.profile(),
222        generation,
223        config: document.as_json(),
224    })
225    .into_response()
226}
227
228#[derive(Serialize)]
229struct PathsBody<'a> {
230    application: &'a str,
231    profile: &'a str,
232    generation: u64,
233    paths: Vec<String>,
234}
235
236/// Which keys exist, without what is at them.
237///
238/// The endpoint a dashboard or a schema check wants: it answers "does this
239/// deployment set `pool.max_size`" without becoming a way to read it.
240async fn paths(
241    State(server): State<Arc<Server>>,
242    headers: HeaderMap,
243    Path((application, profile)): Path<(String, String)>,
244) -> Response {
245    let admitted = match admit(&server, &headers, &application, &profile, "paths") {
246        Ok(admitted) => admitted,
247        Err(response) => return *response,
248    };
249
250    let Some((generation, document)) = admitted.section.installed() else {
251        return unready(&server, &admitted);
252    };
253
254    served(&server, &admitted, generation);
255
256    Json(PathsBody {
257        application: admitted.section.application(),
258        profile: admitted.section.profile(),
259        generation,
260        paths: document.leaf_paths(),
261    })
262    .into_response()
263}
264
265#[derive(Serialize)]
266struct StatusBody<'a> {
267    application: &'a str,
268    profile: &'a str,
269    generation: u64,
270    ready: bool,
271    healthy: bool,
272    consecutive_failures: u32,
273    stale_for_seconds: Option<f64>,
274    last_reason: Option<&'static str>,
275    last_failure: Option<FailureBody>,
276}
277
278#[derive(Serialize)]
279struct FailureBody {
280    kind: &'static str,
281    path: String,
282    seconds_ago: f64,
283}
284
285/// The operational surface: which generation is live, when it landed, why,
286/// and how the reloads since have gone.
287///
288/// A handful of atomic loads and no I/O, so a scrape per second costs
289/// nothing. The reason is the *category* — `file-changed` rather than the
290/// path that changed — because a metric dimension carrying a filesystem path
291/// has unbounded cardinality and because the path is not the caller's
292/// business.
293async fn status(
294    State(server): State<Arc<Server>>,
295    headers: HeaderMap,
296    Path((application, profile)): Path<(String, String)>,
297) -> Response {
298    let admitted = match admit(&server, &headers, &application, &profile, "status") {
299        Ok(admitted) => admitted,
300        Err(response) => return *response,
301    };
302
303    let status = admitted.section.status();
304    let generation = status.generation;
305    served(&server, &admitted, generation);
306
307    Json(StatusBody {
308        application: admitted.section.application(),
309        profile: admitted.section.profile(),
310        generation,
311        ready: admitted.section.is_ready(),
312        healthy: status.is_healthy(),
313        consecutive_failures: status.consecutive_failures,
314        stale_for_seconds: status.stale_for().map(|elapsed| elapsed.as_secs_f64()),
315        last_reason: status
316            .last_reason
317            .as_ref()
318            .map(dynamic_config::ReloadReason::as_str),
319        last_failure: status.last_failure.as_ref().map(|failure| FailureBody {
320            kind: failure.kind.as_str(),
321            path: failure.path.clone(),
322            seconds_ago: failure.at.elapsed().as_secs_f64(),
323        }),
324    })
325    .into_response()
326}
327
328// ---------------------------------------------------------------------------
329// The change stream.
330// ---------------------------------------------------------------------------
331
332/// How long a stream may be silent before a comment goes down it.
333///
334/// Not a configuration key: it is the interval an idle TCP connection needs
335/// to survive the proxies these deployments sit behind, and a deployment
336/// that wants a different one has an idle timeout of its own to set.
337const KEEP_ALIVE: Duration = Duration::from_secs(15);
338
339/// `GET /{application}/{profile}/stream` — one `text/event-stream` per
340/// caller, one event per install.
341///
342/// # What an event carries, and what it deliberately does not
343///
344/// A generation number, and the application and profile it belongs to:
345///
346/// ```text
347/// id: 7
348/// event: generation
349/// data: {"application":"billing","profile":"prod","generation":7}
350/// ```
351///
352/// **Not the document, and not the changed paths either.** The document
353/// endpoint is the one endpoint that serves values, and a stream that
354/// carried them would be a second one — with a different lifetime, a
355/// different failure mode, and a body that outlives the request that
356/// authorised it. Changed *paths* would leak nothing `/paths` does not
357/// already tell the same caller, but they would have to be diffed per
358/// install and carried per connection, which is the memory bound this
359/// design exists to avoid. So the event says *something landed, here is its
360/// number*, and the client re-fetches the endpoint it was already using.
361///
362/// That choice is what makes the rest of it simple:
363///
364/// - **Resumption is a comparison, not a buffer.** A generation is
365///   monotonic and the current one subsumes every one before it, so
366///   `Last-Event-ID: 6` against a section at 9 is one event carrying 9 —
367///   nothing was missed, because there was never anything to miss. There is
368///   no ring of recent events, so there is no bound to choose and no
369///   "reconnected past the end of it" case to answer.
370/// - **Backpressure needs no policy.** The stream carries a level rather
371///   than a log: a client that stops reading is simply not polled, and when
372///   it is polled again it gets the *latest* generation. Nothing queues, so
373///   nothing has to be dropped.
374/// - **Memory is flat.** Per connection: one `Changes` handle (an `Arc`
375///   clone and a `u64`), one registered waker, and two short strings. No
376///   document, no diff, no channel. A thousand pods reconnecting after a
377///   restart cost a thousand of that and one shared install.
378///
379/// # It is an endpoint like every other one
380///
381/// Authenticated, authorised against the caller's grants, and refused with
382/// the same 404 as everything else — a subscription to a section the caller
383/// may not read does the same work and returns the same body as a
384/// subscription to a section nobody serves. The audit log records the
385/// *subscription*, once, with the generation it opened at; the events after
386/// it say no more than a `/status` poll would, and a line per install per
387/// connection would drown the log that matters.
388async fn stream(
389    State(server): State<Arc<Server>>,
390    headers: HeaderMap,
391    Path((application, profile)): Path<(String, String)>,
392) -> Response {
393    let admitted = match admit(&server, &headers, &application, &profile, "stream") {
394        Ok(admitted) => admitted,
395        Err(response) => return *response,
396    };
397
398    // A deployment that turns streaming off does not serve this path, and
399    // says so with the body it says everything else with. Checked after
400    // admission so that the answer costs an unauthorised caller exactly
401    // what every other 404 costs it.
402    if !server.streams_enabled() {
403        return refuse(
404            &server,
405            "stream",
406            Outcome::NotFound,
407            Some(admitted.principal.name().to_owned()),
408            Some((
409                admitted.section.application().to_owned(),
410                admitted.section.profile().to_owned(),
411            )),
412        );
413    }
414
415    let Some(permit) = server.open_stream() else {
416        return at_capacity(&server, &admitted);
417    };
418
419    let section = Arc::clone(admitted.section);
420    let generation = section.generation();
421
422    served(&server, &admitted, generation);
423
424    let stream = Generations {
425        application: section.application().to_owned(),
426        profile: section.profile().to_owned(),
427        changes: section.changes(),
428        resume: last_event_id(&headers),
429        sent: None,
430        section,
431        _permit: permit,
432    };
433
434    Sse::new(stream)
435        .keep_alive(KeepAlive::new().interval(KEEP_ALIVE))
436        .into_response()
437}
438
439/// `Last-Event-ID`, as the generation a client says it already has.
440///
441/// Anything that is not a number is *ignored* rather than refused: the
442/// header is echoed back by browsers and proxies from whatever the last
443/// event carried, and a reconnect that fails because something in the path
444/// mangled a header is a worse failure than one extra event.
445fn last_event_id(headers: &HeaderMap) -> Option<u64> {
446    headers
447        .get("last-event-id")
448        .and_then(|value| value.to_str().ok())
449        .and_then(|value| value.trim().parse::<u64>().ok())
450}
451
452/// One connection's stream of generations.
453///
454/// Everything it holds is fixed-size. That is the property the whole design
455/// turns on, so it is worth naming the fields: two strings from the server's
456/// own configuration, an `Arc` to the section, a `Changes` handle, the last
457/// number sent, the number the client claimed, and the permit that releases
458/// its place in the ceiling on drop.
459struct Generations {
460    application: String,
461    profile: String,
462    section: Arc<Section>,
463    changes: Changes<Document>,
464    /// The last generation this connection emitted.
465    sent: Option<u64>,
466    /// What the client's `Last-Event-ID` claimed, if anything.
467    resume: Option<u64>,
468    _permit: StreamPermit,
469}
470
471impl Generations {
472    /// Whether `generation` is news to this connection.
473    fn is_news(&self, generation: u64) -> bool {
474        // Zero is "nothing has ever been installed here", which is not an
475        // event: a section serves nothing until it has a document, and
476        // `/readyz` is where that is reported.
477        if generation == 0 {
478            return false;
479        }
480
481        match self.sent {
482            Some(sent) => generation > sent,
483            // The opening event. A client that said where it was gets one
484            // unless the section is exactly where it said; a client that
485            // said nothing gets one either way, so that it starts knowing
486            // where it stands rather than having to guess.
487            //
488            // *Different*, not *greater*. A generation counts installs
489            // since this process started, so a restart puts the section
490            // back at 1 while a reconnecting `EventSource` still sends the
491            // `Last-Event-ID` the previous process gave it. Under a
492            // greater-than test, a client resuming from 50 would be told
493            // nothing by the new process until it had reloaded fifty times
494            // — silently missing every change in between, for the life of
495            // the connection. A number that is not the one the client
496            // holds is news, whichever side of it it falls.
497            None => match self.resume {
498                Some(resumed) => generation != resumed,
499                None => true,
500            },
501        }
502    }
503
504    fn event(&self, generation: u64) -> Event {
505        let data = serde_json::json!({
506            "application": self.application,
507            "profile": self.profile,
508            "generation": generation,
509        })
510        .to_string();
511
512        // The id *is* the generation, which is what makes `Last-Event-ID`
513        // resumption a comparison rather than a lookup.
514        Event::default()
515            .id(generation.to_string())
516            .event("generation")
517            .data(data)
518    }
519}
520
521impl Stream for Generations {
522    // Infallible: nothing between an install and an event can fail. A
523    // connection ends because the client went away, which is the body being
524    // dropped rather than an error travelling down it.
525    type Item = Result<Event, std::convert::Infallible>;
526
527    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
528        let this = self.get_mut();
529
530        loop {
531            let generation = this.section.generation();
532
533            if this.is_news(generation) {
534                this.sent = Some(generation);
535
536                return Poll::Ready(Some(Ok(this.event(generation))));
537            }
538
539            // A fresh future per poll on purpose: `changed()` keeps no state
540            // of its own — the generation it has seen lives in the `Changes`
541            // handle and the check-register-check protocol lives in the
542            // cell — so re-creating it is the same future, and it saves
543            // this type from having to be self-referential.
544            let changed = this.changes.changed();
545
546            match std::pin::pin!(changed).poll(context) {
547                // Something installed. Round again to read the generation it
548                // became, rather than trusting the snapshot: the number the
549                // event carries is the one `/status` and the document
550                // endpoint report, and it comes from the same load.
551                Poll::Ready(_) => continue,
552                Poll::Pending => return Poll::Pending,
553            }
554        }
555    }
556}
557
558#[derive(Serialize)]
559struct CheckBody<'a> {
560    application: &'a str,
561    profile: &'a str,
562    clean: bool,
563    resolved: Vec<ResolvedRow>,
564    unknown: Vec<UnknownRow>,
565    failure: Option<String>,
566}
567
568#[derive(Serialize)]
569struct ResolvedRow {
570    path: String,
571    origin: String,
572}
573
574#[derive(Serialize)]
575struct UnknownRow {
576    path: String,
577    suggestion: Option<String>,
578}
579
580/// Would the *next* load succeed, and where would each key come from.
581///
582/// Re-reads the sources — that is the question it answers — so it runs on
583/// the blocking pool rather than on a request worker. It is therefore the
584/// one endpoint here that costs I/O per request; bounding how often an
585/// authorised caller may ask is the job of the thing in front, which is
586/// where rate limiting lives — per-caller limiting needs a view of every
587/// caller, and this process is one replica of several.
588///
589/// `failure` is the library's own error text, which is value-free by policy
590/// and by `dynamic-config/tests/security.rs`. It can name a *file*, which
591/// is the point of the endpoint and is a file the caller is authorised for.
592///
593/// `unknown` is **always empty here**, and that is a property of the shape
594/// rather than a gap in the plumbing: unknown-key detection compares the
595/// resolved keys against a struct's field names, and a config server does
596/// not know its callers' structs — that is the whole reason the served
597/// document is schemaless. A caller that wants the check run against its
598/// own type runs it in its own process, where the type is.
599async fn check(
600    State(server): State<Arc<Server>>,
601    headers: HeaderMap,
602    Path((application, profile)): Path<(String, String)>,
603) -> Response {
604    let admitted = match admit(&server, &headers, &application, &profile, "check") {
605        Ok(admitted) => admitted,
606        Err(response) => return *response,
607    };
608
609    let sources = admitted.section.sources();
610    let report = match dynamic_config::off_thread(move || sources.check()).await {
611        Ok(report) => report,
612        Err(error) => return unavailable(&server, &admitted, &error),
613    };
614
615    let generation = admitted.section.generation();
616    served(&server, &admitted, generation);
617
618    Json(CheckBody {
619        application: admitted.section.application(),
620        profile: admitted.section.profile(),
621        clean: report.is_clean(),
622        resolved: report
623            .resolved
624            .into_iter()
625            .map(|resolved| ResolvedRow {
626                path: resolved.path,
627                origin: resolved.origin.to_string(),
628            })
629            .collect(),
630        unknown: report
631            .unknown
632            .into_iter()
633            .map(|unknown| UnknownRow {
634                path: unknown.path,
635                suggestion: unknown.suggestion,
636            })
637            .collect(),
638        failure: report.failure,
639    })
640    .into_response()
641}
642
643#[derive(Serialize)]
644struct ExplainBody<'a> {
645    application: &'a str,
646    profile: &'a str,
647    path: String,
648    winner: Option<&'static str>,
649    rows: Vec<ExplainRow>,
650}
651
652#[derive(Serialize)]
653struct ExplainRow {
654    layer: &'static str,
655    origin: Option<String>,
656    /// `***` where the layer supplies something, `null` where it does not —
657    /// the shape [`Explanation::redacted`] leaves behind. Never a value.
658    value: Option<String>,
659}
660
661/// Why a value is what it is, without saying what it is.
662///
663/// The feature nobody else has: an operator asks the *server* which layer
664/// won, from a laptop, without shelling into a pod. Every value is `***`.
665async fn explain(
666    State(server): State<Arc<Server>>,
667    headers: HeaderMap,
668    Path((application, profile, path)): Path<(String, String, String)>,
669) -> Response {
670    let admitted = match admit(&server, &headers, &application, &profile, "explain") {
671        Ok(admitted) => admitted,
672        Err(response) => return *response,
673    };
674
675    // After admission, so a caller who may not read this section learns
676    // nothing from the shape of a path it was never going to get.
677    if !is_key_path(&path) {
678        return refuse(
679            &server,
680            "explain",
681            Outcome::Malformed,
682            Some(admitted.principal.name().to_owned()),
683            Some((
684                admitted.section.application().to_owned(),
685                admitted.section.profile().to_owned(),
686            )),
687        );
688    }
689
690    let sources = admitted.section.sources();
691    let asked = path.clone();
692    let explanation = match dynamic_config::off_thread(move || sources.explain(&asked)).await {
693        // Redacted unconditionally: see this module's documentation.
694        Ok(explanation) => explanation.redacted(),
695        Err(error) => return unavailable(&server, &admitted, &error),
696    };
697
698    let generation = admitted.section.generation();
699    served(&server, &admitted, generation);
700
701    Json(ExplainBody {
702        application: admitted.section.application(),
703        profile: admitted.section.profile(),
704        path,
705        winner: explanation.winner().map(|row| row.layer),
706        rows: explanation
707            .rows()
708            .iter()
709            .map(|row| ExplainRow {
710                layer: row.layer,
711                origin: row.origin.as_ref().map(ToString::to_string),
712                value: row.value.clone(),
713            })
714            .collect(),
715    })
716    .into_response()
717}
718
719// ---------------------------------------------------------------------------
720// Admission: authenticate, then check the request's shape, then authorise,
721// then — and only then — look anything up.
722// ---------------------------------------------------------------------------
723
724struct Admitted<'a> {
725    principal: Principal,
726    section: &'a Arc<Section>,
727    endpoint: &'static str,
728}
729
730fn admit<'a>(
731    server: &'a Server,
732    headers: &HeaderMap,
733    application: &str,
734    profile: &str,
735    endpoint: &'static str,
736) -> Result<Admitted<'a>, Box<Response>> {
737    let authorization = headers
738        .get(header::AUTHORIZATION)
739        .and_then(|value| value.to_str().ok());
740
741    let Some(principal) = server.authenticate(authorization) else {
742        return Err(Box::new(refuse(
743            server,
744            endpoint,
745            Outcome::Unauthenticated,
746            None,
747            None,
748        )));
749    };
750
751    // Before either segment is looked up *or logged*: a path segment is
752    // attacker-controlled text, and an audit line is a place where a newline
753    // in one would be a forged second line.
754    if !is_name(application) || !is_name(profile) {
755        return Err(Box::new(refuse(
756            server,
757            endpoint,
758            Outcome::Malformed,
759            Some(principal.name().to_owned()),
760            None,
761        )));
762    }
763
764    let caller = Some(principal.name().to_owned());
765    let where_ = Some((application.to_owned(), profile.to_owned()));
766
767    // Authorisation first, and the map is not touched when it fails. This is
768    // the whole not-an-oracle property: "not yours" and "no such thing"
769    // reach the same line below by the same route.
770    if !principal.may_read(application) {
771        return Err(Box::new(refuse(
772            server,
773            endpoint,
774            Outcome::NotFound,
775            caller,
776            where_,
777        )));
778    }
779
780    let Some(section) = server.section(application, profile) else {
781        return Err(Box::new(refuse(
782            server,
783            endpoint,
784            Outcome::NotFound,
785            caller,
786            where_,
787        )));
788    };
789
790    Ok(Admitted {
791        principal,
792        section,
793        endpoint,
794    })
795}
796
797fn served(server: &Server, admitted: &Admitted<'_>, generation: u64) {
798    server.record(&AuditEntry {
799        caller: Some(admitted.principal.name().to_owned()),
800        application: Some(admitted.section.application().to_owned()),
801        profile: Some(admitted.section.profile().to_owned()),
802        endpoint: admitted.endpoint,
803        outcome: Outcome::Served,
804        generation: Some(generation),
805    });
806}
807
808fn refuse(
809    server: &Server,
810    endpoint: &'static str,
811    outcome: Outcome,
812    caller: Option<String>,
813    where_: Option<(String, String)>,
814) -> Response {
815    let (application, profile) = match where_ {
816        Some((application, profile)) => (Some(application), Some(profile)),
817        None => (None, None),
818    };
819
820    server.record(&AuditEntry {
821        caller,
822        application,
823        profile,
824        endpoint,
825        outcome,
826        generation: None,
827    });
828
829    match outcome {
830        Outcome::Unauthenticated => unauthenticated(),
831        _ => not_found(),
832    }
833}
834
835fn unready(server: &Server, admitted: &Admitted<'_>) -> Response {
836    server.record(&AuditEntry {
837        caller: Some(admitted.principal.name().to_owned()),
838        application: Some(admitted.section.application().to_owned()),
839        profile: Some(admitted.section.profile().to_owned()),
840        endpoint: admitted.endpoint,
841        outcome: Outcome::Unavailable,
842        generation: None,
843    });
844
845    (
846        StatusCode::SERVICE_UNAVAILABLE,
847        Json(serde_json::json!({ "error": "unavailable" })),
848    )
849        .into_response()
850}
851
852/// Every change stream this process will hold is already held.
853///
854/// A 503 with a `Retry-After`, so a herd that hit the ceiling backs off
855/// instead of spinning — the same courtesy the rate-limiting note asks of
856/// whatever ends up doing rate limiting. It is deliberately *not* a 429:
857/// nothing about this caller was excessive, the process is full.
858fn at_capacity(server: &Server, admitted: &Admitted<'_>) -> Response {
859    server.record(&AuditEntry {
860        caller: Some(admitted.principal.name().to_owned()),
861        application: Some(admitted.section.application().to_owned()),
862        profile: Some(admitted.section.profile().to_owned()),
863        endpoint: admitted.endpoint,
864        outcome: Outcome::Unavailable,
865        generation: None,
866    });
867
868    (
869        StatusCode::SERVICE_UNAVAILABLE,
870        [(header::RETRY_AFTER, "5")],
871        Json(serde_json::json!({ "error": "unavailable" })),
872    )
873        .into_response()
874}
875
876/// A diagnostic endpoint whose sources could not be read at all.
877///
878/// The category and the key path, not the message: a 500 body is the one
879/// place free text would travel furthest, and `ErrorKind` plus the path is
880/// what an operator acts on. `/check` is the endpoint that reports *why* a
881/// load would fail, and it reports it in its own `failure` field.
882fn unavailable(
883    server: &Server,
884    admitted: &Admitted<'_>,
885    error: &dynamic_config::Error,
886) -> Response {
887    server.record(&AuditEntry {
888        caller: Some(admitted.principal.name().to_owned()),
889        application: Some(admitted.section.application().to_owned()),
890        profile: Some(admitted.section.profile().to_owned()),
891        endpoint: admitted.endpoint,
892        outcome: Outcome::Unavailable,
893        generation: None,
894    });
895
896    (
897        StatusCode::INTERNAL_SERVER_ERROR,
898        Json(serde_json::json!({
899            "error": "unavailable",
900            "kind": error.kind().as_str(),
901            "path": error.path(),
902        })),
903    )
904        .into_response()
905}
906
907fn unauthenticated() -> Response {
908    (
909        StatusCode::UNAUTHORIZED,
910        [(header::WWW_AUTHENTICATE, "Bearer")],
911        Json(serde_json::json!({ "error": "unauthenticated" })),
912    )
913        .into_response()
914}
915
916/// The single refusal body. "Not yours" and "no such thing" are this, and
917/// nothing else is, so no caller can tell them apart.
918fn not_found() -> Response {
919    (
920        StatusCode::NOT_FOUND,
921        Json(serde_json::json!({ "error": "not_found" })),
922    )
923        .into_response()
924}
925
926/// An application or profile: a first character that is a letter or a digit,
927/// then letters, digits, `.`, `_` and `-`, to 64.
928///
929/// Narrow on purpose. It bounds what can reach the audit log, it rejects
930/// `..` and the empty segment without a special case for either, and there
931/// is no application name anybody wants that it refuses.
932///
933/// `pub(crate)` so that [`ServerConfig::validate`](crate::ServerConfig)
934/// refuses at startup exactly what the handlers refuse at request time: a
935/// section this rejects would load, start and answer nothing.
936pub(crate) fn is_name(value: &str) -> bool {
937    let mut characters = value.chars();
938
939    characters
940        .next()
941        .is_some_and(|first| first.is_ascii_alphanumeric())
942        && value.len() <= 64
943        && characters.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
944}
945
946/// A dotted key path: non-empty segments of letters, digits, `_` and `-`,
947/// to 256 characters overall.
948fn is_key_path(value: &str) -> bool {
949    !value.is_empty()
950        && value.len() <= 256
951        && value.split('.').all(|segment| {
952            !segment.is_empty()
953                && segment
954                    .chars()
955                    .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '-'))
956        })
957}
958
959#[cfg(test)]
960mod tests {
961    use super::*;
962
963    #[test]
964    fn a_name_is_narrow_and_rejects_traversal_and_newlines() {
965        assert!(is_name("billing"));
966        assert!(is_name("billing-api"));
967        assert!(is_name("billing.api_2"));
968
969        assert!(!is_name(""), "the empty segment");
970        assert!(!is_name(".."), "traversal");
971        assert!(!is_name(".hidden"), "a leading dot");
972        assert!(!is_name("bill ing"), "a space");
973        assert!(!is_name("billing\nadmin"), "a forged audit line");
974        assert!(!is_name("bill/ing"), "a separator");
975        assert!(!is_name(&"a".repeat(65)), "unbounded length");
976    }
977
978    #[test]
979    fn a_key_path_is_dotted_and_has_no_empty_segments() {
980        assert!(is_key_path("port"));
981        assert!(is_key_path("pool.max_size"));
982        assert!(is_key_path("a-b.c-d"));
983
984        assert!(!is_key_path(""));
985        assert!(!is_key_path("."));
986        assert!(!is_key_path("pool."));
987        assert!(!is_key_path(".pool"));
988        assert!(!is_key_path("pool..max"));
989        assert!(!is_key_path("pool max"));
990        assert!(!is_key_path(&"a".repeat(257)));
991    }
992}