add CI

Author: isabelatkinson

.evergreen/config.yml

@@ -399,6 +399,14 @@ buildvariants:
     tasks:
       - happy-eyeballs-task-group
 
+  - name: socks5-proxy-no-tls
+    display_name: "SOCKS5 Proxy (No TLS)"
+    patchable: false
+    run_on:
+      - rhel87-small
+    tasks:
+      - test-socks5-proxy
+
   #- name: graviton-legacy
   #  display_name: "Graviton (legacy versions)"
   #  run_on:
@@ -1246,6 +1254,24 @@ tasks:
         include_expansions_in_env:
           - PROJECT_DIRECTORY
 
+  - name: test-socks5-proxy
+    commands:
+    - func: bootstrap mongo-orchestration
+      vars:
+        MONGODB_VERSION: rapid
+        TOPOLOGY: replica-set
+    - command: subprocess.exec
+      type: test
+      params:
+        working_dir: src
+        binary: bash
+        args:
+          - .evergreen/run-socks5-proxy-test.sh
+        include_expansions_in_env:
+          - PROJECT_DIRECTORY
+          - MONGODB_URI
+          - DRIVERS_TOOLS
+
 #############
 # Functions #
 #############

.evergreen/run-socks5-proxy-test.sh

@@ -0,0 +1,27 @@
+#!/bin/bash
+
+set -o errexit
+
+source .evergreen/env.sh
+source .evergreen/cargo-test.sh
+
+FEATURE_FLAGS+=("socks5-proxy")
+CARGO_OPTIONS+=("--ignore-default-filter")
+
+# with auth
+python $DRIVERS_TOOLS/.evergreen/socks5srv.py --map "127.0.0.1:12345 to 127.0.0.1:27017" --port 1080 --auth username:p4ssw0rd &
+AUTH_PID=$!
+# without auth
+python $DRIVERS_TOOLS/.evergreen/socks5srv.py --map "127.0.0.1:12345 to 127.0.0.1:27017" --port 1081 &
+NOAUTH_PID=$!
+
+echo "testing socks5proxy with URI $MONGODB_URI"
+
+set +o errexit
+
+cargo_test socks5_proxy
+
+kill $AUTH_PID
+kill $NOAUTH_PID
+
+exit ${CARGO_RESULT}

src/client/csfle/state_machine.rs

