async password validation (#7171)

## Problem

password hashing can block main thread

## Summary of changes

spawn_blocking the password hash call

proxy/src/auth/backend.rs

@@ -254,7 +254,7 @@ async fn authenticate_with_secret(
     config: &'static AuthenticationConfig,
 ) -> auth::Result<ComputeCredentials> {
     if let Some(password) = unauthenticated_password {
-        let auth_outcome = validate_password_and_exchange(&password, secret)?;
+        let auth_outcome = validate_password_and_exchange(&password, secret).await?;
         let keys = match auth_outcome {
             crate::sasl::Outcome::Success(key) => key,
             crate::sasl::Outcome::Failure(reason) => {

proxy/src/auth/flow.rs

@@ -126,7 +126,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> AuthFlow<'_, S, CleartextPassword> {
             .strip_suffix(&[0])
             .ok_or(AuthErrorImpl::MalformedPassword("missing terminator"))?;
 
-        let outcome = validate_password_and_exchange(password, self.state.0)?;
+        let outcome = validate_password_and_exchange(password, self.state.0).await?;
 
         if let sasl::Outcome::Success(_) = &outcome {
             self.stream.write_message_noflush(&Be::AuthenticationOk)?;
@@ -180,7 +180,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> AuthFlow<'_, S, Scram<'_>> {
     }
 }
 
-pub(crate) fn validate_password_and_exchange(
+pub(crate) async fn validate_password_and_exchange(
     password: &[u8],
     secret: AuthSecret,
 ) -> super::Result<sasl::Outcome<ComputeCredentialKeys>> {
@@ -200,7 +200,8 @@ pub(crate) fn validate_password_and_exchange(
                 &scram_secret,
                 sasl_client,
                 crate::config::TlsServerEndPoint::Undefined,
-            )?;
+            )
+            .await?;
 
             let client_key = match outcome {
                 sasl::Outcome::Success(client_key) => client_key,

proxy/src/scram.rs

@@ -113,7 +113,7 @@ mod tests {
         );
     }
 
-    fn run_round_trip_test(server_password: &str, client_password: &str) {
+    async fn run_round_trip_test(server_password: &str, client_password: &str) {
         let scram_secret = ServerSecret::build(server_password).unwrap();
         let sasl_client =
             ScramSha256::new(client_password.as_bytes(), ChannelBinding::unsupported());
@@ -123,6 +123,7 @@ mod tests {
             sasl_client,
             crate::config::TlsServerEndPoint::Undefined,
         )
+        .await
         .unwrap();
 
         match outcome {
@@ -131,14 +132,14 @@ mod tests {
         }
     }
 
-    #[test]
-    fn round_trip() {
-        run_round_trip_test("pencil", "pencil")
+    #[tokio::test]
+    async fn round_trip() {
+        run_round_trip_test("pencil", "pencil").await
     }
 
-    #[test]
+    #[tokio::test]
     #[should_panic(expected = "password doesn't match")]
-    fn failure() {
-        run_round_trip_test("pencil", "eraser")
+    async fn failure() {
+        run_round_trip_test("pencil", "eraser").await
     }
 }

proxy/src/scram/exchange.rs

@@ -71,7 +71,7 @@ impl<'a> Exchange<'a> {
     }
 }
 
-pub fn exchange(
+pub async fn exchange(
     secret: &ServerSecret,
     mut client: ScramSha256,
     tls_server_end_point: config::TlsServerEndPoint,
@@ -86,7 +86,14 @@ pub fn exchange(
         .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
     let sent = match init.transition(secret, &tls_server_end_point, client_first)? {
         Continue(sent, server_first) => {
-            client.update(server_first.as_bytes())?;
+            // `client.update` might perform `pbkdf2(pw)`, best to spawn it in a blocking thread.
+            // TODO(conrad): take this code from tokio-postgres and make an async-aware pbkdf2 impl
+            client = tokio::task::spawn_blocking(move || {
+                client.update(server_first.as_bytes())?;
+                Ok::<ScramSha256, std::io::Error>(client)
+            })
+            .await
+            .expect("should not panic while performing password hash")?;
             sent
         }
         Success(x, _) => match x {},

proxy/src/serverless/backend.rs

@@ -50,7 +50,7 @@ impl PoolingBackend {
             }
         };
         let auth_outcome =
-            crate::auth::validate_password_and_exchange(&conn_info.password, secret)?;
+            crate::auth::validate_password_and_exchange(&conn_info.password, secret).await?;
         let res = match auth_outcome {
             crate::sasl::Outcome::Success(key) => Ok(key),
             crate::sasl::Outcome::Failure(reason) => {