Password auth

Cargo.toml

@@ -7,6 +7,7 @@ members = [
     "src/alloc-default",
     "src/arrow-util",
     "src/audit-log",
+    "src/auth",
     "src/avro",
     "src/aws-secrets-controller",
     "src/aws-util",
@@ -236,6 +237,7 @@ default-members = [
     "src/workspace-hack",
     "test/metabase/smoketest",
     "test/test-util",
+    "src/auth",
 ]
 
 exclude = [

WORKSPACE

@@ -704,6 +704,7 @@ crates_repository(
         "//:test/metabase/smoketest/Cargo.toml",
         "//:test/test-util/Cargo.toml",
         "//:misc/bazel/cargo-gazelle/Cargo.toml",
+        "//:src/auth/Cargo.toml",
     ],
     rust_version = RUST_VERSION,
     # Restricting the set of platform triples we support _greatly_ reduces the

src/adapter-types/src/dyncfgs.rs

@@ -118,6 +118,13 @@ pub const ENABLE_MULTI_REPLICA_SOURCES: Config<bool> = Config::new(
     "Enable multi-replica sources.",
 );
 
+/// Whether to enable self-managed authentication.
+pub const ENABLE_SELF_MANAGED_AUTH: Config<bool> = Config::new(
+    "enable_self_managed_auth",
+    false,
+    "Enable self-managed authentication.",
+);
+
 pub const CONSTRAINT_BASED_TIMESTAMP_SELECTION: Config<&'static str> = Config::new(
     "constraint_based_timestamp_selection",
     ConstraintBasedTimestampSelection::const_default().as_str(),
@@ -148,6 +155,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&ENABLE_CONTINUAL_TASK_BUILTINS)
         .add(&ENABLE_EXPRESSION_CACHE)
         .add(&ENABLE_MULTI_REPLICA_SOURCES)
+        .add(&ENABLE_SELF_MANAGED_AUTH)
         .add(&CONSTRAINT_BASED_TIMESTAMP_SELECTION)
         .add(&PERSIST_FAST_PATH_ORDER)
 }

src/adapter/BUILD.bazel

