diff --git a/src/handlers/http/cluster/mod.rs b/src/handlers/http/cluster/mod.rs index 1130557d5..ef91339db 100644 --- a/src/handlers/http/cluster/mod.rs +++ b/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, -) -> 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, +) -> 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 { + 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, +) -> 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() { diff --git a/src/storage/field_stats.rs b/src/storage/field_stats.rs index ae204410d..b78ca9b17 100644 --- a/src/storage/field_stats.rs +++ b/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 { + 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")) diff --git a/src/utils/mod.rs b/src/utils/mod.rs index 325f61a2b..654278cb1 100644 --- a/src/utils/mod.rs +++ b/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()); }