diff --git a/src/environmentd/src/http.rs b/src/environmentd/src/http.rs index 04b3fb28252e4..28154d0dad169 100644 --- a/src/environmentd/src/http.rs +++ b/src/environmentd/src/http.rs @@ -188,6 +188,9 @@ pub struct HttpConfig { pub routes_enabled: HttpRoutesEnabled, /// Locator for cluster replica HTTP addresses, used for proxying requests. pub replica_http_locator: Arc, + /// Whether a trusted proxy (`balancerd`) fronts this listener, and thus + /// whether a PROXY protocol header on the connection can be believed. + pub behind_trusted_proxy: bool, } #[derive(Debug, Clone)] @@ -221,6 +224,7 @@ struct HelmChartVersion(Option); pub struct HttpServer { tls: Option, router: Router, + behind_trusted_proxy: bool, } impl HttpServer { @@ -247,6 +251,7 @@ impl HttpServer { internal_route_config, routes_enabled, replica_http_locator, + behind_trusted_proxy, }: HttpConfig, ) -> HttpServer { let tls_enabled = tls.is_some(); @@ -679,7 +684,11 @@ impl HttpServer { .merge(base_router) .apply_default_layers(source, metrics); - HttpServer { tls, router } + HttpServer { + tls, + router, + behind_trusted_proxy, + } } } @@ -693,16 +702,24 @@ impl Server for HttpServer { ) -> ConnectionHandler { let router = self.router.clone(); let tls_context = self.tls.clone(); + let behind_trusted_proxy = self.behind_trusted_proxy; let mut conn = TokioIo::new(conn); - Box::pin(async { + Box::pin(async move { let direct_peer_addr = conn.inner().peer_addr().context("fetching peer addr")?; - let peer_addr = conn - .inner_mut() - .take_proxy_header_address() - .await - .map(|a| a.source) - .unwrap_or(direct_peer_addr); + // A PROXY protocol header is only believable when a proxy fronts + // this listener. Elsewhere the header bytes are left in the stream, + // so the request fails to parse as HTTP, which is the right outcome + // for a client that had no business sending one. + let peer_addr = if behind_trusted_proxy { + conn.inner_mut() + .take_proxy_header_address() + .await + .map(|a| a.source) + .unwrap_or(direct_peer_addr) + } else { + direct_peer_addr + }; let (conn, conn_protocol) = match tls_context { Some(tls_context) => { diff --git a/src/environmentd/src/lib.rs b/src/environmentd/src/lib.rs index 1f0ab1fa41eea..ddef19a2e1963 100644 --- a/src/environmentd/src/lib.rs +++ b/src/environmentd/src/lib.rs @@ -303,6 +303,7 @@ impl Listener { active_connection_counter, helm_chart_version, allowed_roles: self.config.allowed_roles, + behind_trusted_proxy: self.config.behind_trusted_proxy, }); mz_server_core::serve(ServeConfig { conns: self.connection_stream, @@ -450,6 +451,7 @@ impl Listeners { internal_route_config: Arc::clone(&internal_route_config), routes_enabled: listener.config.routes, replica_http_locator: Arc::clone(&config.controller.replica_http_locator), + behind_trusted_proxy: listener.config.behind_trusted_proxy, }; http_listener_handles.insert(name.clone(), listener.serve_http(http_config).await); } diff --git a/src/environmentd/src/test_util.rs b/src/environmentd/src/test_util.rs index 24d0d17749ba6..208600c3c54be 100644 --- a/src/environmentd/src/test_util.rs +++ b/src/environmentd/src/test_util.rs @@ -156,12 +156,14 @@ impl Default for TestHarness { authenticator_kind: AuthenticatorKind::None, allowed_roles: AllowedRoles::Normal, enable_tls: false, + behind_trusted_proxy: false, }, "internal".to_owned() => SqlListenerConfig { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, allowed_roles: AllowedRoles::NormalAndInternal, enable_tls: false, + behind_trusted_proxy: false, }, ], http: btreemap![ @@ -169,6 +171,7 @@ impl Default for TestHarness { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, enable_tls: false, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Enabled(AllowedRoles::Normal), webhook: RouteGroup::Enabled(AllowedRoles::Normal), @@ -184,6 +187,7 @@ impl Default for TestHarness { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, enable_tls: false, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), webhook: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), @@ -311,6 +315,18 @@ impl TestHarness { self } + /// Models listeners that a trusted proxy (`balancerd`) fronts, so that + /// forwarded connection metadata is honored. + pub fn behind_trusted_proxy(mut self) -> Self { + for (_, listener) in &mut self.listeners_config.sql { + listener.behind_trusted_proxy = true; + } + for (_, listener) in &mut self.listeners_config.http { + listener.behind_trusted_proxy = true; + } + self + } + pub fn unsafe_mode(mut self) -> Self { self.unsafe_mode = true; self @@ -339,12 +355,14 @@ impl TestHarness { authenticator_kind: AuthenticatorKind::Frontegg, allowed_roles: AllowedRoles::Normal, enable_tls, + behind_trusted_proxy: false, }, "internal".to_owned() => SqlListenerConfig { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, allowed_roles: AllowedRoles::NormalAndInternal, enable_tls: false, + behind_trusted_proxy: false, }, }, http: btreemap! { @@ -352,6 +370,7 @@ impl TestHarness { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::Frontegg, enable_tls, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Enabled(AllowedRoles::Normal), webhook: RouteGroup::Enabled(AllowedRoles::Normal), @@ -367,6 +386,7 @@ impl TestHarness { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, enable_tls: false, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), webhook: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), @@ -398,12 +418,14 @@ impl TestHarness { authenticator_kind: AuthenticatorKind::Oidc, allowed_roles: AllowedRoles::NormalAndInternal, enable_tls, + behind_trusted_proxy: false, }, "internal".to_owned() => SqlListenerConfig { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, allowed_roles: AllowedRoles::NormalAndInternal, enable_tls: false, + behind_trusted_proxy: false, }, }, http: btreemap! { @@ -411,6 +433,7 @@ impl TestHarness { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::Oidc, enable_tls, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), webhook: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), @@ -426,6 +449,7 @@ impl TestHarness { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, enable_tls: false, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), webhook: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), @@ -478,6 +502,7 @@ impl TestHarness { authenticator_kind: AuthenticatorKind::Password, allowed_roles: AllowedRoles::NormalAndInternal, enable_tls, + behind_trusted_proxy: false, }, }, http: btreemap! { @@ -485,6 +510,7 @@ impl TestHarness { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::Password, enable_tls, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), webhook: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), @@ -500,6 +526,7 @@ impl TestHarness { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, enable_tls: false, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Disabled, webhook: RouteGroup::Disabled, @@ -526,6 +553,7 @@ impl TestHarness { authenticator_kind: AuthenticatorKind::Sasl, allowed_roles: AllowedRoles::NormalAndInternal, enable_tls, + behind_trusted_proxy: false, }, }, http: btreemap! { @@ -533,6 +561,7 @@ impl TestHarness { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::Password, enable_tls, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), webhook: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), @@ -548,6 +577,7 @@ impl TestHarness { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, enable_tls: false, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Disabled, webhook: RouteGroup::Disabled, diff --git a/src/environmentd/tests/pgwire.rs b/src/environmentd/tests/pgwire.rs index 24835b387efab..67af1b46795e9 100644 --- a/src/environmentd/tests/pgwire.rs +++ b/src/environmentd/tests/pgwire.rs @@ -1294,3 +1294,83 @@ fn test_pgtest_mz_frontend_occ_pipelined_dml() { }, ); } + +const CLIENT_IP_QUERY: &str = + "SELECT client_ip FROM mz_internal.mz_sessions WHERE connection_id = pg_backend_pid()"; + +/// Runs `query` on a connection whose startup packet carries `params`, and +/// returns the single text value it selects. +/// +/// The whole exchange is pipelined and terminated up front so the server closes +/// the connection, which lets the response be read and parsed in one shot. +fn query_with_startup_params( + addr: std::net::SocketAddr, + params: Vec<(&str, &str)>, + query: &str, +) -> String { + use postgres_protocol::message::backend::Message; + use postgres_protocol::message::frontend; + + let mut buf = BytesMut::new(); + frontend::startup_message(params, &mut buf).unwrap(); + frontend::query(query, &mut buf).unwrap(); + frontend::terminate(&mut buf); + + let mut stream = TcpStream::connect(addr).unwrap(); + stream.write_all(&buf).unwrap(); + let mut response = vec![]; + stream.read_to_end(&mut response).unwrap(); + + let mut response = BytesMut::from(&response[..]); + let mut values = vec![]; + while let Some(message) = Message::parse(&mut response).unwrap() { + if let Message::DataRow(body) = message { + let buf = body.buffer().to_vec(); + values.extend( + body.ranges() + .map(|range| Ok(String::from_utf8(buf[range.unwrap()].to_vec()).unwrap())) + .collect::>() + .unwrap(), + ); + } + } + values.into_element() +} + +// A client that reaches a listener directly must not be able to choose the +// client IP recorded for its session, as network policies are evaluated +// against it. Only a listener that a trusted proxy fronts may believe +// `mz_forwarded_for`. +#[mz_ore::test] +#[allow(clippy::disallowed_methods)] +fn test_forwarded_client_ip_requires_trusted_proxy() { + let untrusted = test_util::TestHarness::default().start_blocking(); + assert_eq!( + query_with_startup_params( + untrusted.sql_local_addr(), + vec![ + ("user", "materialize"), + ("mz_forwarded_for", "1.2.3.4"), + ("welcome_message", "off"), + ], + CLIENT_IP_QUERY, + ), + "127.0.0.1", + ); + + let trusted = test_util::TestHarness::default() + .behind_trusted_proxy() + .start_blocking(); + assert_eq!( + query_with_startup_params( + trusted.sql_local_addr(), + vec![ + ("user", "materialize"), + ("mz_forwarded_for", "1.2.3.4"), + ("welcome_message", "off"), + ], + CLIENT_IP_QUERY, + ), + "1.2.3.4", + ); +} diff --git a/src/environmentd/tests/server.rs b/src/environmentd/tests/server.rs index e0e8b7b01e91d..8e4b24f72cab6 100644 --- a/src/environmentd/tests/server.rs +++ b/src/environmentd/tests/server.rs @@ -7356,3 +7356,63 @@ fn test_startup_only_system_var_warns() { "RESET ALL warned without changing anything, notices: {notices:?}" ); } + +// A client that reaches a listener directly must not be able to choose the +// client IP recorded for its session, as network policies are evaluated +// against it. Only a listener that a trusted proxy fronts may believe a PROXY +// protocol header. +#[mz_ore::test] +#[allow(clippy::disallowed_methods)] +fn test_proxy_header_requires_trusted_proxy() { + const SPOOFED_IP: Ipv4Addr = Ipv4Addr::new(1, 2, 3, 4); + const CLIENT_IP_QUERY: &str = + "SELECT client_ip FROM mz_internal.mz_sessions WHERE connection_id = pg_backend_pid()"; + + /// A PROXY protocol v2 header for an IPv4 TCP connection from `source`. + fn proxy_v2_header(source: Ipv4Addr) -> Vec { + let mut header = b"\r\n\r\n\x00\r\nQUIT\n".to_vec(); + header.extend([0x21, 0x11]); // v2 PROXY command, IPv4 over TCP + header.extend(12u16.to_be_bytes()); // length of the address block + header.extend(source.octets()); + header.extend(Ipv4Addr::LOCALHOST.octets()); + header.extend(1111u16.to_be_bytes()); // source port + header.extend(1111u16.to_be_bytes()); // destination port + header + } + + fn query_behind_proxy_header(addr: std::net::SocketAddr) -> String { + let body = serde_json::json!({ "query": CLIENT_IP_QUERY }).to_string(); + let request = format!( + "POST /api/sql HTTP/1.1\r\n\ + Host: {addr}\r\n\ + Content-Type: application/json\r\n\ + Content-Length: {}\r\n\ + Connection: close\r\n\r\n{body}", + body.len(), + ); + let mut stream = std::net::TcpStream::connect(addr).unwrap(); + stream.write_all(&proxy_v2_header(SPOOFED_IP)).unwrap(); + stream.write_all(request.as_bytes()).unwrap(); + let mut response = vec![]; + std::io::Read::read_to_end(&mut stream, &mut response).unwrap(); + String::from_utf8_lossy(&response).into_owned() + } + + // The header bytes stay in the stream, so the request does not parse as + // HTTP and never reaches the session. + let untrusted = test_util::TestHarness::default().start_blocking(); + let response = query_behind_proxy_header(untrusted.http_local_addr()); + assert!( + !response.contains(&SPOOFED_IP.to_string()), + "unexpected response: {response}" + ); + + let trusted = test_util::TestHarness::default() + .behind_trusted_proxy() + .start_blocking(); + let response = query_behind_proxy_header(trusted.http_local_addr()); + assert!( + response.contains(&SPOOFED_IP.to_string()), + "unexpected response: {response}" + ); +} diff --git a/src/materialized/ci/listener_configs/v0_147_0/frontegg_https.json b/src/materialized/ci/listener_configs/v0_147_0/frontegg_https.json index 5ecd3ebcc5202..44cf218c60975 100644 --- a/src/materialized/ci/listener_configs/v0_147_0/frontegg_https.json +++ b/src/materialized/ci/listener_configs/v0_147_0/frontegg_https.json @@ -4,7 +4,8 @@ "addr": "0.0.0.0:6875", "authenticator_kind": "Frontegg", "allowed_roles": "Normal", - "enable_tls": false + "enable_tls": false, + "behind_trusted_proxy": true }, "internal": { "addr": "0.0.0.0:6877", @@ -19,6 +20,7 @@ "authenticator_kind": "Frontegg", "allowed_roles": "Normal", "enable_tls": true, + "behind_trusted_proxy": true, "routes": { "base": true, "webhook": true, diff --git a/src/materialized/ci/listener_configs/v0_147_0/no_auth_https.json b/src/materialized/ci/listener_configs/v0_147_0/no_auth_https.json index b332af78f0c6c..1557d0d1f5d16 100644 --- a/src/materialized/ci/listener_configs/v0_147_0/no_auth_https.json +++ b/src/materialized/ci/listener_configs/v0_147_0/no_auth_https.json @@ -4,7 +4,8 @@ "addr": "0.0.0.0:6875", "authenticator_kind": "None", "allowed_roles": "Normal", - "enable_tls": false + "enable_tls": false, + "behind_trusted_proxy": true }, "internal": { "addr": "0.0.0.0:6877", @@ -19,6 +20,7 @@ "authenticator_kind": "None", "allowed_roles": "NormalAndInternal", "enable_tls": true, + "behind_trusted_proxy": true, "routes": { "base": true, "webhook": true, diff --git a/src/materialized/ci/listener_configs/v26_32_0/frontegg_https.json b/src/materialized/ci/listener_configs/v26_32_0/frontegg_https.json index 4c2428fe99254..edea7cf258e9c 100644 --- a/src/materialized/ci/listener_configs/v26_32_0/frontegg_https.json +++ b/src/materialized/ci/listener_configs/v26_32_0/frontegg_https.json @@ -5,7 +5,8 @@ "addr": "0.0.0.0:6875", "authenticator_kind": "Frontegg", "allowed_roles": "Normal", - "enable_tls": false + "enable_tls": false, + "behind_trusted_proxy": true }, "internal": { "addr": "0.0.0.0:6877", @@ -19,6 +20,7 @@ "addr": "0.0.0.0:6876", "authenticator_kind": "Frontegg", "enable_tls": true, + "behind_trusted_proxy": true, "routes": { "base": { "enabled": true, diff --git a/src/materialized/ci/listener_configs/v26_32_0/no_auth_https.json b/src/materialized/ci/listener_configs/v26_32_0/no_auth_https.json index 4e311411c4ef0..7f9084ea91c68 100644 --- a/src/materialized/ci/listener_configs/v26_32_0/no_auth_https.json +++ b/src/materialized/ci/listener_configs/v26_32_0/no_auth_https.json @@ -5,7 +5,8 @@ "addr": "0.0.0.0:6875", "authenticator_kind": "None", "allowed_roles": "Normal", - "enable_tls": false + "enable_tls": false, + "behind_trusted_proxy": true }, "internal": { "addr": "0.0.0.0:6877", @@ -19,6 +20,7 @@ "addr": "0.0.0.0:6876", "authenticator_kind": "None", "enable_tls": true, + "behind_trusted_proxy": true, "routes": { "base": { "enabled": true, diff --git a/src/orchestratord/src/controller/materialize/generation.rs b/src/orchestratord/src/controller/materialize/generation.rs index c99c43c7b1765..ad3e18e30cf05 100644 --- a/src/orchestratord/src/controller/materialize/generation.rs +++ b/src/orchestratord/src/controller/materialize/generation.rs @@ -1319,6 +1319,10 @@ fn create_v0_147_0_listeners_config( &mz.spec.internal_certificate_spec, ); let authenticator_kind = mz.spec.authenticator_kind; + // Only `balancerd` may set a session's client IP, and only the external + // listeners sit behind it. Without a `balancerd` deployment nothing between + // the client and `environmentd` is trusted to describe the connection. + let behind_trusted_proxy = config.create_balancers; let mut listeners_config = v0_147_0::ListenersConfig { sql: btreemap! { @@ -1330,6 +1334,7 @@ fn create_v0_147_0_listeners_config( authenticator_kind, allowed_roles: AllowedRoles::Normal, enable_tls: external_enable_tls, + behind_trusted_proxy, }, "internal".to_owned() => SqlListenerConfig{ addr: SocketAddr::new( @@ -1340,6 +1345,7 @@ fn create_v0_147_0_listeners_config( // Should this just be Internal? allowed_roles: AllowedRoles::NormalAndInternal, enable_tls: false, + behind_trusted_proxy: false, }, }, http: btreemap! { @@ -1359,6 +1365,7 @@ fn create_v0_147_0_listeners_config( }, allowed_roles: AllowedRoles::Normal, enable_tls: external_enable_tls, + behind_trusted_proxy, }, routes: v0_147_0::HttpRoutes{ base: true, @@ -1381,6 +1388,7 @@ fn create_v0_147_0_listeners_config( // Should this just be Internal? allowed_roles: AllowedRoles::NormalAndInternal, enable_tls: false, + behind_trusted_proxy: false, }, routes: v0_147_0::HttpRoutes{ base: true, @@ -1427,6 +1435,7 @@ fn create_v0_147_0_listeners_config( authenticator_kind: AuthenticatorKind::None, allowed_roles: AllowedRoles::NormalAndInternal, enable_tls: false, + behind_trusted_proxy: false, }, routes: v0_147_0::HttpRoutes { base: false, diff --git a/src/pgwire/src/server.rs b/src/pgwire/src/server.rs index 0204d947502a7..7660666c98a96 100644 --- a/src/pgwire/src/server.rs +++ b/src/pgwire/src/server.rs @@ -27,7 +27,7 @@ use openssl::ssl::Ssl; use tokio::io::AsyncWriteExt; use tokio_metrics::TaskMetrics; use tokio_openssl::SslStream; -use tracing::{debug, error, trace}; +use tracing::{debug, error, trace, warn}; use crate::codec::FramedConn; use crate::metrics::{Metrics, MetricsConfig}; @@ -60,6 +60,10 @@ pub struct Config { pub helm_chart_version: Option, /// Whether to allow reserved users (ie: mz_system). pub allowed_roles: AllowedRoles, + /// Whether a trusted proxy (`balancerd`) fronts this listener, and thus + /// whether the forwarded connection metadata in the startup parameters can + /// be believed. + pub behind_trusted_proxy: bool, } /// A server that communicates with clients via the pgwire protocol. @@ -73,6 +77,7 @@ pub struct Server { active_connection_counter: ConnectionCounter, helm_chart_version: Option, allowed_roles: AllowedRoles, + behind_trusted_proxy: bool, } #[async_trait] @@ -108,6 +113,7 @@ impl Server { active_connection_counter: config.active_connection_counter, helm_chart_version: config.helm_chart_version, allowed_roles: config.allowed_roles, + behind_trusted_proxy: config.behind_trusted_proxy, } } @@ -126,6 +132,7 @@ impl Server { let active_connection_counter = self.active_connection_counter.clone(); let helm_chart_version = self.helm_chart_version.clone(); let allowed_roles = self.allowed_roles; + let behind_trusted_proxy = self.behind_trusted_proxy; // TODO(guswynn): remove this redundant_closure_call #[allow(clippy::redundant_closure_call)] @@ -152,11 +159,30 @@ impl Server { version, mut params, }) => { - // If someone (usually the balancer) forwarded a connection UUID, + // These parameters describe the client as the + // proxy in front of this listener saw it, so + // they are only believable when such a proxy + // exists. Remove them either way so they never + // reach session var initialization. + let (forwarded_conn_uuid, forwarded_for) = { + let conn_uuid = params.remove(CONN_UUID_KEY); + let forwarded_for = params.remove(MZ_FORWARDED_FOR_KEY); + if behind_trusted_proxy { + (conn_uuid, forwarded_for) + } else { + if conn_uuid.is_some() || forwarded_for.is_some() { + warn!( + "ignoring forwarded connection metadata: listener is not behind a trusted proxy", + ); + } + (None, None) + } + }; + + // If the balancer forwarded a connection UUID, // then use that, otherwise generate one. let conn_uuid_handle = conn.inner_mut().uuid_handle(); - let conn_uuid = params - .remove(CONN_UUID_KEY) + let conn_uuid = forwarded_conn_uuid .and_then(|uuid| { uuid.parse() .inspect_err(|e| { @@ -187,7 +213,7 @@ impl Server { .peer_addr() .context("fetching peer addr")? .ip(); - let peer_addr= match params.remove(MZ_FORWARDED_FOR_KEY) { + let peer_addr= match forwarded_for { Some(ip_str) => { match IpAddr::from_str(&ip_str) { Ok(ip) => Some(ip), diff --git a/src/server-core/src/listeners.rs b/src/server-core/src/listeners.rs index 2bfa6b7de22f6..beff16cdeba00 100644 --- a/src/server-core/src/listeners.rs +++ b/src/server-core/src/listeners.rs @@ -165,6 +165,16 @@ pub struct BaseListenerConfig { pub authenticator_kind: AuthenticatorKind, pub allowed_roles: AllowedRoles, pub enable_tls: bool, + /// Whether a trusted proxy (`balancerd`) fronts this listener. + /// + /// Client-supplied connection metadata (the `mz_forwarded_for` and + /// `mz_connection_uuid` startup parameters on pgwire, the PROXY protocol + /// header on HTTP) describes the client as `balancerd` saw it, and is + /// honored only when this is set. `balancerd` rejects clients that supply + /// it themselves, so nothing else can choose the address that network + /// policies are evaluated against. + #[serde(default)] + pub behind_trusted_proxy: bool, } pub type SqlListenerConfig = BaseListenerConfig; @@ -176,6 +186,16 @@ pub struct HttpListenerConfig { pub authenticator_kind: AuthenticatorKind, pub enable_tls: bool, pub routes: HttpRoutesEnabled, + /// Whether a trusted proxy (`balancerd`) fronts this listener. + /// + /// Client-supplied connection metadata (the `mz_forwarded_for` and + /// `mz_connection_uuid` startup parameters on pgwire, the PROXY protocol + /// header on HTTP) describes the client as `balancerd` saw it, and is + /// honored only when this is set. `balancerd` rejects clients that supply + /// it themselves, so nothing else can choose the address that network + /// policies are evaluated against. + #[serde(default)] + pub behind_trusted_proxy: bool, } pub trait ListenerConfig { @@ -310,6 +330,7 @@ impl From for HttpListenerConfig { addr: legacy.base.addr, authenticator_kind: legacy.base.authenticator_kind, enable_tls: legacy.base.enable_tls, + behind_trusted_proxy: legacy.base.behind_trusted_proxy, routes: HttpRoutesEnabled { base: group(legacy.routes.base), webhook: group(legacy.routes.webhook), @@ -349,6 +370,7 @@ mod tests { authenticator_kind: AuthenticatorKind::None, allowed_roles: AllowedRoles::NormalAndInternal, enable_tls: false, + behind_trusted_proxy: false, }, routes: v0_147_0::HttpRoutes { base: true, @@ -401,6 +423,8 @@ mod tests { } }"#; let config = parse(json); + // A config without the field must keep the safe default. + assert!(!config.http["external"].behind_trusted_proxy); let routes = config.http["external"].routes; assert_eq!(routes.base, RouteGroup::Enabled(AllowedRoles::Normal)); assert_eq!(routes.internal, RouteGroup::Enabled(AllowedRoles::Internal)); @@ -489,6 +513,7 @@ mod tests { authenticator_kind: AuthenticatorKind::None, allowed_roles: AllowedRoles::NormalAndInternal, enable_tls: false, + behind_trusted_proxy: false, }, routes: v0_147_0::HttpRoutes { base: true, diff --git a/src/sqllogictest/src/runner.rs b/src/sqllogictest/src/runner.rs index f1f7648eb8612..e2ef145e576f5 100644 --- a/src/sqllogictest/src/runner.rs +++ b/src/sqllogictest/src/runner.rs @@ -1243,18 +1243,21 @@ impl<'a> RunnerInner<'a> { authenticator_kind: AuthenticatorKind::None, allowed_roles: AllowedRoles::Normal, enable_tls: false, + behind_trusted_proxy: false, }, "internal".to_owned() => SqlListenerConfig { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, allowed_roles: AllowedRoles::Internal, enable_tls: false, + behind_trusted_proxy: false, }, "password".to_owned() => SqlListenerConfig { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::Password, allowed_roles: AllowedRoles::Normal, enable_tls: false, + behind_trusted_proxy: false, }, }, http: btreemap![ @@ -1262,6 +1265,7 @@ impl<'a> RunnerInner<'a> { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, enable_tls: false, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Enabled(AllowedRoles::Normal), webhook: RouteGroup::Enabled(AllowedRoles::Normal), @@ -1277,6 +1281,7 @@ impl<'a> RunnerInner<'a> { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0), authenticator_kind: AuthenticatorKind::None, enable_tls: false, + behind_trusted_proxy: false, routes: HttpRoutesEnabled { base: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), webhook: RouteGroup::Enabled(AllowedRoles::NormalAndInternal), diff --git a/test/balancerd/mzcompose.py b/test/balancerd/mzcompose.py index bbcdb948e4152..b4372971800ee 100644 --- a/test/balancerd/mzcompose.py +++ b/test/balancerd/mzcompose.py @@ -417,6 +417,46 @@ def create_proxy_protocol_v2_header( ) +def sql_over_proxy_protocol( + port: int, query: str, header_fragments: list[bytes], fragment_delay: float = 0.0 +) -> Any: + """Run `query` against environmentd's external HTTP port over a raw socket. + + The PROXY protocol header in `header_fragments` precedes the TLS handshake, + exactly as balancerd sends it. Sending it in more than one fragment + exercises the server's handling of a header split across TCP segments. + """ + json_data = json.dumps({"query": query}) + request = dedent(f"""\ + POST /api/sql HTTP/1.1\r + Host: 127.0.0.1:{port}\r + Authorization: Basic {OTHER_USER}:{app_password(OTHER_USER)}\r + Content-Type: application/json\r + Content-Length: {len(json_data.encode())}\r + \r + {json_data}""").encode() + + tls = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + tls.check_hostname = False + tls.verify_mode = ssl.CERT_NONE + + with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock: + sock.connect(("127.0.0.1", port)) + sock.settimeout(30) + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + for i, fragment in enumerate(header_fragments): + if i > 0: + time.sleep(fragment_delay) + sock.sendall(fragment) + with tls.wrap_socket(sock) as tls_sock: + tls_sock.sendall(request) + response = tls_sock.recv(8192) + + _, separator, body = response.partition(b"\r\n\r\n") + assert separator, f"expected response with header and body, found: {response!r}" + return json.loads(body) + + def workflow_ip_forwarding(c: Composition) -> None: """Test that forwarding the client IP through the balancer works over both HTTP and SQL.""" c.up("balancerd", "frontegg-mock", "materialized") @@ -426,8 +466,9 @@ def workflow_ip_forwarding(c: Composition) -> None: # and that we can use proxy_protocol when talking to # envd directly. balancer_port = c.port("balancerd", 6876) - # mz internal (unencrypted port) - materialize_port = c.port("materialized", 6878) + # The external port, the only one balancerd fronts and therefore the only + # one that honors a PROXY protocol header. + materialize_port = c.port("materialized", 6876) # We want to make sure the request we're making through the balancer does not use the balancers # ip for the sessions. @@ -469,41 +510,15 @@ def workflow_ip_forwarding(c: Composition) -> None: session_ip != balancer_ip ), f"requests from ({session_ip}) proxied by balancer should not use balancer ip ({balancer_ip}) in session" - with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock: - sock.connect(("127.0.0.1", materialize_port)) - - # Pick an ip we couldn't normal connect from and trick envd into - # thinking we're connecting with - proxy_header = create_proxy_protocol_v2_header( - "1.1.1.1", 1111, "127.0.0.1", 1111 - ) - # Make an http request over the socket - - json_data = { - "query": "select client_ip from mz_internal.mz_sessions where connection_id = pg_backend_pid();" - } - json_data = json.dumps(json_data) - content_length = len(json_data.encode()) - http_sql_query_request = dedent(f"""\ - POST /api/sql HTTP/1.1\r - Host: 127.0.0.1:{materialize_port}\r - Authorization: Basic {OTHER_USER}:{app_password(OTHER_USER)}\r - Content-Type: application/json\r - Content-Length: {content_length}\r - \r - {json_data}""") - sock.sendall(proxy_header + http_sql_query_request.encode()) - - # read and parse the response - body_separator = "\r\n\r\n" - tcp_resp = sock.recv(8192) - resp_split = tcp_resp.split(body_separator.encode()) - assert ( - len(resp_split) > 1 - ), f"expected response with header and body, found: {resp_split}" - body = resp_split[1] - # assert that we tricked environmentd - assert json.loads(body)["results"][0]["rows"][0][0] == "1.1.1.1" + # Pick an ip we couldn't normally connect from and trick envd into + # thinking we're connecting with it. + result = sql_over_proxy_protocol( + materialize_port, + "select client_ip from mz_internal.mz_sessions where connection_id = pg_backend_pid();", + [create_proxy_protocol_v2_header("1.1.1.1", 1111, "127.0.0.1", 1111)], + ) + # assert that we tricked environmentd + assert result["results"][0]["rows"][0][0] == "1.1.1.1" def workflow_wide_result(c: Composition) -> None: @@ -900,43 +915,17 @@ def workflow_split_proxy_header(c: Composition) -> None: bytes remain in the stream and corrupt the subsequent HTTP parsing. """ c.up("balancerd", "frontegg-mock", "materialized") - materialize_port = c.port("materialized", 6878) + # Split the 28-byte proxy header at byte 8, in the middle of the 12-byte + # signature. proxy_hdr = create_proxy_protocol_v2_header("2.2.2.2", 2222, "127.0.0.1", 2222) - json_data = json.dumps({"query": "SELECT 42 AS answer"}) - content_length = len(json_data.encode()) - http_request = dedent(f"""\ - POST /api/sql HTTP/1.1\r - Host: 127.0.0.1:{materialize_port}\r - Authorization: Basic {OTHER_USER}:{app_password(OTHER_USER)}\r - Content-Type: application/json\r - Content-Length: {content_length}\r - \r - {json_data}""").encode() - - # Split the 28-byte proxy header at byte 8 (middle of the 12-byte - # signature). Send the first fragment, wait for the server to peek - # it, then send the rest along with the HTTP request. - split_point = 8 - with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock: - sock.connect(("127.0.0.1", materialize_port)) - sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) - - sock.sendall(proxy_hdr[:split_point]) - time.sleep(0.2) - sock.sendall(proxy_hdr[split_point:] + http_request) - - sock.settimeout(10) - tcp_resp = sock.recv(8192) - body_separator = b"\r\n\r\n" - resp_split = tcp_resp.split(body_separator) - assert ( - len(resp_split) > 1 - ), f"expected response with header and body, found: {resp_split}" - body = resp_split[1] - assert ( - json.loads(body)["results"][0]["rows"][0][0] == "42" - ), f"unexpected response body: {body}" + result = sql_over_proxy_protocol( + c.port("materialized", 6876), + "SELECT 42 AS answer", + [proxy_hdr[:8], proxy_hdr[8:]], + fragment_delay=0.2, + ) + assert result["results"][0]["rows"][0][0] == "42", f"unexpected response: {result}" def _frontegg_curl(