Catalog: add migration for mock_authentication_nonce setting

The
(SASL/SCRAM)[#33468]
PR introduces the need for some stable, cluster wide, cryptographically
random key material. We use this material to be able to present
deterministic challenges for even users that don't exist to guard
against enumeration attacks. 

However that PR made a bad assumption that the initialize step of
catalog opening would always add this new key. But old versions that
have already been initialized wouldn't have it! This PR add code to
generate it for old versions

Author: DAlperin

src/catalog/src/durable/persist.rs

@@ -45,13 +45,15 @@ use timely::progress::{Antichain, Timestamp as TimelyTimestamp};
 use tracing::{debug, info, warn};
 use uuid::Uuid;
 
+use crate::durable::MOCK_AUTHENTICATION_NONCE_KEY;
 use crate::durable::debug::{Collection, CollectionType, DebugCatalogState, Trace};
 use crate::durable::error::FenceError;
 use crate::durable::initialize::{
     ENABLE_0DT_DEPLOYMENT_PANIC_AFTER_TIMEOUT, SYSTEM_CONFIG_SYNCED_KEY, USER_VERSION_KEY,
     WITH_0DT_DEPLOYMENT_DDL_CHECK_INTERVAL, WITH_0DT_DEPLOYMENT_MAX_WAIT,
 };
 use crate::durable::metrics::Metrics;
+use crate::durable::objects::serialization::proto;
 use crate::durable::objects::state_update::{
     IntoStateUpdateKindJson, StateUpdate, StateUpdateKind, StateUpdateKindJson,
     TryIntoStateUpdateKind,
@@ -65,6 +67,8 @@ use crate::durable::{
     ReadOnlyDurableCatalogState, Transaction, initialize,
 };
 use crate::memory;
+use base64::Engine as _;
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD_ENGINE;
 
 /// New-type used to represent timestamps in persist.
 pub(crate) type Timestamp = mz_repr::Timestamp;
@@ -971,6 +975,35 @@ pub(crate) type UnopenedPersistCatalogState =
     PersistHandle<StateUpdateKindJson, UnopenedCatalogStateInner>;
 
 impl UnopenedPersistCatalogState {
+    /// Generate setting backfill updates for settings that are part of the
+    /// always-deserializable `Setting` collection. These cannot be handled by the
+    /// versioned migrations, so we insert them here during open.
+    fn setting_migration_updates(&self) -> Vec<(StateUpdateKindJson, Diff)> {
+        let mut updates = Vec::new();
+
+        let mut ensure_setting = |name: &str, value_fn: &mut dyn FnMut() -> String| {
+            if !self.update_applier.settings.contains_key(name) {
+                let key = proto::SettingKey {
+                    name: name.to_string(),
+                };
+                let value = proto::SettingValue { value: value_fn() };
+                let kind = StateUpdateKind::Setting(key, value);
+                updates.push((kind.into(), Diff::ONE));
+            }
+        };
+
+        // Backfill MOCK_AUTHENTICATION_NONCE_KEY if missing.
+        {
+            let mut mk_nonce = || {
+                let mut nonce = [0u8; 24];
+                let _ = openssl::rand::rand_bytes(&mut nonce); // best-effort
+                BASE64_STANDARD_ENGINE.encode(nonce)
+            };
+            ensure_setting(MOCK_AUTHENTICATION_NONCE_KEY, &mut mk_nonce);
+        }
+
+        updates
+    }
     /// Create a new [`UnopenedPersistCatalogState`] to the catalog state associated with
     /// `organization_id`.
     ///
@@ -1231,6 +1264,17 @@ impl UnopenedPersistCatalogState {
             commit_ts = upgrade(&mut self, commit_ts).await?;
         }
 
+        // Perform setting backfills
+        if !read_only {
+            let setting_updates = self.setting_migration_updates();
+            if !setting_updates.is_empty() {
+                commit_ts = self
+                    .compare_and_append(setting_updates, commit_ts)
+                    .await
+                    .map_err(|e| CatalogError::from(e.unwrap_fence_error()))?;
+            }
+        }
+
         debug!(
             ?is_initialized,
             ?self.upper,