micromegas/servers/
firehose_common.rs

1//! Shared Kinesis Data Firehose HTTP Endpoint Delivery plumbing, reused by every
2//! Firehose-backed ingestion route (metrics, CloudWatch Logs, and any future signal).
3//!
4//! Everything here is signal-agnostic — it only knows about the Firehose transport
5//! (access-key header, ack shape), not what's inside a record. Firehose's only credential
6//! channel is the non-standard `X-Amz-Firehose-Access-Key` header — it cannot send
7//! `Authorization: Bearer`. So a Firehose route cannot sit under the global Bearer
8//! `auth_middleware`; it has its own auth step that synthesizes a bearer header from the
9//! Firehose header and reuses the same `AuthProvider` (constant-time keyring check)
10//! verbatim.
11
12use axum::body::Body;
13use axum::extract::Request;
14use axum::http::{HeaderMap, HeaderValue, StatusCode, header};
15use axum::middleware::Next;
16use axum::response::Response;
17use chrono::Utc;
18use micromegas_auth::client_ip::resolve_client_ip;
19use micromegas_auth::types::{AuthProvider, HttpRequestParts, ProviderUnavailable, RequestParts};
20use micromegas_tracing::prelude::*;
21use std::sync::Arc;
22
23pub(crate) const HEADER_ACCESS_KEY: &str = "X-Amz-Firehose-Access-Key";
24pub(crate) const HEADER_REQUEST_ID: &str = "X-Amz-Firehose-Request-Id";
25
26/// Ack/error response body per the Firehose HTTP Endpoint Delivery contract:
27/// `{requestId, timestamp}` on success, `{requestId, timestamp, errorMessage}` on failure.
28#[derive(serde::Serialize)]
29struct FirehoseResponseBody<'a> {
30    #[serde(rename = "requestId")]
31    request_id: &'a str,
32    timestamp: i64,
33    #[serde(rename = "errorMessage", skip_serializing_if = "Option::is_none")]
34    error_message: Option<&'a str>,
35}
36
37pub(crate) fn firehose_response(
38    status: StatusCode,
39    request_id: &str,
40    error_message: Option<&str>,
41) -> Response {
42    let body = FirehoseResponseBody {
43        request_id,
44        timestamp: Utc::now().timestamp_millis(),
45        error_message,
46    };
47    Response::builder()
48        .status(status)
49        .header(
50            header::CONTENT_TYPE,
51            HeaderValue::from_static("application/json"),
52        )
53        .body(Body::from(
54            serde_json::to_vec(&body).expect("serializing firehose response"),
55        ))
56        .expect("building firehose response")
57}
58
59pub(crate) fn request_id_from(headers: &HeaderMap) -> String {
60    headers
61        .get(HEADER_REQUEST_ID)
62        .and_then(|v| v.to_str().ok())
63        .unwrap_or("")
64        .to_string()
65}
66
67/// Firehose-specific auth: read `X-Amz-Firehose-Access-Key`, synthesize an
68/// `Authorization: Bearer <key>` header, and validate via the same `AuthProvider` the rest
69/// of the ingestion service uses (reuses the constant-time keyring check verbatim). On
70/// failure, return the Firehose error shape (non-200 JSON) so Firehose retries/spills
71/// rather than dropping data.
72pub async fn firehose_auth_middleware(
73    provider: Arc<dyn AuthProvider>,
74    mut req: Request,
75    next: Next,
76) -> Response {
77    let request_id = request_id_from(req.headers());
78    let Some(access_key) = req
79        .headers()
80        .get(HEADER_ACCESS_KEY)
81        .and_then(|v| v.to_str().ok())
82    else {
83        return firehose_response(
84            StatusCode::UNAUTHORIZED,
85            &request_id,
86            Some("missing X-Amz-Firehose-Access-Key"),
87        );
88    };
89    let client_ip = resolve_client_ip(req.headers(), req.extensions());
90    let mut headers = req.headers().clone();
91    if let Ok(bearer) = HeaderValue::from_str(&format!("Bearer {access_key}")) {
92        headers.insert(header::AUTHORIZATION, bearer);
93    }
94    let parts = HttpRequestParts {
95        headers,
96        method: req.method().clone(),
97        uri: req.uri().clone(),
98        client_ip,
99    };
100    match provider.validate_request(&parts as &dyn RequestParts).await {
101        Ok(ctx) => {
102            // SECURITY: strip any client-provided auth headers to prevent spoofing, matching
103            // the shared `auth_middleware` (auth/src/axum.rs). These headers must only ever be
104            // trusted when set by the authentication layer, never by the incoming request.
105            req.headers_mut().remove("x-auth-subject");
106            req.headers_mut().remove("x-auth-email");
107            req.headers_mut().remove("x-auth-issuer");
108            req.headers_mut().remove("x-allow-delegation");
109            req.headers_mut().remove("x-auth-is-admin");
110            // Insert the validated `AuthContext` into the request extensions, mirroring
111            // `auth_middleware`'s success path, so downstream handlers can resolve a write
112            // audience from it.
113            req.extensions_mut().insert(ctx);
114            next.run(req).await
115        }
116        Err(e) => {
117            // A DB-backed `ingestion_api_keys` outage must not look like a
118            // rejected Firehose access key: both Firehose routers validate
119            // through the same `MultiAuthProvider` chain as every other
120            // ingestion route, so an outage here gets the same 503 treatment
121            // `micromegas_auth::axum::auth_middleware` gives the plain routes.
122            if e.downcast_ref::<ProviderUnavailable>().is_some() {
123                debug!("[firehose auth_failure] {e}");
124                firehose_response(
125                    StatusCode::SERVICE_UNAVAILABLE,
126                    &request_id,
127                    Some("auth provider unavailable"),
128                )
129            } else {
130                warn!("[firehose auth_failure] {e}");
131                firehose_response(
132                    StatusCode::UNAUTHORIZED,
133                    &request_id,
134                    Some("invalid access key"),
135                )
136            }
137        }
138    }
139}
micromegas/servers/firehose_common.rs

micromegas/servers/
firehose_common.rs
