RUST-577 Versioned API implementation (#266)

Author: alcaeus

src/client/auth/aws.rs

@@ -5,11 +5,14 @@ use sha2::{Digest, Sha256};
 
 use crate::{
     bson::{doc, spec::BinarySubtype, Binary, Bson, Document},
-    client::auth::{
-        self,
-        sasl::{SaslContinue, SaslResponse, SaslStart},
-        AuthMechanism,
-        Credential,
+    client::{
+        auth::{
+            self,
+            sasl::{SaslContinue, SaslResponse, SaslStart},
+            AuthMechanism,
+            Credential,
+        },
+        options::ServerApi,
     },
     cmap::Connection,
     error::{Error, Result},
@@ -24,6 +27,7 @@ const AWS_LONG_DATE_FMT: &str = "%Y%m%dT%H%M%SZ";
 pub(super) async fn authenticate_stream(
     conn: &mut Connection,
     credential: &Credential,
+    server_api: Option<&ServerApi>,
     http_client: &HttpClient,
 ) -> Result<()> {
     let source = match credential.source.as_deref() {
@@ -51,6 +55,7 @@ pub(super) async fn authenticate_stream(
         source.into(),
         AuthMechanism::MongoDbAws,
         client_first_payload_bytes,
+        server_api.cloned(),
     );
     let client_first = sasl_start.into_command();
 
@@ -85,6 +90,7 @@ pub(super) async fn authenticate_stream(
         source.into(),
         server_first.conversation_id.clone(),
         client_second_payload_bytes,
+        server_api.cloned(),
     );
 
     let client_second = sasl_continue.into_command();

src/client/auth/mod.rs

@@ -20,6 +20,7 @@ use typed_builder::TypedBuilder;
 use self::scram::ScramVersion;
 use crate::{
     bson::Document,
+    client::options::ServerApi,
     cmap::{Command, Connection, StreamDescription},
     error::{Error, ErrorKind, Result},
     runtime::HttpClient,
@@ -227,9 +228,9 @@ impl AuthMechanism {
 
                 Ok(Some(ClientFirst::Scram(ScramVersion::Sha256, client_first)))
             }
-            Self::MongoDbX509 => Ok(Some(ClientFirst::X509(x509::build_client_first(
-                credential,
-            )))),
+            Self::MongoDbX509 => Ok(Some(ClientFirst::X509(
+                x509::build_speculative_client_first(credential),
+            ))),
             Self::Plain => Ok(None),
             #[cfg(feature = "tokio-runtime")]
             AuthMechanism::MongoDbAws => Ok(None),
@@ -250,26 +251,31 @@ impl AuthMechanism {
         &self,
         stream: &mut Connection,
         credential: &Credential,
+        server_api: Option<&ServerApi>,
         #[cfg_attr(not(feature = "tokio-runtime"), allow(unused))] http_client: &HttpClient,
     ) -> Result<()> {
         self.validate_credential(credential)?;
 
         match self {
             AuthMechanism::ScramSha1 => {
                 ScramVersion::Sha1
-                    .authenticate_stream(stream, credential, None)
+                    .authenticate_stream(stream, credential, server_api, None)
                     .await
             }
             AuthMechanism::ScramSha256 => {
                 ScramVersion::Sha256
-                    .authenticate_stream(stream, credential, None)
+                    .authenticate_stream(stream, credential, server_api, None)
                     .await
             }
-            AuthMechanism::MongoDbX509 => x509::authenticate_stream(stream, credential, None).await,
-            AuthMechanism::Plain => plain::authenticate_stream(stream, credential).await,
+            AuthMechanism::MongoDbX509 => {
+                x509::authenticate_stream(stream, credential, server_api, None).await
+            }
+            AuthMechanism::Plain => {
+                plain::authenticate_stream(stream, credential, server_api).await
+            }
             #[cfg(feature = "tokio-runtime")]
             AuthMechanism::MongoDbAws => {
-                aws::authenticate_stream(stream, credential, http_client).await
+                aws::authenticate_stream(stream, credential, server_api, http_client).await
             }
             AuthMechanism::MongoDbCr => Err(ErrorKind::AuthenticationError {
                 message: "MONGODB-CR is deprecated and not supported by this driver. Use SCRAM \
@@ -393,6 +399,7 @@ impl Credential {
         &self,
         conn: &mut Connection,
         http_client: &HttpClient,
+        server_api: Option<&ServerApi>,
         first_round: Option<FirstRound>,
     ) -> Result<()> {
         let stream_description = conn.stream_description()?;
@@ -407,10 +414,12 @@ impl Credential {
         if let Some(first_round) = first_round {
             return match first_round {
                 FirstRound::Scram(version, first_round) => {
-                    version.authenticate_stream(conn, self, first_round).await
+                    version
+                        .authenticate_stream(conn, self, server_api, first_round)
+                        .await
                 }
                 FirstRound::X509(server_first) => {
-                    x509::authenticate_stream(conn, self, server_first).await
+                    x509::authenticate_stream(conn, self, server_api, server_first).await
                 }
             };
         }
@@ -421,7 +430,9 @@ impl Credential {
         };
 
         // Authenticate according to the chosen mechanism.
-        mechanism.authenticate_stream(conn, self, http_client).await
+        mechanism
+            .authenticate_stream(conn, self, server_api, http_client)
+            .await
     }
 }
 

src/client/auth/plain.rs

@@ -1,8 +1,11 @@
 use crate::{
-    client::auth::{
-        sasl::{SaslResponse, SaslStart},
-        AuthMechanism,
-        Credential,
+    client::{
+        auth::{
+            sasl::{SaslResponse, SaslStart},
+            AuthMechanism,
+            Credential,
+        },
+        options::ServerApi,
     },
     cmap::Connection,
     error::{Error, Result},
@@ -11,6 +14,7 @@ use crate::{
 pub(crate) async fn authenticate_stream(
     conn: &mut Connection,
     credential: &Credential,
+    server_api: Option<&ServerApi>,
 ) -> Result<()> {
     let source = credential.source.as_deref().unwrap_or("$external");
     let username = credential
@@ -27,6 +31,7 @@ pub(crate) async fn authenticate_stream(
         source.into(),
         AuthMechanism::Plain,
         payload_bytes(username, password),
+        server_api.cloned(),
     )
     .into_command();
 

src/client/auth/sasl.rs

@@ -1,7 +1,7 @@
 use crate::{
     bson::{doc, spec::BinarySubtype, Binary, Bson, Document},
     bson_util,
-    client::auth::AuthMechanism,
+    client::{auth::AuthMechanism, options::ServerApi},
     cmap::Command,
     error::{Error, Result},
 };
@@ -11,14 +11,21 @@ pub(super) struct SaslStart {
     source: String,
     mechanism: AuthMechanism,
     payload: Vec<u8>,
+    server_api: Option<ServerApi>,
 }
 
 impl SaslStart {
-    pub(super) fn new(source: String, mechanism: AuthMechanism, payload: Vec<u8>) -> Self {
+    pub(super) fn new(
+        source: String,
+        mechanism: AuthMechanism,
+        payload: Vec<u8>,
+        server_api: Option<ServerApi>,
+    ) -> Self {
         Self {
             source,
             mechanism,
             payload,
+            server_api,
         }
     }
 
@@ -34,7 +41,12 @@ impl SaslStart {
             body.insert("options", doc! { "skipEmptyExchange": true });
         }
 
-        Command::new("saslStart".into(), self.source, body)
+        let mut command = Command::new("saslStart".into(), self.source, body);
+        if let Some(server_api) = self.server_api {
+            command.set_server_api(&server_api);
+        }
+
+        command
     }
 }
 
@@ -43,14 +55,21 @@ pub(super) struct SaslContinue {
     source: String,
     conversation_id: Bson,
     payload: Vec<u8>,
+    server_api: Option<ServerApi>,
 }
 
 impl SaslContinue {
-    pub(super) fn new(source: String, conversation_id: Bson, payload: Vec<u8>) -> Self {
+    pub(super) fn new(
+        source: String,
+        conversation_id: Bson,
+        payload: Vec<u8>,
+        server_api: Option<ServerApi>,
+    ) -> Self {
         Self {
             source,
             conversation_id,
             payload,
+            server_api,
         }
     }
 
@@ -61,7 +80,12 @@ impl SaslContinue {
             "payload": Binary { subtype: BinarySubtype::Generic, bytes: self.payload },
         };
 
-        Command::new("saslContinue".into(), self.source, body)
+        let mut command = Command::new("saslContinue".into(), self.source, body);
+        if let Some(server_api) = self.server_api {
+            command.set_server_api(&server_api);
+        }
+
+        command
     }
 }