SQL-204: only trust forwarded client IPs behind a proxy

A client reaching environmentd directly could set the `mz_forwarded_for`
pgwire startup parameter, or prepend a PROXY protocol v2 header on HTTP,
and have that address recorded as the session's client IP. Network
policies are evaluated against that address, so any client could claim to
connect from an allowed one.

Add a per-listener `behind_trusted_proxy` flag, defaulting to false, and
honor `mz_forwarded_for`, `mz_connection_uuid`, and the PROXY protocol
header only where it is set. orchestratord sets it on the external
listeners, which balancerd fronts, and only when it also creates
balancerd.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: jubrad

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<ReplicaHttpLocator>,
+    /// 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<String>);
 pub struct HttpServer {
     tls: Option<ReloadingSslContext>,
     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) => {

src/environmentd/src/lib.rs

@@ -303,6 +303,7 @@ impl Listener<SqlListenerConfig> {
                 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);
         }

src/environmentd/src/test_util.rs

@@ -156,19 +156,22 @@ 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![
                     "external".to_owned() => HttpListenerConfig {
                         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,19 +355,22 @@ 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! {
                 "external".to_owned() => HttpListenerConfig {
                     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,19 +418,22 @@ 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! {
                 "external".to_owned() => HttpListenerConfig {
                     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,13 +502,15 @@ impl TestHarness {
                     authenticator_kind: AuthenticatorKind::Password,
                     allowed_roles: AllowedRoles::NormalAndInternal,
                     enable_tls,
+                    behind_trusted_proxy: false,
                 },
             },
             http: btreemap! {
                 "external".to_owned() => HttpListenerConfig {
                     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,13 +553,15 @@ impl TestHarness {
                     authenticator_kind: AuthenticatorKind::Sasl,
                     allowed_roles: AllowedRoles::NormalAndInternal,
                     enable_tls,
+                    behind_trusted_proxy: false,
                 },
             },
             http: btreemap! {
                 "external".to_owned() => HttpListenerConfig {
                     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,

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::<Vec<_>>()
+                    .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",
+    );
+}

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<u8> {
+        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}"
+    );
+}