Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jubrad wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 25 additions & 8 deletions src/environmentd/src/http.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)]
Expand Down Expand Up @@ -221,6 +224,7 @@ struct HelmChartVersion(Option<String>);
pub struct HttpServer {
tls: Option<ReloadingSslContext>,
router: Router,
behind_trusted_proxy: bool,
}

impl HttpServer {
Expand All @@ -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();
Expand Down Expand Up @@ -679,7 +684,11 @@ impl HttpServer {
.merge(base_router)
.apply_default_layers(source, metrics);

HttpServer { tls, router }
HttpServer {
tls,
router,
behind_trusted_proxy,
}
}
}

Expand All @@ -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) => {
Expand Down
2 changes: 2 additions & 0 deletions src/environmentd/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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);
}
Expand Down
30 changes: 30 additions & 0 deletions src/environmentd/src/test_util.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand All @@ -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),
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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),
Expand All @@ -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),
Expand Down Expand Up @@ -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),
Expand All @@ -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),
Expand Down Expand Up @@ -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),
Expand All @@ -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,
Expand All @@ -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),
Expand All @@ -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,
Expand Down
80 changes: 80 additions & 0 deletions src/environmentd/tests/pgwire.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
);
}
60 changes: 60 additions & 0 deletions src/environmentd/tests/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}"
);
}
Loading
Loading