storage: Deny localhost/private-ips in storage connections

Cargo.toml

@@ -189,6 +189,9 @@ postgres-types = { git = "https://github.com/MaterializeInc/rust-postgres" }
 postgres-openssl = { git = "https://github.com/MaterializeInc/rust-postgres" }
 postgres_array = { git = "https://github.com/MaterializeInc/rust-postgres-array" }
 
+# Waiting on https://github.com/blackbeam/mysql_async/pull/300
+mysql_async = { git = "https://github.com/MaterializeInc/mysql_async", rev = "6afd2136181f2ec93b7ca7de524f6d02b6f10c1d" }
+
 # Waiting on https://github.com/MaterializeInc/serde-value/pull/35.
 serde-value = { git = "https://github.com/MaterializeInc/serde-value.git" }
 

misc/cargo-vet/audits.toml

@@ -187,6 +187,26 @@ who = "Gus Wynn <gus@materialize.com>"
 criteria = "maintained-and-necessary"
 version = "4.2.0"
 
+[[audits.postgres]]
+who = "Roshan Jobanputra <roshan@materialize.com>"
+criteria = "safe-to-deploy"
+version = "0.19.5@git:91522e47643ebb6d6a5e392957b2319e5bb522ad"
+
+[[audits.postgres-openssl]]
+who = "Roshan Jobanputra <roshan@materialize.com>"
+criteria = "safe-to-deploy"
+version = "0.5.0@git:91522e47643ebb6d6a5e392957b2319e5bb522ad"
+
+[[audits.postgres-protocol]]
+who = "Roshan Jobanputra <roshan@materialize.com>"
+criteria = "safe-to-deploy"
+version = "0.6.5@git:91522e47643ebb6d6a5e392957b2319e5bb522ad"
+
+[[audits.postgres-types]]
+who = "Roshan Jobanputra <roshan@materialize.com>"
+criteria = "safe-to-deploy"
+version = "0.2.5@git:91522e47643ebb6d6a5e392957b2319e5bb522ad"
+
 [[audits.quanta]]
 who = "Roshan Jobanputra <roshan@materialize.com>"
 criteria = "safe-to-deploy"
@@ -292,6 +312,11 @@ who = "Moritz Hoffmann <mh@materialize.com>"
 criteria = "safe-to-deploy"
 version = "0.12.0@git:46b28dc48bf5ed26dc0a0d5dbbd53c7964526f27"
 
+[[audits.tokio-postgres]]
+who = "Roshan Jobanputra <roshan@materialize.com>"
+criteria = "safe-to-deploy"
+version = "0.7.8@git:91522e47643ebb6d6a5e392957b2319e5bb522ad"
+
 [[audits.tracing-capture]]
 who = "Matt Jibson <matt.jibson@gmail.com>"
 criteria = "safe-to-deploy"

misc/cargo-vet/config.toml

@@ -1010,22 +1010,6 @@ criteria = "safe-to-deploy"
 version = "0.3.15"
 criteria = "safe-to-deploy"
 
-[[exemptions.postgres]]
-version = "0.19.5@git:b759caa33610403aa74b1cfdd37f45eb3100c9af"
-criteria = "safe-to-deploy"
-
-[[exemptions.postgres-openssl]]
-version = "0.5.0@git:b759caa33610403aa74b1cfdd37f45eb3100c9af"
-criteria = "safe-to-deploy"
-
-[[exemptions.postgres-protocol]]
-version = "0.6.5@git:b759caa33610403aa74b1cfdd37f45eb3100c9af"
-criteria = "safe-to-deploy"
-
-[[exemptions.postgres-types]]
-version = "0.2.5@git:b759caa33610403aa74b1cfdd37f45eb3100c9af"
-criteria = "safe-to-deploy"
-
 [[exemptions.postgres_array]]
 version = "0.11.0@git:f58d0101e5198e04e8692629018d9b58f8543534"
 criteria = "safe-to-deploy"
