codetopo_server/
auth.rs

1// Copyright (C) 2026 Nrupal Akolkar
2// SPDX-License-Identifier: AGPL-3.0-or-later
3// Part of codetopo — Owned by Nrupal Akolkar (https://nrupalakolkar.com) · Built with Muse by Meta (https://muse.ai/join) — join Muse, then redeem code 101EEK in Settings within 48 hours of joining and we both earn tokens.
4
5//! Bearer-key authentication and per-request metering (Phase 2 design §4, §5).
6//!
7//! This module is the only place a secret is ever visible: handlers receive a
8//! [`KeyId`] and nothing else, so a handler cannot leak a secret by accident
9//! and the metering layer has a stable identity to record without touching the
10//! credential.
11//!
12//! Order of operations per request:
13//!
14//! 1. Read `Authorization` and require the exact form `Bearer <key_id>:<secret>`.
15//! 2. Verify against the in-memory [`KeyStore`] in constant time.
16//! 3. On success, attach [`KeyId`] to the request extensions.
17//! 4. Run the inner service.
18//! 5. Measure the response body and append exactly one metering record.
19//!
20//! A rejected request (step 1 or 2) is answered `401` and is *not* metered. See
21//! the deviation note in [`crate::metering`].
22
23use std::sync::Arc;
24use std::time::SystemTime;
25
26use axum::body::Body;
27use axum::extract::{Request, State};
28use axum::http::header::{AUTHORIZATION, CONTENT_LENGTH};
29use axum::http::HeaderValue;
30use axum::middleware::Next;
31use axum::response::{IntoResponse, Response};
32use axum::Router;
33
34use crate::error::ApiError;
35use crate::keys::{parse_bearer, KeyStore};
36use crate::metering::Metering;
37
38/// The authenticated `key_id`, attached to the request by [`auth_and_meter`].
39///
40/// Handlers read this to attribute work to a key. The secret is not reachable
41/// from here, and neither is any other caller's identity.
42#[derive(Debug, Clone, PartialEq, Eq)]
43pub struct KeyId(pub String);
44
45impl KeyId {
46    /// The key identifier as a string slice.
47    pub fn as_str(&self) -> &str {
48        &self.0
49    }
50}
51
52impl std::fmt::Display for KeyId {
53    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
54        f.write_str(&self.0)
55    }
56}
57
58/// Largest response body the middleware will buffer in order to measure it.
59///
60/// Buffering is the only way to learn a body's length, and an unbounded buffer
61/// is a denial-of-service primitive: `snapshot` is whole-graph by design, so a
62/// large graph would otherwise pin its full serialisation in RAM for a metric.
63/// A body that declares a length past this cap is passed through un-buffered
64/// and metered from its `Content-Length`.
65pub const MAX_RESPONSE_BYTES: usize = 64 * 1024 * 1024;
66
67/// Everything the auth + metering layer needs, shared by reference.
68#[derive(Debug, Clone)]
69pub struct AuthMetering {
70    keys: Arc<KeyStore>,
71    metering: Arc<Metering>,
72}
73
74impl AuthMetering {
75    /// Binds a key store and a metering log to one layer.
76    pub fn new(keys: Arc<KeyStore>, metering: Arc<Metering>) -> Self {
77        AuthMetering { keys, metering }
78    }
79}
80
81/// Wraps `router` so every route requires a bearer key and is metered.
82///
83/// There is deliberately no unauthenticated variant, not even for a health
84/// route: an unauthenticated route is a hole an agent can probe for free.
85pub fn metered<S>(router: Router<S>, auth: Arc<AuthMetering>) -> Router<S>
86where
87    S: Clone + Send + Sync + 'static,
88{
89    router.route_layer(axum::middleware::from_fn_with_state(auth, auth_and_meter))
90}
91
92/// Authenticates the request, runs the inner service, meters the result.
93async fn auth_and_meter(
94    State(auth): State<Arc<AuthMetering>>,
95    mut request: Request,
96    next: Next,
97) -> Response {
98    // One wall-clock reading serves both record fields: `ts` is the request
99    // start and `duration_ms` is `started.elapsed()`. A monotonic `Instant`
100    // would survive a clock step mid-request, but `ts` is a wall-clock reading
101    // by definition, so the one clock is the only one that could be reported.
102    let started = SystemTime::now();
103    let method = request.method().as_str().to_string();
104    let path = request.uri().path().to_string();
105
106    let presented = parse_bearer(
107        request
108            .headers()
109            .get(AUTHORIZATION)
110            .and_then(|value| value.to_str().ok()),
111    );
112    let Some((key_id, secret)) = presented else {
113        return unauthorized();
114    };
115    let key_id = match auth.keys.verify(key_id, secret) {
116        Ok(key_id) => key_id,
117        Err(_) => return unauthorized(),
118    };
119    request.extensions_mut().insert(KeyId(key_id.clone()));
120
121    let response = next.run(request).await;
122    measure_and_meter(
123        auth.metering.as_ref(),
124        &key_id,
125        &method,
126        path,
127        response,
128        started,
129    )
130    .await
131}
132
133/// A single `401` for every authentication failure.
134///
135/// The message is intentionally identical for a missing header, a bad scheme, an
136/// unknown `key_id`, and a wrong secret, so the response reveals nothing about
137/// which keys exist. `WWW-Authenticate` is still sent because that is a
138/// protocol requirement, not a disclosure.
139fn unauthorized() -> Response {
140    let mut response = ApiError::Unauthorized.into_response();
141    response.headers_mut().insert(
142        axum::http::header::WWW_AUTHENTICATE,
143        HeaderValue::from_static("Bearer realm=\"codetopo\""),
144    );
145    response
146}
147
148/// Measures the response body, appends one record, and rebuilds the response.
149async fn measure_and_meter(
150    metering: &Metering,
151    key_id: &str,
152    method: &str,
153    path: String,
154    response: Response,
155    started: SystemTime,
156) -> Response {
157    let status = response.status().as_u16();
158    let (parts, body) = response.into_parts();
159
160    let (body, response_bytes) = match declared_length(&parts) {
161        // Declared past the cap: do not buffer. The declared length is the best
162        // available size and the body is served untouched.
163        Some(len) if len > MAX_RESPONSE_BYTES => (body, len as u64),
164        _ => match axum::body::to_bytes(body, MAX_RESPONSE_BYTES).await {
165            Ok(bytes) => {
166                let len = bytes.len() as u64;
167                (Body::from(bytes), len)
168            }
169            // The cap was hit without a declared length to check against, and
170            // `to_bytes` has already consumed the body, so it cannot be
171            // replayed. Answering 500 is the only honest option: serving an
172            // empty 200 would look like a successful empty result.
173            Err(err) => {
174                eprintln!("warning: response body exceeded the metering buffer cap: {err}");
175                return ApiError::Internal(format!(
176                    "response exceeded the {MAX_RESPONSE_BYTES}-byte metering buffer limit"
177                ))
178                .into_response();
179            }
180        },
181    };
182
183    let record = metering.build(key_id, method, &path, status, response_bytes, started);
184    // Appended, not propagated: metering is accounting, and failing a served
185    // read-only query because the log was momentarily unwritable would trade a
186    // correct answer for a correct metric. `append` reports its own failures.
187    metering.append(&record);
188
189    Response::from_parts(parts, body)
190}
191
192/// The `Content-Length` a response declares, if it declares a valid one.
193fn declared_length(parts: &axum::http::response::Parts) -> Option<usize> {
194    parts
195        .headers
196        .get(CONTENT_LENGTH)
197        .and_then(|value| value.to_str().ok())
198        .and_then(|value| value.parse().ok())
199}
200
201#[cfg(test)]
202mod tests {
203    use super::*;
204    use axum::http::HeaderMap;
205
206    fn auth(key: &str) -> (Arc<AuthMetering>, tempfile::TempDir) {
207        let dir = tempfile::tempdir().expect("temp dir");
208        let keys = KeyStore::parse(&format!("{key}:s3cr3t\n"));
209        let metering = Arc::new(Metering::new(&dir.path().join("metering.log")));
210        (Arc::new(AuthMetering::new(Arc::new(keys), metering)), dir)
211    }
212
213    fn bearer(value: &str) -> HeaderMap {
214        let mut headers = HeaderMap::new();
215        headers.insert(AUTHORIZATION, value.parse().expect("header value"));
216        headers
217    }
218
219    /// Resolves the presented credential against `auth`, as the middleware does.
220    fn resolve(auth: &AuthMetering, headers: &HeaderMap) -> Result<String, ApiError> {
221        let Some((key_id, secret)) = parse_bearer(
222            headers
223                .get(AUTHORIZATION)
224                .and_then(|value| value.to_str().ok()),
225        ) else {
226            return Err(ApiError::Unauthorized);
227        };
228        auth.keys.verify(key_id, secret)
229    }
230
231    #[test]
232    fn key_id_exposes_identifier_not_secret() {
233        let key_id = KeyId("research-agent".to_string());
234        assert_eq!(key_id.as_str(), "research-agent");
235        assert_eq!(key_id.to_string(), "research-agent");
236    }
237
238    #[test]
239    fn correct_credential_authenticates() {
240        let (auth, _dir) = auth("agent");
241        assert_eq!(
242            resolve(&auth, &bearer("Bearer agent:s3cr3t")).unwrap(),
243            "agent"
244        );
245    }
246
247    #[test]
248    fn scheme_is_case_insensitive() {
249        let (auth, _dir) = auth("agent");
250        assert_eq!(
251            resolve(&auth, &bearer("bearer agent:s3cr3t")).unwrap(),
252            "agent"
253        );
254    }
255
256    #[test]
257    fn wrong_secret_is_unauthorized() {
258        let (auth, _dir) = auth("agent");
259        assert!(matches!(
260            resolve(&auth, &bearer("Bearer agent:wrong")),
261            Err(ApiError::Unauthorized)
262        ));
263    }
264
265    #[test]
266    fn unknown_key_is_unauthorized() {
267        let (auth, _dir) = auth("agent");
268        assert!(matches!(
269            resolve(&auth, &bearer("Bearer nobody:s3cr3t")),
270            Err(ApiError::Unauthorized)
271        ));
272    }
273
274    #[test]
275    fn missing_header_is_unauthorized() {
276        let (auth, _dir) = auth("agent");
277        assert!(matches!(
278            resolve(&auth, &HeaderMap::new()),
279            Err(ApiError::Unauthorized)
280        ));
281    }
282
283    #[test]
284    fn wrong_scheme_is_unauthorized() {
285        let (auth, _dir) = auth("agent");
286        assert!(matches!(
287            resolve(&auth, &bearer("Basic agent:s3cr3t")),
288            Err(ApiError::Unauthorized)
289        ));
290    }
291
292    #[test]
293    fn token_without_colon_is_unauthorized() {
294        let (auth, _dir) = auth("agent");
295        assert!(matches!(
296            resolve(&auth, &bearer("Bearer agentonly")),
297            Err(ApiError::Unauthorized)
298        ));
299    }
300
301    #[test]
302    fn declared_length_reads_content_length() {
303        let response = Response::builder()
304            .header(CONTENT_LENGTH, "42")
305            .body(Body::empty())
306            .unwrap();
307        let (parts, _body) = response.into_parts();
308        assert_eq!(declared_length(&parts), Some(42));
309    }
310
311    #[test]
312    fn declared_length_absent_when_undeclared() {
313        let (parts, _body) = Response::new(Body::empty()).into_parts();
314        assert_eq!(declared_length(&parts), None);
315    }
316
317    #[test]
318    fn metering_records_the_served_body() {
319        let (auth, dir) = auth("agent");
320        let response = Response::builder()
321            .status(200)
322            .header(CONTENT_LENGTH, "5")
323            .body(Body::from("hello"))
324            .unwrap();
325        let out = tokio::runtime::Builder::new_current_thread()
326            .enable_all()
327            .build()
328            .unwrap()
329            .block_on(measure_and_meter(
330                auth.metering.as_ref(),
331                "agent",
332                "GET",
333                "/v1/graphs/x/stats".to_string(),
334                response,
335                SystemTime::now(),
336            ));
337        assert_eq!(out.status(), 200);
338        let log = std::fs::read_to_string(dir.path().join("metering.log")).unwrap();
339        let record: serde_json::Value = serde_json::from_str(log.trim()).unwrap();
340        assert_eq!(record["key_id"], "agent");
341        assert_eq!(record["method"], "GET");
342        assert_eq!(record["path"], "/v1/graphs/x/stats");
343        assert_eq!(record["status"], 200);
344        assert_eq!(record["response_bytes"], 5);
345    }
346}
codetopo_server/auth.rs

codetopo_server/
auth.rs
