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 §ion.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}