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::types::{AuthProvider, HttpRequestParts, RequestParts};
19use micromegas_tracing::prelude::*;
20use std::sync::Arc;
21
22pub(crate) const HEADER_ACCESS_KEY: &str = "X-Amz-Firehose-Access-Key";
23pub(crate) const HEADER_REQUEST_ID: &str = "X-Amz-Firehose-Request-Id";
24
25/// Ack/error response body per the Firehose HTTP Endpoint Delivery contract:
26/// `{requestId, timestamp}` on success, `{requestId, timestamp, errorMessage}` on failure.
27#[derive(serde::Serialize)]
28struct FirehoseResponseBody<'a> {
29    #[serde(rename = "requestId")]
30    request_id: &'a str,
31    timestamp: i64,
32    #[serde(rename = "errorMessage", skip_serializing_if = "Option::is_none")]
33    error_message: Option<&'a str>,
34}
35
36pub(crate) fn firehose_response(
37    status: StatusCode,
38    request_id: &str,
39    error_message: Option<&str>,
40) -> Response {
41    let body = FirehoseResponseBody {
42        request_id,
43        timestamp: Utc::now().timestamp_millis(),
44        error_message,
45    };
46    Response::builder()
47        .status(status)
48        .header(
49            header::CONTENT_TYPE,
50            HeaderValue::from_static("application/json"),
51        )
52        .body(Body::from(
53            serde_json::to_vec(&body).expect("serializing firehose response"),
54        ))
55        .expect("building firehose response")
56}
57
58pub(crate) fn request_id_from(headers: &HeaderMap) -> String {
59    headers
60        .get(HEADER_REQUEST_ID)
61        .and_then(|v| v.to_str().ok())
62        .unwrap_or("")
63        .to_string()
64}
65
66/// Firehose-specific auth: read `X-Amz-Firehose-Access-Key`, synthesize an
67/// `Authorization: Bearer <key>` header, and validate via the same `AuthProvider` the rest
68/// of the ingestion service uses (reuses the constant-time keyring check verbatim). On
69/// failure, return the Firehose error shape (non-200 JSON) so Firehose retries/spills
70/// rather than dropping data.
71pub(crate) async fn firehose_auth_middleware(
72    provider: Arc<dyn AuthProvider>,
73    mut req: Request,
74    next: Next,
75) -> Response {
76    let request_id = request_id_from(req.headers());
77    let Some(access_key) = req
78        .headers()
79        .get(HEADER_ACCESS_KEY)
80        .and_then(|v| v.to_str().ok())
81    else {
82        return firehose_response(
83            StatusCode::UNAUTHORIZED,
84            &request_id,
85            Some("missing X-Amz-Firehose-Access-Key"),
86        );
87    };
88    let mut headers = req.headers().clone();
89    if let Ok(bearer) = HeaderValue::from_str(&format!("Bearer {access_key}")) {
90        headers.insert(header::AUTHORIZATION, bearer);
91    }
92    let parts = HttpRequestParts {
93        headers,
94        method: req.method().clone(),
95        uri: req.uri().clone(),
96    };
97    match provider.validate_request(&parts as &dyn RequestParts).await {
98        Ok(_ctx) => {
99            // SECURITY: strip any client-provided auth headers to prevent spoofing, matching
100            // the shared `auth_middleware` (auth/src/axum.rs). These headers must only ever be
101            // trusted when set by the authentication layer, never by the incoming request.
102            req.headers_mut().remove("x-auth-subject");
103            req.headers_mut().remove("x-auth-email");
104            req.headers_mut().remove("x-auth-issuer");
105            req.headers_mut().remove("x-allow-delegation");
106            req.headers_mut().remove("x-auth-is-admin");
107            next.run(req).await
108        }
109        Err(e) => {
110            warn!("[firehose auth_failure] {e}");
111            firehose_response(
112                StatusCode::UNAUTHORIZED,
113                &request_id,
114                Some("invalid access key"),
115            )
116        }
117    }
118}
micromegas/servers/firehose_common.rs

micromegas/servers/
firehose_common.rs
