fix: don't forward session cookies on intra-cluster syncs

src/handlers/http/cluster/mod.rs

@@ -43,6 +43,7 @@ use url::Url;
 use utils::{IngestionStats, QueriedStats, StorageStats, check_liveness, to_url_string};
 
 use crate::INTRA_CLUSTER_CLIENT;
+use crate::handlers::http::middleware::CLUSTER_SECRET_HEADER;
 use crate::handlers::http::modal::ingest::SyncRole;
 use crate::handlers::http::query::{Query, QueryError, TIME_ELAPSED_HEADER};
 use crate::metrics::prom_utils::Metrics;
@@ -426,23 +427,83 @@ pub fn fallback_intracluster_userid(
     generated_text_reqwest_header_value(username, "configured username").map(Some)
 }
 
-// forward the create/update stream request to all ingestors to keep them in sync
-pub async fn sync_streams_with_ingestors(
-    headers: HeaderMap,
-    body: Bytes,
-    stream_name: &str,
-    tenant_id: &Option<String>,
-) -> Result<(), StreamError> {
-    let mut reqwest_headers = reqwest::header::HeaderMap::new();
-
+/// Copies the caller's headers for a stream sync, minus the session cookie:
+/// sessions live only on the node that issued them, so a peer would reject
+/// it. The flag is true when the copy carries no other credential, in which
+/// case the sync authenticates with the target node's token.
+fn stream_sync_headers(headers: &HeaderMap) -> (reqwest::header::HeaderMap, bool) {
+    let mut forwarded = reqwest::header::HeaderMap::new();
     for (key, value) in headers.iter() {
+        if key == actix_web::http::header::COOKIE {
+            continue;
+        }
         // Convert actix header name/value to reqwest header name/value
         if let Ok(name) = reqwest::header::HeaderName::from_bytes(key.as_str().as_bytes())
             && let Ok(val) = reqwest::header::HeaderValue::from_bytes(value.as_bytes())
         {
-            reqwest_headers.insert(name, val);
+            forwarded.insert(name, val);
         }
     }
+    let has_credentials = [
+        CLUSTER_SECRET_HEADER,
+        reqwest::header::AUTHORIZATION.as_str(),
+        "x-api-key",
+    ]
+    .into_iter()
+    .any(|name| forwarded.contains_key(name));
+    (forwarded, !has_credentials)
+}
+
+/// Without a credential from the caller, a stream sync authenticates with the
+/// node token, i.e. as root in the default tenant. The ingestor would create a
+/// tenant's stream in the default tenant, so that needs the cluster secret.
+fn check_stream_sync_tenant(
+    needs_node_token: bool,
+    stream_name: &str,
+    tenant_id: &Option<String>,
+) -> Result<(), StreamError> {
+    if needs_node_token && let Some(tenant) = tenant_id {
+        return Err(StreamError::Custom {
+            msg: format!(
+                "failed to sync stream {stream_name} of tenant {tenant} to ingestors: P_CLUSTER_SECRET is required"
+            ),
+            status: StatusCode::INTERNAL_SERVER_ERROR,
+        });
+    }
+    Ok(())
+}
+
+/// The headers for one node's stream sync. A node token that can't be encoded
+/// fails only this node's sync: the stream already exists locally, so it is a
+/// sync failure rather than a header-build error.
+fn stream_sync_node_headers(
+    headers: &reqwest::header::HeaderMap,
+    needs_node_token: bool,
+    node: &str,
+    node_token: &str,
+) -> Result<reqwest::header::HeaderMap, StreamError> {
+    let mut headers = headers.clone();
+    if needs_node_token {
+        let token = generated_text_reqwest_header_value(node_token, "node token").map_err(|e| {
+            StreamError::Custom {
+                msg: format!("failed to sync stream to ingestor {node}: {e}"),
+                status: StatusCode::INTERNAL_SERVER_ERROR,
+            }
+        })?;
+        headers.insert(reqwest::header::AUTHORIZATION, token);
+    }
+    Ok(headers)
+}
+
+// forward the create/update stream request to all ingestors to keep them in sync
+pub async fn sync_streams_with_ingestors(
+    headers: HeaderMap,
+    body: Bytes,
+    stream_name: &str,
+    tenant_id: &Option<String>,
+) -> Result<(), StreamError> {
+    let (mut reqwest_headers, needs_node_token) = stream_sync_headers(&headers);
+    check_stream_sync_tenant(needs_node_token, stream_name, tenant_id)?;
 
     if let Some(userid) = fallback_intracluster_userid(
         reqwest_headers.contains_key(INTRA_CLUSTER_USERID),
@@ -466,9 +527,16 @@ pub async fn sync_streams_with_ingestors(
                 base_path_without_preceding_slash(),
                 stream_name
             );
-            let headers = reqwest_headers_clone.clone();
+            let headers =
+                stream_sync_node_headers(
+                &reqwest_headers_clone,
+                needs_node_token,
+                &ingestor.domain_name,
+                &ingestor.token,
+            );
             let body = body_clone.clone();
             async move {
+                let headers = headers?;
                 let res = INTRA_CLUSTER_CLIENT
                     .put(url)
                     .headers(headers)
@@ -2092,6 +2160,91 @@ pub async fn send_query_request(
 #[cfg(test)]
 mod header_build_tests {
     use super::*;
+    use actix_web::http::header::{HeaderName, HeaderValue};
+
+    fn actix_headers(pairs: &[(&'static str, &'static str)]) -> HeaderMap {
+        let mut map = HeaderMap::new();
+        for (name, value) in pairs {
+            map.append(
+                HeaderName::from_static(name),
+                HeaderValue::from_static(value),
+            );
+        }
+        map
+    }
+
+    #[test]
+    fn stream_sync_drops_session_cookie_and_needs_node_token() {
+        let (forwarded, needs_node_token) = stream_sync_headers(&actix_headers(&[
+            ("cookie", "session=abc"),
+            ("x-p-log-source", "json"),
+        ]));
+        assert!(forwarded.get(reqwest::header::COOKIE).is_none());
+        assert_eq!(forwarded.get("x-p-log-source").unwrap(), "json");
+        assert!(needs_node_token);
+    }
+
+    #[test]
+    fn stream_sync_node_headers_add_token_only_without_credentials() {
+        let (base, needs_node_token) = stream_sync_headers(&actix_headers(&[
+            ("cookie", "session=abc"),
+            (INTRA_CLUSTER_USERID, "alice"),
+        ]));
+        let headers =
+            stream_sync_node_headers(&base, needs_node_token, "node", "Basic node").unwrap();
+        assert_eq!(
+            headers.get(reqwest::header::AUTHORIZATION).unwrap(),
+            "Basic node"
+        );
+        assert_eq!(headers.get(INTRA_CLUSTER_USERID).unwrap(), "alice");
+
+        let headers = stream_sync_node_headers(&base, false, "node", "Basic node").unwrap();
+        assert!(headers.get(reqwest::header::AUTHORIZATION).is_none());
+    }
+
+    #[test]
+    fn stream_sync_bad_node_token_is_a_sync_failure() {
+        let err = stream_sync_node_headers(
+            &reqwest::header::HeaderMap::new(),
+            true,
+            "node",
+            "Basic josé",
+        )
+        .unwrap_err();
+        assert!(matches!(err, StreamError::Custom { .. }));
+        assert!(!err.to_string().contains("josé"));
+        assert!(err.to_string().contains("ingestor node"));
+    }
+
+    #[test]
+    fn stream_sync_node_token_is_only_for_the_default_tenant() {
+        let tenant = Some("acme".to_string());
+        assert!(check_stream_sync_tenant(true, "logs", &None).is_ok());
+        assert!(check_stream_sync_tenant(false, "logs", &tenant).is_ok());
+        assert!(matches!(
+            check_stream_sync_tenant(true, "logs", &tenant),
+            Err(StreamError::Custom { .. })
+        ));
+        let msg = check_stream_sync_tenant(true, "logs", &tenant)
+            .unwrap_err()
+            .to_string();
+        assert!(msg.contains("logs") && msg.contains("acme"), "{msg}");
+    }
+
+    #[test]
+    fn stream_sync_keeps_caller_credentials() {
+        for credential in [
+            ("authorization", "Basic dXNlcjpwYXNz"),
+            ("x-api-key", "key"),
+            (CLUSTER_SECRET_HEADER, "hash"),
+        ] {
+            let (forwarded, needs_node_token) =
+                stream_sync_headers(&actix_headers(&[("cookie", "session=abc"), credential]));
+            assert!(forwarded.get(reqwest::header::COOKIE).is_none());
+            assert_eq!(forwarded.get(credential.0).unwrap(), credential.1);
+            assert!(!needs_node_token, "{} is a credential", credential.0);
+        }
+    }
 
     #[test]
     fn node_header_failure_wins_over_earlier_network_failure() {

src/storage/field_stats.rs

@@ -37,7 +37,6 @@ use crate::parseable::PARSEABLE;
 use crate::storage::ObjectStorageError;
 use crate::storage::StreamType;
 use crate::tenants::TENANT_METADATA;
-use crate::utils::create_intracluster_auth_headermap;
 use crate::utils::get_tenant_id_from_request;
 use crate::utils::get_user_from_request;
 use crate::utils::header_parsing::{
@@ -1000,17 +999,7 @@ pub async fn get_dataset_stats(
                     TENANT_METADATA.get_global_query_auth(tenant).as_deref(),
                 )?)
             } else {
-                let auth = create_intracluster_auth_headermap(req.headers(), "", &userid)?;
-                let mut map = HeaderMap::new();
-
-                for (key, value) in auth.iter() {
-                    if let Ok(name) = HeaderName::from_bytes(key.as_str().as_bytes())
-                        && let Ok(val) = HeaderValue::from_bytes(value.as_bytes())
-                    {
-                        map.insert(name, val);
-                    }
-                }
-                Some(map)
+                Some(prism_auth_without_secret(req.headers())?)
             };
 
             let response = match send_query_request(auth, &query_request, &tenant_id).await {
@@ -1035,6 +1024,24 @@ pub async fn get_dataset_stats(
     Ok(response)
 }
 
+/// Auth for forwarding a Prism dataset-stats query to a querier without the
+/// cluster secret: the caller's `Authorization` header, so the querier checks
+/// the caller's own access. Anything else is rejected. A session cookie is
+/// only valid on this node, and leaving the auth empty would make
+/// `send_query_request` use the querier's own token and skip those checks.
+fn prism_auth_without_secret(headers: &HeaderMap) -> Result<HeaderMap, QueryError> {
+    let auth = headers
+        .get(actix_web::http::header::AUTHORIZATION)
+        .ok_or_else(|| {
+            QueryError::CustomError(
+                "P_CLUSTER_SECRET is required to forward this request to a querier".to_string(),
+            )
+        })?;
+    let mut map = HeaderMap::new();
+    map.insert(actix_web::http::header::AUTHORIZATION, auth.clone());
+    Ok(map)
+}
+
 /// Headers for forwarding a Prism dataset-stats query to a querier with the
 /// cluster secret. All values are server-side (stored global query auth,
 /// cluster secret hash, the authenticated user id) apart from the tenant,
@@ -1859,6 +1866,37 @@ mod tests {
 mod forwarding_header_tests {
     use super::*;
 
+    #[test]
+    fn prism_auth_without_secret_forwards_only_authorization() {
+        let mut headers = HeaderMap::new();
+        headers.insert(
+            actix_web::http::header::AUTHORIZATION,
+            HeaderValue::from_static("Basic dXNlcjpwYXNz"),
+        );
+        headers.insert(
+            actix_web::http::header::COOKIE,
+            HeaderValue::from_static("session=abc"),
+        );
+        let map = prism_auth_without_secret(&headers).unwrap();
+        assert_eq!(map.get("authorization").unwrap(), "Basic dXNlcjpwYXNz");
+        assert_eq!(map.len(), 1);
+    }
+
+    #[test]
+    fn prism_auth_without_secret_rejects_cookie_and_api_key_callers() {
+        for (name, value) in [("cookie", "session=abc"), ("x-api-key", "key")] {
+            let mut headers = HeaderMap::new();
+            headers.insert(
+                HeaderName::from_static(name),
+                HeaderValue::from_static(value),
+            );
+            assert!(matches!(
+                prism_auth_without_secret(&headers),
+                Err(QueryError::CustomError(_))
+            ));
+        }
+    }
+
     #[test]
     fn prism_forwarding_headers_include_auth_secret_tenant_and_user() {
         let map = prism_forwarding_headers("hash", "acme", "alice", Some("Basic abc"))

src/utils/mod.rs

@@ -265,6 +265,12 @@ pub fn is_admin_for_session(session_key: &SessionKey) -> bool {
 
 /// Builds the auth headers used to forward a request to another node.
 ///
+/// With a cluster secret the peer authenticates the caller by `userid`.
+/// Without one, the caller's `Authorization` header is forwarded if present;
+/// otherwise the request authenticates as this node with `node_token`. A
+/// session cookie is never forwarded, since sessions live only on the node
+/// that issued them and the peer would reject it.
+///
 /// A malformed tenant header on the incoming request is a client error
 /// (`HeaderBuildError::Input`, 400). Values the server stored or generated
 /// itself (cluster secret hash, user id, node token) failing to encode is a
@@ -318,11 +324,6 @@ fn build_intracluster_auth_headermap(
         );
     } else if let Some(auth) = req.get(actix_web::http::header::AUTHORIZATION) {
         map.insert(reqwest::header::AUTHORIZATION, copied_header_value(auth)?);
-    } else if req.contains_key(actix_web::http::header::COOKIE) {
-        // multiple cookies
-        for cookie in req.get_all(actix_web::http::header::COOKIE) {
-            map.insert(reqwest::header::COOKIE, copied_header_value(cookie)?);
-        }
     } else {
         map.insert(
             reqwest::header::AUTHORIZATION,
@@ -456,11 +457,38 @@ mod tests {
     }
 
     #[test]
-    fn forwarding_copies_cookie_header() {
+    fn forwarding_replaces_session_cookie_with_node_token() {
+        let req = headers(&[(
+            "cookie",
+            HeaderValue::from_static("session=01M6ARA6TCK200TXKYC6CWRN5D; username=admin"),
+        )]);
+        let map = build_intracluster_auth_headermap(&req, "Basic node", "alice", None).unwrap();
+        assert!(map.get(reqwest::header::COOKIE).is_none());
+        assert_eq!(
+            map.get(reqwest::header::AUTHORIZATION).unwrap(),
+            "Basic node"
+        );
+    }
+
+    #[test]
+    fn forwarding_never_copies_cookie() {
         let req = headers(&[("cookie", HeaderValue::from_static("session=abc"))]);
-        let map = build_intracluster_auth_headermap(&req, "node", "alice", None).unwrap();
-        assert_eq!(map.get(reqwest::header::COOKIE).unwrap(), "session=abc");
-        assert!(map.get(reqwest::header::AUTHORIZATION).is_none());
+        for (req, secret) in [
+            (req.clone(), Some("hash")),
+            (
+                headers(&[
+                    ("cookie", HeaderValue::from_static("session=abc")),
+                    (
+                        "authorization",
+                        HeaderValue::from_static("Basic dXNlcjpwYXNz"),
+                    ),
+                ]),
+                None,
+            ),
+        ] {
+            let map = build_intracluster_auth_headermap(&req, "node", "alice", secret).unwrap();
+            assert!(map.get(reqwest::header::COOKIE).is_none());
+        }
     }
 
     #[test]
@@ -531,11 +559,14 @@ mod tests {
     #[test]
     fn forwarding_ignores_unused_generated_text() {
         // a copied credential wins over the node token, so a bad token is never encoded
-        let req = headers(&[("cookie", HeaderValue::from_bytes(b"s=caf\xc3\xa9").unwrap())]);
+        let req = headers(&[(
+            "authorization",
+            HeaderValue::from_bytes(b"Basic caf\xc3\xa9").unwrap(),
+        )]);
         let map = build_intracluster_auth_headermap(&req, "Basic josé", "josé", None).unwrap();
         assert_eq!(
-            map.get(reqwest::header::COOKIE).unwrap().as_bytes(),
-            b"s=caf\xc3\xa9"
+            map.get(reqwest::header::AUTHORIZATION).unwrap().as_bytes(),
+            b"Basic caf\xc3\xa9"
         );
         assert!(map.get("intra-cluster-userid").is_none());
     }