@@ -188,8 +188,15 @@ impl CryptExecutor {
                             .and_then(|tls| tls.get(&provider))
                             .cloned()
                             .unwrap_or_default();
-                        let mut stream =
-                            AsyncStream::connect(addr, Some(&TlsConfig::new(tls_options)?)).await?;
+                        // isabeltodo propagate proxy, cache config?
+                        let mut stream = AsyncStream::connect(
+                            addr,
+                            Some(&TlsConfig::new(tls_options)?),
+                            Duration::MAX,
+                            #[cfg(feature = "socks5-proxy")]
+                            None,
+                        )
+                        .await?;
                         stream.write_all(kms_ctx.message()?).await?;
                         let mut buf = vec![0];
                         while kms_ctx.bytes_needed() > 0 {

src/client/options.rs

@@ -59,6 +59,10 @@ const TLS_INSECURE: &str = "tlsinsecure";
 const TLS_ALLOW_INVALID_CERTIFICATES: &str = "tlsallowinvalidcertificates";
 #[cfg(feature = "openssl-tls")]
 const TLS_ALLOW_INVALID_HOSTNAMES: &str = "tlsallowinvalidhostnames";
+const PROXY_HOST: &str = "proxyhost";
+const PROXY_PORT: &str = "proxyport";
+const PROXY_USERNAME: &str = "proxyusername";
+const PROXY_PASSWORD: &str = "proxypassword";
 const URI_OPTIONS: &[&str] = &[
     "appname",
     "authmechanism",
@@ -75,10 +79,10 @@ const URI_OPTIONS: &[&str] = &[
     "maxpoolsize",
     "minpoolsize",
     "maxconnecting",
-    "proxyHost",
-    "proxyPort",
-    "proxyUsername",
-    "proxyPassword",
+    PROXY_HOST,
+    PROXY_PORT,
+    PROXY_USERNAME,
+    PROXY_PASSWORD,
     "readconcernlevel",
     "readpreference",
     "readpreferencetags",
@@ -993,6 +997,14 @@ pub struct ConnectionString {
     /// Overrides the default "mongodb" service name for SRV lookup in both discovery and polling
     pub srv_service_name: Option<String>,
 
+    pub proxy_host: Option<String>,
+
+    pub proxy_port: Option<u16>,
+
+    pub proxy_username: Option<String>,
+
+    pub proxy_password: Option<String>,
+
     #[serde(serialize_with = "serde_util::serialize_duration_option_as_int_millis")]
     wait_queue_timeout: Option<Duration>,
     tls_insecure: Option<bool>,
@@ -1333,6 +1345,42 @@ impl ClientOptions {
             }
         }
 
+        #[cfg(feature = "socks5-proxy")]
+        {
+            if self.proxy_host.is_some() {
+                if self
+                    .hosts
+                    .iter()
+                    .any(|address| !matches!(address, ServerAddress::Tcp { .. }))
+                {
+                    return Err(Error::invalid_argument(
+                        "cannot specify a non-TCP address when connected to a proxy host",
+                    ));
+                }
+
+                if self.proxy_username.is_some() != self.proxy_password.is_some() {
+                    return Err(Error::invalid_argument(
+                        "proxy username and proxy password must both be specified or unset",
+                    ));
+                }
+            } else {
+                let error = |option: &str| {
+                    Error::invalid_argument(format!(
+                        "cannot specify a proxy {option} without a proxy host"
+                    ))
+                };
+                if self.proxy_port.is_some() {
+                    return Err(error("port"));
+                }
+                if self.proxy_username.is_some() {
+                    return Err(error("username"));
+                }
+                if self.proxy_password.is_some() {
+                    return Err(error("password"));
+                }
+            }
+        }
+
         Ok(())
     }
 
@@ -1720,6 +1768,10 @@ impl ConnectionString {
             srv_service_name: _,
             wait_queue_timeout,
             tls_insecure,
+            proxy_host,
+            proxy_port,
+            proxy_username,
+            proxy_password,
             #[cfg(test)]
                 original_uri: _,
         } = self;
@@ -2021,6 +2073,22 @@ impl ConnectionString {
             opts.push_str(&format!("&srvMaxHosts={srv_max_hosts}"));
         }
 
+        if let Some(proxy_host) = proxy_host {
+            opts.push_str(&format!("&proxyHost={proxy_host}"));
+        }
+
+        if let Some(proxy_port) = proxy_port {
+            opts.push_str(&format!("&proxyPort={proxy_port}"));
+        }
+
+        if let Some(proxy_username) = proxy_username {
+            opts.push_str(&format!("&proxyUsername={proxy_username}"));
+        }
+
+        if let Some(proxy_password) = proxy_password {
+            opts.push_str(&format!("&proxyPassword={proxy_password}"));
+        }
+
         if !opts.is_empty() {
             opts.replace_range(0..1, "?"); // mark start of options
             res.push_str(&opts);
@@ -2599,6 +2667,24 @@ impl ConnectionString {
 
                 parts.zlib_compression = Some(i);
             }
+            #[cfg(feature = "socks5-proxy")]
+            PROXY_HOST => self.proxy_host = Some(value.to_string()),
+            #[cfg(feature = "socks5-proxy")]
+            PROXY_PORT => {
+                let port = u16::from_str(value)
+                    .map_err(|_| Error::invalid_argument(format!("invalid proxy port: {value}")))?;
+                self.proxy_port = Some(port);
+            }
+            #[cfg(feature = "socks5-proxy")]
+            PROXY_USERNAME => self.proxy_username = Some(value.to_string()),
+            #[cfg(feature = "socks5-proxy")]
+            PROXY_PASSWORD => self.proxy_password = Some(value.to_string()),
+            #[cfg(not(feature = "socks5-proxy"))]
+            PROXY_HOST | PROXY_PORT | PROXY_USERNAME | PROXY_PASSWORD => {
+                return Err(Error::invalid_argument(format!(
+                    "cannot specify {key} if socks5-proxy feature is not enabled"
+                )));
+            }
 
             other => {
                 let (jaro_winkler, option) = URI_OPTIONS.iter().fold((0.0, ""), |acc, option| {

src/client/options/parse.rs

@@ -164,15 +164,14 @@ impl ClientOptions {
             srv_service_name: conn_str.srv_service_name,
             #[cfg(feature = "opentelemetry")]
             tracing: None,
-            // isabeltodo propagate from connstr
             #[cfg(feature = "socks5-proxy")]
-            proxy_host: None,
+            proxy_host: conn_str.proxy_host,
             #[cfg(feature = "socks5-proxy")]
-            proxy_port: None,
+            proxy_port: conn_str.proxy_port,
             #[cfg(feature = "socks5-proxy")]
-            proxy_username: None,
+            proxy_username: conn_str.proxy_username,
             #[cfg(feature = "socks5-proxy")]
-            proxy_password: None,
+            proxy_password: conn_str.proxy_password,
         }
     }
 }

src/cmap/establish.rs

@@ -22,7 +22,11 @@ use crate::{
     },
     error::{Error as MongoError, ErrorKind, Result},
     hello::HelloReply,
-    runtime::{self, stream::DEFAULT_CONNECT_TIMEOUT, AsyncStream, TlsConfig},
+    runtime::{
+        stream::{Proxy, DEFAULT_CONNECT_TIMEOUT},
+        AsyncStream,
+        TlsConfig,
+    },
     sdam::{topology::TopologySpec, HandshakePhase},
 };
 
@@ -55,62 +59,6 @@ pub(crate) struct EstablisherOptions {
     proxy: Option<Proxy>,
 }
 
-#[derive(Clone)]
-#[cfg(feature = "socks5-proxy")]
-struct Proxy {
-    host: String,
-    port: u16,
-    authentication: Option<(String, String)>,
-}
-
-#[cfg(feature = "socks5-proxy")]
-impl Proxy {
-    fn from_client_options(options: &crate::options::ClientOptions) -> Option<Self> {
-        let host = options.proxy_host.as_ref()?.clone();
-        let port = options.proxy_port.unwrap_or(1080);
-        let authentication = match (&options.proxy_username, &options.proxy_password) {
-            (Some(username), Some(password)) => Some((username.clone(), password.clone())),
-            // ClientOptions::validate will return an error if the username and password are not
-            // provided together or both omitted.
-            _ => None,
-        };
-        Some(Self {
-            host,
-            port,
-            authentication,
-        })
-    }
-
-    async fn connect(
-        &self,
-        address: ServerAddress,
-        connect_timeout: Duration,
-    ) -> Result<AsyncStream> {
-        use fast_socks5::client::{Config, Socks5Stream};
-
-        let mut config = Config::default();
-        config.set_connect_timeout(connect_timeout.as_secs());
-
-        let stream = if let Some((username, password)) = self.authentication.as_ref() {
-            Socks5Stream::connect_with_password(
-                address.to_string(),
-                self.host.clone(),
-                self.port,
-                username.clone(),
-                password.clone(),
-                config,
-            )
-            .await
-        } else {
-            Socks5Stream::connect(address.to_string(), self.host.clone(), self.port, config).await
-        }
-        .map_err(|error| ErrorKind::ProxyConnect {
-            message: error.to_string(),
-        })?;
-        Ok(AsyncStream::Socks5(stream))
-    }
-}
-
 impl From<&TopologySpec> for EstablisherOptions {
     fn from(spec: &TopologySpec) -> Self {
         Self {
@@ -155,15 +103,20 @@ impl ConnectionEstablisher {
             connect_timeout,
             #[cfg(test)]
             test_patch_reply: options.test_patch_reply,
+            #[cfg(feature = "socks5-proxy")]
+            proxy: options.proxy,
         })
     }
 
     async fn make_stream(&self, address: ServerAddress) -> Result<AsyncStream> {
-        runtime::timeout(
+        AsyncStream::connect(
+            address,
+            self.tls_config.as_ref(),
             self.connect_timeout,
-            AsyncStream::connect(address, self.tls_config.as_ref()),
+            #[cfg(feature = "socks5-proxy")]
+            self.proxy.as_ref(),
         )
-        .await?
+        .await
     }
 
     /// Establishes a connection.
@@ -176,8 +129,6 @@ impl ConnectionEstablisher {
         let address = pending_connection.address.clone();
         let cancellation_receiver = pending_connection.cancellation_receiver.take();
 
-        if let Some(ref proxy) = self.proxy {}
-
         let stream = self
             .make_stream(address)
             .await