@@ -1514,10 +1498,6 @@ criteria = "safe-to-deploy"
 version = "0.2.12"
 criteria = "safe-to-deploy"
 
-[[exemptions.tokio-postgres]]
-version = "0.7.8@git:b759caa33610403aa74b1cfdd37f45eb3100c9af"
-criteria = "safe-to-deploy"
-
 [[exemptions.tokio-tungstenite]]
 version = "0.20.0"
 criteria = "safe-to-deploy"

src/mysql-util/src/tunnel.rs

@@ -7,6 +7,7 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use std::net::IpAddr;
 use std::ops::{Deref, DerefMut};
 use std::time::Duration;
 
@@ -27,7 +28,7 @@ use crate::MySqlError;
 #[derive(Debug, PartialEq, Clone)]
 pub enum TunnelConfig {
     /// Establish a direct TCP connection to the database host.
-    Direct,
+    Direct { resolved_ips: Option<Vec<IpAddr>> },
     /// Establish a TCP connection to the database via an SSH tunnel.
     /// This means first establishing an SSH connection to a bastion host,
     /// and then opening a separate connection from that host to the database.
@@ -237,12 +238,15 @@ impl Config {
         ssh_tunnel_manager: &SshTunnelManager,
     ) -> Result<MySqlConn, MySqlError> {
         match &self.tunnel {
-            TunnelConfig::Direct => Ok(MySqlConn {
-                conn: Conn::new(self.inner.clone())
-                    .await
-                    .map_err(MySqlError::from)?,
-                _ssh_tunnel_handle: None,
-            }),
+            TunnelConfig::Direct { resolved_ips } => {
+                let opts_builder =
+                    OptsBuilder::from_opts(self.inner.clone()).resolved_ips(resolved_ips.clone());
+
+                Ok(MySqlConn {
+                    conn: Conn::new(opts_builder).await.map_err(MySqlError::from)?,
+                    _ssh_tunnel_handle: None,
+                })
+            }
             TunnelConfig::Ssh { config } => {
                 let (host, port) = self.address();
                 let tunnel = ssh_tunnel_manager

src/ore/src/netio.rs

@@ -16,11 +16,13 @@
 //! Network I/O utilities.
 
 mod async_ready;
+mod dns;
 mod framed;
 mod read_exact;
 mod socket;
 
 pub use crate::netio::async_ready::AsyncReady;
+pub use crate::netio::dns::resolve_address;
 pub use crate::netio::framed::{FrameTooBig, MAX_FRAME_SIZE};
 pub use crate::netio::read_exact::{read_exact_or_eof, ReadExactOrEof};
 pub use crate::netio::socket::{Listener, SocketAddr, SocketAddrType, Stream, UnixSocketAddr};

src/ore/src/netio/dns.rs

@@ -0,0 +1,70 @@
+// Copyright Materialize, Inc. and contributors. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License in the LICENSE file at the
+// root of this repository, or online at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::io;
+use std::net::IpAddr;
+
+use tokio::net::lookup_host;
+
+const DUMMY_PORT: u16 = 11111;
+
+/// Resolves a host address and ensures it is a global address when `enforce_global` is set.
+/// This parameter is useful when connecting to user-defined unverified addresses.
+pub async fn resolve_address(
+    mut host: &str,
+    enforce_global: bool,
+) -> Result<Vec<IpAddr>, io::Error> {
+    // `net::lookup_host` requires a port to be specified, but we don't care about the port.
+    let mut port = DUMMY_PORT;
+    // If a port is already specified, use it and remove it from the host.
+    if let Some(idx) = host.find(':') {
+        if let Ok(p) = host[idx + 1..].parse() {
+            port = p;
+            host = &host[..idx];
+        }
+    }
+
+    let mut addrs = lookup_host((host, port)).await?;
+    let mut ips = Vec::new();
+    while let Some(addr) = addrs.next() {
+        let ip = addr.ip();
+        if enforce_global && !is_global(ip) {
+            Err(io::Error::new(
+                io::ErrorKind::AddrNotAvailable,
+                "address is not global",
+            ))?
+        } else {
+            ips.push(ip);
+        }
+    }
+
+    if ips.len() == 0 {
+        Err(io::Error::new(
+            io::ErrorKind::AddrNotAvailable,
+            "no addresses found",
+        ))?
+    }
+    Ok(ips)
+}
+
+fn is_global(addr: IpAddr) -> bool {
+    // TODO: Switch to `addr.is_global()` once stable: https://github.com/rust-lang/rust/issues/27709
+    match addr {
+        IpAddr::V4(ip) => {
+            !(ip.is_unspecified() || ip.is_private() || ip.is_loopback() || ip.is_link_local())
+        }
+        IpAddr::V6(ip) => !(ip.is_loopback() || ip.is_unspecified()),
+    }
+}

src/postgres-util/src/tunnel.rs

@@ -7,6 +7,7 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use std::net::IpAddr;
 use std::time::Duration;
 
 use mz_ore::future::{InTask, OreFutureExt};
@@ -38,7 +39,7 @@ macro_rules! bail_generic {
 #[derive(Debug, PartialEq, Clone)]
 pub enum TunnelConfig {
     /// Establish a direct TCP connection to the database host.
-    Direct,
+    Direct { resolved_ips: Option<Vec<IpAddr>> },
     /// Establish a TCP connection to the database via an SSH tunnel.
     /// This means first establishing an SSH connection to a bastion host,
     /// and then opening a separate connection from that host to the database.
@@ -226,7 +227,24 @@ impl Config {
         })?;
 
         match &self.tunnel {
-            TunnelConfig::Direct => {
+            TunnelConfig::Direct { resolved_ips } => {
+                if let Some(ips) = resolved_ips {
+                    let host = match postgres_config.get_hosts() {
+                        [Host::Tcp(host)] => host,
+                        _ => bail_generic!(
+                            "only TCP connections to a single PostgreSQL server are supported"
+                        ),
+                    }
+                    .to_owned();
+                    // The number of 'host' and 'hostaddr' values must be the same.
+                    for (idx, ip) in ips.iter().enumerate() {
+                        if idx != 0 {
+                            postgres_config.host(&host);
+                        }
+                        postgres_config.hostaddr(ip.clone());
+                    }
+                };
+
                 let (client, connection) = async move { postgres_config.connect(tls).await }
                     .run_in_task_if(self.in_task, || "pg_connect".to_string())
                     .await?;
@@ -277,17 +295,25 @@ impl Config {
                 // `tokio_postgres::Config` to do this is somewhat confusing, and requires we edit
                 // the singular host in place.
 
-                let (host, _) = self.address()?;
-                postgres_config.tls_verify_host(host);
-
                 let privatelink_host = mz_cloud_resources::vpc_endpoint_name(*connection_id);
+                let privatelink_addrs = tokio::net::lookup_host(privatelink_host).await?;
 
-                match postgres_config.get_hosts_mut() {
-                    [Host::Tcp(host)] => *host = privatelink_host,
+                // Override the actual IPs to connect to for the TCP connection, leaving the original host in-place
+                // for TLS verification
+                let host = match postgres_config.get_hosts() {
+                    [Host::Tcp(host)] => host,
                     _ => bail_generic!(
                         "only TCP connections to a single PostgreSQL server are supported"
                     ),
                 }
+                .to_owned();
+                // The number of 'host' and 'hostaddr' values must be the same.
+                for (idx, addr) in privatelink_addrs.enumerate() {
+                    if idx != 0 {
+                        postgres_config.host(&host);
+                    }
+                    postgres_config.hostaddr(addr.ip());
+                }
 
                 let (client, connection) = async move { postgres_config.connect(tls).await }
                     .run_in_task_if(self.in_task, || "pg_connect".to_string())