@@ -34,6 +34,7 @@ rust_library(
     deps = [
         "//src/adapter-types:mz_adapter_types",
         "//src/audit-log:mz_audit_log",
+        "//src/auth:mz_auth",
         "//src/build-info:mz_build_info",
         "//src/catalog:mz_catalog",
         "//src/cloud-provider:mz_cloud_provider",
@@ -104,6 +105,7 @@ rust_test(
     deps = [
         "//src/adapter-types:mz_adapter_types",
         "//src/audit-log:mz_audit_log",
+        "//src/auth:mz_auth",
         "//src/build-info:mz_build_info",
         "//src/catalog:mz_catalog",
         "//src/cloud-provider:mz_cloud_provider",
@@ -153,6 +155,7 @@ rust_doc_test(
     deps = [
         "//src/adapter-types:mz_adapter_types",
         "//src/audit-log:mz_audit_log",
+        "//src/auth:mz_auth",
         "//src/build-info:mz_build_info",
         "//src/catalog:mz_catalog",
         "//src/cloud-provider:mz_cloud_provider",
@@ -223,6 +226,7 @@ rust_test(
         ":mz_adapter",
         "//src/adapter-types:mz_adapter_types",
         "//src/audit-log:mz_audit_log",
+        "//src/auth:mz_auth",
         "//src/build-info:mz_build_info",
         "//src/catalog:mz_catalog",
         "//src/cloud-provider:mz_cloud_provider",
@@ -293,6 +297,7 @@ rust_test(
         ":mz_adapter",
         "//src/adapter-types:mz_adapter_types",
         "//src/audit-log:mz_audit_log",
+        "//src/auth:mz_auth",
         "//src/build-info:mz_build_info",
         "//src/catalog:mz_catalog",
         "//src/cloud-provider:mz_cloud_provider",
@@ -363,6 +368,7 @@ rust_test(
         ":mz_adapter",
         "//src/adapter-types:mz_adapter_types",
         "//src/audit-log:mz_audit_log",
+        "//src/auth:mz_auth",
         "//src/build-info:mz_build_info",
         "//src/catalog:mz_catalog",
         "//src/cloud-provider:mz_cloud_provider",

src/adapter/Cargo.toml

@@ -32,6 +32,7 @@ launchdarkly-server-sdk = { version = "2.5.1", default-features = false }
 maplit = "1.0.2"
 mz-adapter-types = { path = "../adapter-types" }
 mz-audit-log = { path = "../audit-log" }
+mz-auth = { path = "../auth" }
 mz-build-info = { path = "../build-info" }
 mz-catalog = { path = "../catalog" }
 mz-cloud-provider = { path = "../cloud-provider", default-features = false }

src/adapter/src/catalog.rs

@@ -41,7 +41,7 @@ use mz_catalog::expr_cache::{ExpressionCacheHandle, GlobalExpressions, LocalExpr
 use mz_catalog::memory::error::{Error, ErrorKind};
 use mz_catalog::memory::objects::{
     CatalogCollectionEntry, CatalogEntry, CatalogItem, Cluster, ClusterReplica, Database,
-    NetworkPolicy, Role, Schema,
+    NetworkPolicy, Role, RoleAuth, Schema,
 };
 use mz_compute_types::dataflows::DataflowDescription;
 use mz_controller::clusters::ReplicaLocation;
@@ -1114,6 +1114,10 @@ impl Catalog {
         self.state.try_get_role_by_name(role_name)
     }
 
+    pub fn try_get_role_auth_by_id(&self, id: &RoleId) -> Option<&RoleAuth> {
+        self.state.try_get_role_auth_by_id(id)
+    }
+
     /// Creates a new schema in the `Catalog` for temporary items
     /// indicated by the TEMPORARY or TEMP keywords.
     pub fn create_temporary_schema(

src/adapter/src/catalog/apply.rs

@@ -24,13 +24,14 @@ use mz_catalog::builtin::{
     BUILTIN_LOG_LOOKUP, BUILTIN_LOOKUP, Builtin, BuiltinLog, BuiltinTable, BuiltinView,
 };
 use mz_catalog::durable::objects::{
-    ClusterKey, DatabaseKey, DurableType, ItemKey, NetworkPolicyKey, RoleKey, SchemaKey,
+    ClusterKey, DatabaseKey, DurableType, ItemKey, NetworkPolicyKey, RoleAuthKey, RoleKey,
+    SchemaKey,
 };
 use mz_catalog::durable::{CatalogError, SystemObjectMapping};
 use mz_catalog::memory::error::{Error, ErrorKind};
 use mz_catalog::memory::objects::{
     CatalogEntry, CatalogItem, Cluster, ClusterReplica, DataSourceDesc, Database, Func, Index, Log,
-    NetworkPolicy, Role, Schema, Source, StateDiff, StateUpdate, StateUpdateKind, Table,
+    NetworkPolicy, Role, RoleAuth, Schema, Source, StateDiff, StateUpdate, StateUpdateKind, Table,
     TableDataSource, TemporaryItem, Type, UpdateFrom,
 };
 use mz_compute_types::config::ComputeReplicaConfig;
@@ -75,6 +76,7 @@ use crate::util::index_sql;
 #[derive(Debug, Clone, Default)]
 struct InProgressRetractions {
     roles: BTreeMap<RoleKey, Role>,
+    role_auths: BTreeMap<RoleAuthKey, RoleAuth>,
     databases: BTreeMap<DatabaseKey, Database>,
     schemas: BTreeMap<SchemaKey, Schema>,
     clusters: BTreeMap<ClusterKey, Cluster>,
@@ -214,6 +216,9 @@ impl CatalogState {
             StateUpdateKind::Role(role) => {
                 self.apply_role_update(role, diff, retractions);
             }
+            StateUpdateKind::RoleAuth(role_auth) => {
+                self.apply_role_auth_update(role_auth, diff, retractions);
+            }
             StateUpdateKind::Database(database) => {
                 self.apply_database_update(database, diff, retractions);
             }
@@ -283,6 +288,22 @@ impl CatalogState {
         Ok(())
     }
 
+    #[instrument(level = "debug")]
+    fn apply_role_auth_update(
+        &mut self,
+        role_auth: mz_catalog::durable::RoleAuth,
+        diff: StateDiff,
+        retractions: &mut InProgressRetractions,
+    ) {
+        apply_with_update(
+            &mut self.role_auth_by_id,
+            role_auth,
+            |role_auth| role_auth.role_id,
+            diff,
+            &mut retractions.role_auths,
+        );
+    }
+
     #[instrument(level = "debug")]
     fn apply_role_update(
         &mut self,
@@ -1256,7 +1277,8 @@ impl CatalogState {
                 .pack_network_policy_update(&policy.id, diff)
                 .expect("could not pack audit log update"),
             StateUpdateKind::StorageCollectionMetadata(_)
-            | StateUpdateKind::UnfinalizedShard(_) => Vec::new(),
+            | StateUpdateKind::UnfinalizedShard(_)
+            | StateUpdateKind::RoleAuth(_) => Vec::new(),
         }
     }
 
@@ -1814,6 +1836,7 @@ fn sort_updates_inner(updates: Vec<StateUpdate>) -> Vec<StateUpdate> {
         let diff = update.diff.clone();
         match update.kind {
             StateUpdateKind::Role(_)
+            | StateUpdateKind::RoleAuth(_)
             | StateUpdateKind::Database(_)
             | StateUpdateKind::Schema(_)
             | StateUpdateKind::DefaultPrivilege(_)

src/adapter/src/catalog/consistency.rs

@@ -186,6 +186,11 @@ impl CatalogState {
                 }
             }
         }
+        for (role_id, _) in &self.role_auth_by_id {
+            if !self.roles_by_id.contains_key(role_id) {
+                inconsistencies.push(RoleInconsistency::RoleAuth(role_id.clone()));
+            }
+        }
         for (default_priv, privileges) in self.default_privileges.iter() {
             if !self.roles_by_id.contains_key(&default_priv.role_id) {
                 inconsistencies.push(RoleInconsistency::DefaultPrivilege(default_priv.clone()));
@@ -647,6 +652,7 @@ enum RoleInconsistency {
     Cluster(ClusterId, RoleId),
     ClusterReplica(ClusterId, ReplicaId, RoleId),
     DefaultPrivilege(DefaultPrivilegeObject),
+    RoleAuth(RoleId),
     DefaultPrivilegeItem {
         grantor: RoleId,
         grantee: RoleId,

src/adapter/src/catalog/open.rs

@@ -142,6 +142,7 @@ impl Catalog {
             roles_by_name: BTreeMap::new(),
             roles_by_id: BTreeMap::new(),
             network_policies_by_id: BTreeMap::new(),
+            role_auth_by_id: BTreeMap::new(),
             network_policies_by_name: BTreeMap::new(),
             system_configuration,
             default_privileges: DefaultPrivileges::default(),
@@ -266,6 +267,7 @@ impl Catalog {
         for (kind, ts, diff) in updates {
             match kind {
                 BootstrapStateUpdateKind::Role(_)
+                | BootstrapStateUpdateKind::RoleAuth(_)
                 | BootstrapStateUpdateKind::Database(_)
                 | BootstrapStateUpdateKind::Schema(_)
                 | BootstrapStateUpdateKind::DefaultPrivilege(_)

src/adapter/src/catalog/state.rs

@@ -32,8 +32,8 @@ use mz_catalog::memory::error::{Error, ErrorKind};
 use mz_catalog::memory::objects::{
     CatalogCollectionEntry, CatalogEntry, CatalogItem, Cluster, ClusterReplica, CommentsMap,
     Connection, DataSourceDesc, Database, DefaultPrivileges, Index, MaterializedView,
-    NetworkPolicy, Role, Schema, Secret, Sink, Source, SourceReferences, Table, TableDataSource,
-    Type, View,
+    NetworkPolicy, Role, RoleAuth, Schema, Secret, Sink, Source, SourceReferences, Table,
+    TableDataSource, Type, View,
 };
 use mz_controller::clusters::{
     ManagedReplicaAvailabilityZones, ManagedReplicaLocation, ReplicaAllocation, ReplicaLocation,
@@ -129,6 +129,8 @@ pub struct CatalogState {
     pub(super) network_policies_by_name: BTreeMap<String, NetworkPolicyId>,
     #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
     pub(super) network_policies_by_id: BTreeMap<NetworkPolicyId, NetworkPolicy>,
+    #[serde(serialize_with = "mz_ore::serde::map_key_to_string")]
+    pub(super) role_auth_by_id: BTreeMap<RoleId, RoleAuth>,
 
     #[serde(skip)]
     pub(super) system_configuration: SystemVars,
@@ -277,6 +279,7 @@ impl CatalogState {
             roles_by_name: Default::default(),
             roles_by_id: Default::default(),
             network_policies_by_id: Default::default(),
+            role_auth_by_id: Default::default(),
             config: CatalogConfig {
                 start_time: Default::default(),
                 start_instant: Instant::now(),
@@ -863,6 +866,10 @@ impl CatalogState {
             .map(|id| &self.roles_by_id[id])
     }
 
+    pub(super) fn try_get_role_auth_by_id(&self, id: &RoleId) -> Option<&RoleAuth> {
+        self.role_auth_by_id.get(id)
+    }
+
     pub(super) fn try_get_network_policy_by_name(
         &self,
         policy_name: &str,