storage: don't resolve remote hosts when using ssh tunnels

The `openssh` crate has a bug in which it attempts to resolve the target
host, which my only be resolvable on the SSH bastion host, on the
client.

Upgrade to a patched version of `openssh` that does not have this bug,
and add a test for this scenario (that would fail without the patched
version of `openssh`).

Author: guswynn

Cargo.lock

@@ -172,3 +172,7 @@ proptest-derive = { git = "https://github.com/MaterializeInc/proptest.git" }
 # Waiting on https://github.com/edenhill/librdkafka/pull/4051.
 rdkafka = { git = "https://github.com/MaterializeInc/rust-rdkafka.git" }
 rdkafka-sys = { git = "https://github.com/MaterializeInc/rust-rdkafka.git" }
+
+# Waiting on https://github.com/openssh-rust/openssh/pull/120 to make it into
+# a release.
+openssh = { git = "https://github.com/MaterializeInc/openssh.git" }

Cargo.toml

@@ -10,7 +10,7 @@ rust-version.workspace = true
 anyhow = "1.0.66"
 axum = { version = "0.6.7" }
 clap = { version = "3.2.24", features = [ "derive" ] }
-dirs = "4.0.0"
+dirs = "5.0.0"
 indicatif = "0.17.2"
 mz-build-info = { path = "../build-info" }
 mz-ore = { path = "../ore", features = ["async", "cli"] }

src/mz/Cargo.toml

@@ -199,8 +199,8 @@ async fn connect(
         // Force use of IPv4 loopback. Do not use the hostname `localhost`,
         // as that can resolve to IPv6, and the SSH tunnel is only listening
         // for IPv4 connections.
-        let local = openssh::Socket::new(&(Ipv4Addr::LOCALHOST, local_port))?;
-        let remote = openssh::Socket::new(&(host, port))?;
+        let local = openssh::Socket::from((Ipv4Addr::LOCALHOST, local_port));
+        let remote = openssh::Socket::new(host, port);
 
         match session
             .request_port_forward(ForwardType::Local, local, remote)

src/ssh-util/src/tunnel.rs

@@ -0,0 +1,54 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+# Test creating a source that uses Kafka/CSR via an SSH tunnel where the Kafka
+# broker and CSR server have hostnames that can only be resolved from the
+# SSH bastion host.
+#
+# The `openssh` crate previously had a bug where it would attempt to resolve
+# the target host on the client.
+# See: https://github.com/openssh-rust/openssh/pull/120
+
+$ kafka-create-topic topic=thetopic
+
+$ kafka-ingest topic=thetopic format=bytes
+one
+
+> CREATE CONNECTION kafka_conn
+  TO KAFKA (BROKER 'hidden-kafka:9092' USING SSH TUNNEL thancred);
+
+
+> CREATE CONNECTION csr_conn TO CONFLUENT SCHEMA REGISTRY (
+    URL 'http://hidden-schema-registry:8081',
+    SSH TUNNEL thancred
+  );
+
+$ set schema={
+        "type" : "record",
+        "name" : "test",
+        "fields" : [
+            {"name":"f1", "type":"string"},
+            {"name":"f2", "type":"long"}
+        ]
+    }
+
+$ kafka-create-topic topic=avroavro
+
+$ kafka-ingest format=avro topic=avroavro schema=${schema}
+{"f1": "fish", "f2": 1000}
+
+> CREATE SOURCE csr_source
+  FROM KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-avroavro-${testdrive.seed}')
+  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn
+  ENVELOPE NONE
+
+> SELECT * FROM csr_source
+f1    f2
+----------
+fish  1000

test/ssh-connection/hidden-hosts.td

@@ -115,6 +115,45 @@ def workflow_kafka_csr_via_ssh_tunnel(c: Composition, redpanda: bool = False) ->
         c.run("testdrive", "--no-reset", "kafka-source-after-ssh-restart.td")
 
 
+def workflow_hidden_hosts(c: Composition, redpanda: bool = False) -> None:
+    c.down()
+    dependencies = ["materialized", "ssh-bastion-host"]
+    if redpanda:
+        dependencies += ["redpanda"]
+    else:
+        dependencies += ["zookeeper", "kafka", "schema-registry"]
+    c.up(*dependencies)
+
+    c.run("testdrive", "setup.td")
+
+    public_key = c.sql_query("select public_key_1 from mz_ssh_tunnel_connections;")[0][
+        0
+    ]
+
+    c.exec(
+        "ssh-bastion-host",
+        "bash",
+        "-c",
+        f"echo '{public_key}' > /etc/authorized_keys/mz",
+    )
+
+    def add_hidden_host(container: str) -> None:
+        ip = c.exec(
+            "ssh-bastion-host", "getent", "hosts", container, capture=True
+        ).stdout.split(" ")[0]
+        c.exec(
+            "ssh-bastion-host",
+            "bash",
+            "-c",
+            f"echo '{ip} hidden-{container}' >> /etc/hosts",
+        )
+
+    add_hidden_host("kafka")
+    add_hidden_host("schema-registry")
+
+    c.run("testdrive", "--no-reset", "hidden-hosts.td")
+
+
 # Test that if we restart the bastion AND change its server keys(s), we can
 # still reconnect in the replication stream.
 def workflow_pg_restart_bastion(c: Composition) -> None:
@@ -268,3 +307,4 @@ def workflow_default(c: Composition) -> None:
     workflow_pg_via_ssh_tunnel(c)
     workflow_pg_via_ssh_tunnel_with_ssl(c)
     workflow_pg_restart_bastion(c)
+    workflow_hidden_hosts(c)