sasl/scram

src/adapter/Cargo.toml

@@ -14,6 +14,7 @@ anyhow = "1.0.98"
 arrow = { version = "55.2.0", default-features = false }
 async-stream = "0.3.6"
 async-trait = "0.1.88"
+base64 = "0.22.1"
 bytes = "1.10.1"
 bytesize = "1.3.0"
 chrono = { version = "0.4.39", default-features = false, features = ["std"] }

src/adapter/src/catalog/open.rs

@@ -151,6 +151,7 @@ impl Catalog {
             source_references: BTreeMap::new(),
             storage_metadata: Default::default(),
             temporary_schemas: BTreeMap::new(),
+            mock_authentication_nonce: Default::default(),
             config: mz_sql::catalog::CatalogConfig {
                 start_time: to_datetime((config.now)()),
                 start_instant: Instant::now(),
@@ -401,6 +402,13 @@ impl Catalog {
             .unwrap_or("new")
             .to_string();
 
+        let mz_authentication_mock_nonce =
+            txn.get_authentication_mock_nonce().ok_or_else(|| {
+                Error::new(ErrorKind::SettingError("authentication nonce".to_string()))
+            })?;
+
+        state.mock_authentication_nonce = Some(mz_authentication_mock_nonce);
+
         // Migrate item ASTs.
         let builtin_table_update = if !config.skip_migrations {
             let migrate_result = migrate::migrate(

src/adapter/src/catalog/state.rs

@@ -142,6 +142,7 @@ pub struct CatalogState {
     #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
     pub(super) source_references: BTreeMap<CatalogItemId, SourceReferences>,
     pub(super) storage_metadata: StorageMetadata,
+    pub(super) mock_authentication_nonce: Option<String>,
 
     // Mutable state not derived from the durable catalog.
     #[serde(skip)]
@@ -316,6 +317,7 @@ impl CatalogState {
             source_references: Default::default(),
             storage_metadata: Default::default(),
             license_key: ValidatedLicenseKey::for_tests(),
+            mock_authentication_nonce: Default::default(),
         }
     }
 
@@ -2635,6 +2637,10 @@ impl CatalogState {
             CommentObjectId::NetworkPolicy(id) => self.get_network_policy(&id).name.clone(),
         }
     }
+
+    pub fn mock_authentication_nonce(&self) -> String {
+        self.mock_authentication_nonce.clone().unwrap_or_default()
+    }
 }
 
 impl ConnectionResolver for CatalogState {

src/adapter/src/client.rs

@@ -50,6 +50,7 @@ use uuid::Uuid;
 use crate::catalog::Catalog;
 use crate::command::{
     AuthResponse, CatalogDump, CatalogSnapshot, Command, ExecuteResponse, Response,
+    SASLChallengeResponse, SASLVerifyProofResponse,
 };
 use crate::coord::{Coordinator, ExecuteContextExtra};
 use crate::error::AdapterError;
@@ -168,6 +169,40 @@ impl Client {
         Ok(response)
     }
 
+    pub async fn generate_sasl_challenge(
+        &self,
+        user: &String,
+        client_nonce: &String,
+    ) -> Result<SASLChallengeResponse, AdapterError> {
+        let (tx, rx) = oneshot::channel();
+        self.send(Command::AuthenticateGetSASLChallenge {
+            role_name: user.to_string(),
+            nonce: client_nonce.to_string(),
+            tx,
+        });
+        let response = rx.await.expect("sender dropped")?;
+        Ok(response)
+    }
+
+    pub async fn verify_sasl_proof(
+        &self,
+        user: &String,
+        proof: &String,
+        nonce: &String,
+        mock_hash: &String,
+    ) -> Result<SASLVerifyProofResponse, AdapterError> {
+        let (tx, rx) = oneshot::channel();
+        self.send(Command::AuthenticateVerifySASLProof {
+            role_name: user.to_string(),
+            proof: proof.to_string(),
+            auth_message: nonce.to_string(),
+            mock_hash: mock_hash.to_string(),
+            tx,
+        });
+        let response = rx.await.expect("sender dropped")?;
+        Ok(response)
+    }
+
     /// Upgrades this client to a session client.
     ///
     /// A session is a connection that has successfully negotiated parameters,
@@ -927,6 +962,8 @@ impl SessionClient {
                 Command::GetWebhook { .. } => typ = Some("webhook"),
                 Command::Startup { .. }
                 | Command::AuthenticatePassword { .. }
+                | Command::AuthenticateGetSASLChallenge { .. }
+                | Command::AuthenticateVerifySASLProof { .. }
                 | Command::CatalogSnapshot { .. }
                 | Command::Commit { .. }
                 | Command::CancelRequest { .. }

src/adapter/src/command.rs

@@ -73,6 +73,20 @@ pub enum Command {
         password: Option<Password>,
     },
 
+    AuthenticateGetSASLChallenge {
+        tx: oneshot::Sender<Result<SASLChallengeResponse, AdapterError>>,
+        role_name: String,
+        nonce: String,
+    },
+
+    AuthenticateVerifySASLProof {
+        tx: oneshot::Sender<Result<SASLVerifyProofResponse, AdapterError>>,
+        role_name: String,
+        proof: String,
+        auth_message: String,
+        mock_hash: String,
+    },
+
     Execute {
         portal_name: String,
         session: Session,
@@ -148,6 +162,8 @@ impl Command {
             Command::CancelRequest { .. }
             | Command::Startup { .. }
             | Command::AuthenticatePassword { .. }
+            | Command::AuthenticateGetSASLChallenge { .. }
+            | Command::AuthenticateVerifySASLProof { .. }
             | Command::CatalogSnapshot { .. }
             | Command::PrivilegedCancelRequest { .. }
             | Command::GetWebhook { .. }
@@ -166,6 +182,8 @@ impl Command {
             Command::CancelRequest { .. }
             | Command::Startup { .. }
             | Command::AuthenticatePassword { .. }
+            | Command::AuthenticateGetSASLChallenge { .. }
+            | Command::AuthenticateVerifySASLProof { .. }
             | Command::CatalogSnapshot { .. }
             | Command::PrivilegedCancelRequest { .. }
             | Command::GetWebhook { .. }
@@ -210,6 +228,22 @@ pub struct AuthResponse {
     pub superuser: bool,
 }
 
+#[derive(Derivative)]
+#[derivative(Debug)]
+pub struct SASLChallengeResponse {
+    pub iteration_count: usize,
+    /// Base64-encoded salt for the SASL challenge.
+    pub salt: String,
+    pub nonce: String,
+}
+
+#[derive(Derivative)]
+#[derivative(Debug)]
+pub struct SASLVerifyProofResponse {
+    pub verifier: String,
+    pub auth_resp: AuthResponse,
+}
+
 // Facile implementation for `StartupResponse`, which does not use the `allowed`
 // feature of `ClientTransmitter`.
 impl Transmittable for StartupResponse {

src/adapter/src/coord.rs

@@ -356,6 +356,8 @@ impl Message {
                 Command::CheckConsistency { .. } => "command-check_consistency",
                 Command::Dump { .. } => "command-dump",
                 Command::AuthenticatePassword { .. } => "command-auth_check",
+                Command::AuthenticateGetSASLChallenge { .. } => "command-auth_get_sasl_challenge",
+                Command::AuthenticateVerifySASLProof { .. } => "command-auth_verify_sasl_proof",
             },
             Message::ControllerReady {
                 controller: ControllerReadiness::Compute,