sasl/scram (#33468)

<!--
Describe the contents of the PR briefly but completely.

If you write detailed commit messages, it is acceptable to copy/paste
them
here, or write "see commit messages for details." If there is only one
commit
in the PR, GitHub will have already added its commit message above.
-->

### Motivation

<!--
Which of the following best describes the motivation behind this PR?

  * This PR fixes a recognized bug.

    [Ensure issue is linked somewhere.]

  * This PR adds a known-desirable feature.

    [Ensure issue is linked somewhere.]

  * This PR fixes a previously unreported bug.

    [Describe the bug in detail, as if you were filing a bug report.]

  * This PR adds a feature that has not yet been specified.

[Write a brief specification for the feature, including justification
for its inclusion in Materialize, as if you were writing the original
     feature specification.]

   * This PR refactors existing code.

[Describe what was wrong with the existing code, if it is not obvious.]
-->

### Tips for reviewer

<!--
Leave some tips for your reviewer, like:

    * The diff is much smaller if viewed with whitespace hidden.
    * [Some function/module/file] deserves extra attention.
* [Some function/module/file] is pure code movement and only needs a
skim.

Delete this section if no tips.
-->

### Checklist

- [ ] This PR has adequate test coverage / QA involvement has been duly
considered. ([trigger-ci for additional test/nightly
runs](https://trigger-ci.dev.materialize.com/))
- [ ] This PR has an associated up-to-date [design
doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md),
is a design doc
([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)),
or is sufficiently small to not require a design.
  <!-- Reference the design in the description. -->
- [ ] If this PR evolves [an existing `$T ⇔ Proto$T`
mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md)
(possibly in a backwards-incompatible way), then it is tagged with a
`T-proto` label.
- [ ] If this PR will require changes to cloud orchestration or tests,
there is a companion cloud PR to account for those changes that is
tagged with the release-blocker label
([example](MaterializeInc/cloud#5021)).
<!-- Ask in #team-cloud on Slack if you need help preparing the cloud
PR. -->
- [ ] If this PR includes major [user-facing behavior
changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note),
I have pinged the relevant PM to schedule a changelog post.

Author: DAlperin

Cargo.lock

@@ -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/Cargo.toml

@@ -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/open.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/catalog/state.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/client.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/command.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,