Allow multiple users to connect to Readyset

This commit allows multiple users to connect to Readyset by
adding a new parameter --allowed-users containing a comma-separated
list of username:password that will be used to authenticate
connections to the Readyset server. The same user and password will
be used for both the Readyset server and the upstream database.

This commit also remove the obsolete --username and --password
parameters.

Closes: REA-4641
Closes: #1349

Release-Note-Core: It's now possible to connect to Readyset with
  multiple users by specifying a comma-separated list of
  username:password pairs in the --allowed-users parameter.

Change-Id: Ie1d77daa6ae6430bce92e56ed2726b7016bbd0f9
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/8804
Tested-by: Buildkite CI
Reviewed-by: Jason Brown <jason.b@readyset.io>

Author: altmannmarcelo

Cargo.lock

@@ -32,7 +32,7 @@ impl Backend {
 
         match DatabaseURL::from_str(url)? {
             DatabaseURL::MySQL(_) => {
-                let upstream = MySqlUpstream::connect(upstream_config.clone()).await?;
+                let upstream = MySqlUpstream::connect(upstream_config.clone(), None, None).await?;
                 let status_reporter = ReadySetStatusReporter::new(
                     upstream_config.clone(),
                     noria.handle(),
@@ -56,7 +56,8 @@ impl Backend {
                 ))
             }
             DatabaseURL::PostgreSQL(_) => {
-                let upstream = PostgreSqlUpstream::connect(upstream_config.clone()).await?;
+                let upstream =
+                    PostgreSqlUpstream::connect(upstream_config.clone(), None, None).await?;
                 let status_reporter = ReadySetStatusReporter::new(
                     upstream_config.clone(),
                     noria.handle(),

benchmarks/src/utils/backend.rs

@@ -31,6 +31,7 @@ tracing = { workspace = true }
 database-utils = { path = "../database-utils" }
 readyset-adapter-types = { path = "../readyset-adapter-types" }
 readyset-data = { path = "../readyset-data" }
+readyset-util = { path = "../readyset-util" }
 
 [dev-dependencies]
 tokio-postgres = { workspace = true }

mysql-srv/Cargo.toml

@@ -22,6 +22,7 @@
 //! use std::collections::HashMap;
 //! use std::iter;
 //!
+//! use readyset_util::redacted::RedactedString;
 //! use database_utils::TlsMode;
 //! use mysql::prelude::*;
 //! use mysql_srv::*;
@@ -64,6 +65,10 @@
 //!         Ok(())
 //!     }
 //!
+//!     async fn set_auth_info(&mut self, _: &str, _: Option<RedactedString>) -> io::Result<()> {
+//!         Ok(())
+//!     }
+//!
 //!     async fn on_query(
 //!         &mut self,
 //!         query: &str,
@@ -182,6 +187,7 @@ use error::{other_error, OtherErrorKind};
 use mysql_common::constants::CapabilityFlags;
 use readyset_adapter_types::{DeallocateId, ParsedCommand};
 use readyset_data::DfType;
+use readyset_util::redacted::RedactedString;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::net;
 use tokio_native_tls::TlsAcceptor;
@@ -312,6 +318,9 @@ pub trait MySqlShim<S: AsyncRead + AsyncWrite + Unpin + Send> {
     /// Called when client switches user.
     async fn on_change_user(&mut self, _: &str, _: &str, _: &str) -> io::Result<()>;
 
+    /// Called when client authenticates to inform which users we should use.
+    async fn set_auth_info(&mut self, _: &str, _: Option<RedactedString>) -> io::Result<()>;
+
     /// Retrieve the password for the user with the given username, if any.
     ///
     /// If the user doesn't exist, return [`None`].
@@ -438,7 +447,8 @@ impl<B: MySqlShim<S> + Send, S: AsyncWrite + AsyncRead + Unpin + Send> MySqlInte
             tls_acceptor,
             tls_mode,
         };
-        if let (true, database) = mi.init().await? {
+        if let (true, username, password, database) = mi.init().await? {
+            mi.shim.set_auth_info(&username, password).await?;
             if let Some(database) = database {
                 mi.shim.on_init(&database, None).await?;
             }
@@ -455,9 +465,11 @@ impl<B: MySqlShim<S> + Send, S: AsyncWrite + AsyncRead + Unpin + Send> MySqlInte
     /// sent and received as needed to complete authentication.
     ///
     /// If no errors are encountered, the return value contains a tuple of a boolean to indicate
-    /// whether authentication was successful, and a database name if one was specified by the
-    /// client in the handshake response.
-    async fn init(&mut self) -> Result<(bool, Option<String>), io::Error> {
+    /// whether authentication was successful, the username, the plaintext password if one was
+    /// provided, and a database name if one was specified by the client in the handshake response.
+    async fn init(
+        &mut self,
+    ) -> Result<(bool, String, Option<RedactedString>, Option<String>), io::Error> {
         let auth_data =
             generate_auth_data().map_err(|_| other_error(OtherErrorKind::AuthDataErr))?;
         self.auth_data = auth_data;
@@ -546,7 +558,7 @@ impl<B: MySqlShim<S> + Send, S: AsyncWrite + AsyncRead + Unpin + Send> MySqlInte
                 )
                 .await?;
                 self.conn.flush().await?;
-                return Ok((false, None));
+                return Ok((false, "".to_string(), None, None));
             }
         }
 
@@ -601,7 +613,7 @@ impl<B: MySqlShim<S> + Send, S: AsyncWrite + AsyncRead + Unpin + Send> MySqlInte
                     &mut self.conn,
                 )
                 .await?;
-                return Ok((false, database));
+                return Ok((false, "".to_string(), None, database));
             }
 
             debug!(
@@ -632,17 +644,24 @@ impl<B: MySqlShim<S> + Send, S: AsyncWrite + AsyncRead + Unpin + Send> MySqlInte
             password
         };
 
-        let auth_success = !self.shim.require_authentication()
-            || self
-                .shim
-                .password_for_username(&username)
-                .is_some_and(|password| {
-                    let expected = hash_password(&password, &auth_data);
-                    let actual = handshake_password.as_slice();
-                    trace!(?expected, ?actual);
-                    expected == actual
-                });
-
+        let plain_password = self.shim.password_for_username(&username);
+        let require_auth = self.shim.require_authentication();
+        let auth_success = !require_auth
+            || plain_password.as_ref().is_some_and(|password| {
+                let expected = hash_password(password, &auth_data);
+                let actual = handshake_password.as_slice();
+                trace!(?expected, ?actual);
+                expected == actual
+            });
+        let plain_password = if require_auth {
+            Some(RedactedString::from(
+                plain_password
+                    .map(|p| String::from_utf8_lossy(&p).into_owned())
+                    .unwrap_or_default(),
+            ))
+        } else {
+            None
+        };
         if auth_success {
             debug!(%username, "Successfully authenticated client");
             writers::write_ok_packet(&mut self.conn, 0, 0, StatusFlags::empty()).await?;
@@ -656,8 +675,7 @@ impl<B: MySqlShim<S> + Send, S: AsyncWrite + AsyncRead + Unpin + Send> MySqlInte
             .await?;
         }
         self.conn.flush().await?;
-
-        Ok((auth_success, database))
+        Ok((auth_success, username, plain_password, database))
     }
 
     async fn run(mut self) -> Result<(), io::Error> {

mysql-srv/src/lib.rs

@@ -20,6 +20,7 @@ use mysql_srv::{
     QueryResultWriter, QueryResultsResponse, StatementMetaWriter,
 };
 use readyset_adapter_types::DeallocateId;
+use readyset_util::redacted::RedactedString;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::net::TcpStream;
 
@@ -73,6 +74,14 @@ where
         info.reply(id, &self.params, &self.columns).await
     }
 
+    async fn set_auth_info(
+        &mut self,
+        _user: &str,
+        _password: Option<RedactedString>,
+    ) -> io::Result<()> {
+        Ok(())
+    }
+
     async fn on_execute(
         &mut self,
         id: u32,

mysql-srv/tests/main.rs

@@ -36,6 +36,7 @@ use postgres_types::Type;
 use protocol::Protocol;
 use readyset_adapter_types::{DeallocateId, PreparedStatementType};
 use readyset_sql::ast::SqlIdentifier;
+use readyset_util::redacted::RedactedString;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_native_tls::TlsAcceptor;
 use tokio_postgres::OwnedField;
@@ -77,6 +78,9 @@ pub trait PsqlBackend {
     /// [libpq-server-version]: https://github.com/postgres/postgres/blob/d22646922d66012705e0e2948cfb5b4a07092a29/src/interfaces/libpq/fe-exec.c#L1146-L1178
     fn version(&self) -> String;
 
+    /// Called when client authenticates to inform which users we should use.
+    async fn set_auth_info(&mut self, user: &str, password: Option<RedactedString>);
+
     /// Initializes the backend.
     ///
     /// * `database` - The name of the database that will be used for queries to this `Backend`

psql-srv/src/lib.rs

@@ -8,6 +8,7 @@ use postgres::SimpleQueryMessage;
 use postgres_protocol::Oid;
 use postgres_types::{Kind, Type};
 use readyset_adapter_types::{DeallocateId, PreparedStatementType};
+use readyset_util::redacted::RedactedString;
 use smallvec::smallvec;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_postgres::CommandCompleteContents;
@@ -144,6 +145,9 @@ pub struct Protocol {
 
     /// Whether the client is connected using TLS
     is_tls: bool,
+
+    /// The backend password for the current user
+    backend_password: Option<RedactedString>,
 }
 
 /// A prepared statement allows a frontend to specify the general form of a SQL statement while
@@ -179,6 +183,7 @@ impl Protocol {
             tls_mode,
             tls_server_end_point: None,
             is_tls: false,
+            backend_password: None,
         }
     }
 
@@ -428,6 +433,11 @@ impl Protocol {
 
         self.state = State::Ready;
 
+        // Set it here in case the backend gets closed and reconnect
+        self.backend_password = Some(RedactedString::from(password.to_string()));
+        let _ = backend
+            .set_auth_info(&user, Some(RedactedString::from(password.to_string())))
+            .await;
         Ok(Response::Messages(Self::get_ready_message(
             backend.version(),
         )))
@@ -448,11 +458,15 @@ impl Protocol {
             }
             Some(Credentials::Any) => {
                 self.state = State::Ready;
+                self.backend_password = None;
                 return Ok(Response::Messages(Self::get_ready_message(
                     backend.version(),
                 )));
             }
-            Some(Credentials::CleartextPassword(pw)) => pw,
+            Some(Credentials::CleartextPassword(pw)) => {
+                self.backend_password = Some(RedactedString::from(pw.to_string()));
+                pw
+            }
         };
 
         let SaslInitialResponse {
@@ -547,6 +561,14 @@ impl Protocol {
                 sasl_data: server_final_message.to_string().into(),
             }];
             messages.extend(Self::get_ready_message(backend.version()));
+            let _ = backend
+                .set_auth_info(
+                    &user,
+                    self.backend_password
+                        .as_ref()
+                        .map(|password| RedactedString::from(password.to_string())),
+                )
+                .await;
             Ok(Response::Messages(messages.into()))
         } else {
             Err(Error::AuthenticationFailure {
@@ -1261,6 +1283,8 @@ mod tests {
             "14.5 ReadySet".to_string()
         }
 
+        async fn set_auth_info(&mut self, _user: &str, _password: Option<RedactedString>) {}
+
         async fn on_init(&mut self, database: &str) -> Result<CredentialsNeeded, Error> {
             self.database = Some(database.to_string());
             match &self.needed_credentials {

psql-srv/src/protocol.rs

@@ -14,6 +14,7 @@ use psql_srv::{
     PsqlValue, QueryResponse, TransferFormat,
 };
 use readyset_adapter_types::{DeallocateId, PreparedStatementType};
+use readyset_util::redacted::RedactedString;
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
 use tokio_native_tls::{native_tls, TlsAcceptor};
@@ -41,6 +42,7 @@ impl PsqlBackend for ScramSha256Backend {
         "13.4 ReadySet".to_string()
     }
 
+    async fn set_auth_info(&mut self, _user: &str, _password: Option<RedactedString>) {}
     fn credentials_for_user(&self, user: &str) -> Option<Credentials> {
         if self.username == user {
             Some(Credentials::CleartextPassword(self.password))

psql-srv/tests/authentication.rs

@@ -11,6 +11,7 @@ use psql_srv::{
     PsqlSrvRow, PsqlValue, QueryResponse, TransferFormat,
 };
 use readyset_adapter_types::{DeallocateId, PreparedStatementType};
+use readyset_util::redacted::RedactedString;
 use tokio::join;
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
@@ -34,6 +35,7 @@ impl PsqlBackend for ErrorBackend {
         Some(Credentials::Any)
     }
 
+    async fn set_auth_info(&mut self, _user: &str, _password: Option<RedactedString>) {}
     async fn on_init(&mut self, _database: &str) -> Result<CredentialsNeeded, Error> {
         Ok(CredentialsNeeded::None)
     }

psql-srv/tests/errors.rs

@@ -10,6 +10,7 @@ use psql_srv::{
     run_backend, Credentials, CredentialsNeeded, Error, PsqlBackend, PsqlSrvRow, TransferFormat,
 };
 use readyset_adapter_types::{DeallocateId, PreparedStatementType};
+use readyset_util::redacted::RedactedString;
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
 use tokio_native_tls::{native_tls, TlsAcceptor};
@@ -34,6 +35,8 @@ impl PsqlBackend for TestBackend {
         Some(Credentials::Any)
     }
 
+    async fn set_auth_info(&mut self, _user: &str, _password: Option<RedactedString>) {}
+
     async fn on_init(
         &mut self,
         _database: &str,