Authorize source access before purification

src/adapter/src/coord/command_handler.rs

@@ -49,7 +49,6 @@ use mz_sql::pure::{
     materialized_view_option_contains_temporal, purify_create_materialized_view_options,
 };
 use mz_sql::rbac;
-use mz_sql::rbac::CREATE_ITEM_USAGE;
 use mz_sql::session::user::User;
 use mz_sql::session::vars::{
     EndTransactionAction, NETWORK_POLICY, OwnedVarInput, STATEMENT_LOGGING_SAMPLE_RATE,
@@ -1593,14 +1592,16 @@ impl Coordinator {
                 task::spawn(|| format!("purify:{conn_id}"), async move {
                     let conn_catalog = catalog.for_session(ctx.session());
 
-                    // Checks if the session is authorized to purify a statement. Usually
-                    // authorization is checked after planning, however purification happens before
-                    // planning, which may require the use of some connections and secrets.
-                    if let Err(e) = rbac::check_usage(
+                    let statement_source = mz_sql::pure::statement_source(&conn_catalog, &stmt);
+
+                    // Authorization is usually checked after planning, but purification
+                    // happens before planning and may use connections and secrets, so it
+                    // gets its own check.
+                    if let Err(e) = rbac::check_purification(
                         &conn_catalog,
                         ctx.session(),
+                        statement_source,
                         &resolved_ids,
-                        &CREATE_ITEM_USAGE,
                     ) {
                         return ctx.retire(Err(e.into()));
                     }
@@ -1817,6 +1818,10 @@ impl Coordinator {
     }
 
     /// Whether the statement must be purified off of the Coordinator thread.
+    ///
+    /// Every statement listed here is authorized by [`rbac::check_purification`]
+    /// before purification runs, against the source that
+    /// [`mz_sql::pure::statement_source`] resolves for it.
     fn must_spawn_purification<A: AstInfo>(stmt: &Statement<A>) -> bool {
         // `CREATE` and `ALTER` `SOURCE` and `SINK` statements must be purified off the main
         // coordinator thread.

src/sql/src/pure.rs

@@ -273,6 +273,55 @@ pub enum PurifiedExportDetails {
     },
 }
 
+/// The existing source a statement drives, and how the statement uses it.
+///
+/// Resolved from the statement rather than from `ResolvedIds`: `ALTER SOURCE`
+/// carries its target as an `UnresolvedItemName` that name resolution never
+/// records.
+#[derive(Debug, Clone, Copy)]
+pub enum StatementSource {
+    Altered(CatalogItemId),
+    Read(CatalogItemId),
+}
+
+impl StatementSource {
+    pub fn id(&self) -> CatalogItemId {
+        match self {
+            StatementSource::Altered(id) | StatementSource::Read(id) => *id,
+        }
+    }
+}
+
+/// Resolves the existing source that `stmt` drives, mirroring how the
+/// purification paths below resolve it themselves.
+///
+/// `None` means the statement drives no such source: it names its connection
+/// outright (`CREATE SOURCE`, `CREATE SINK`), or the name does not resolve to a
+/// source. Purification reports the latter itself, and bails on it before
+/// opening any connection. A [`Statement`] variant that instead reaches
+/// upstream through an existing catalog item needs an arm here, or callers see
+/// no source at all.
+pub fn statement_source(
+    catalog: &impl SessionCatalog,
+    stmt: &Statement<Aug>,
+) -> Option<StatementSource> {
+    let scx = StatementContext::new(None, catalog);
+    match stmt {
+        Statement::AlterSource(stmt) => {
+            let item = scx
+                .resolve_item(RawItemName::Name(stmt.source_name.clone()))
+                .ok()?;
+            (item.item_type() == CatalogItemType::Source)
+                .then(|| StatementSource::Altered(item.id()))
+        }
+        Statement::CreateTableFromSource(stmt) => {
+            let item = scx.get_item_by_resolved_name(&stmt.source).ok()?;
+            (item.item_type() == CatalogItemType::Source).then(|| StatementSource::Read(item.id()))
+        }
+        _ => None,
+    }
+}
+
 /// Purifies a statement, removing any dependencies on external state.
 ///
 /// See the section on [purification](crate#purification) in the crate

src/sql/src/rbac.rs

@@ -34,6 +34,7 @@ use crate::plan::{
     DataSourceDesc, Explainee, MutationKind, Plan, SideEffectingFunc, TableDataSource,
     UpdatePrivilege,
 };
+use crate::pure::StatementSource;
 use crate::session::metadata::SessionMetadata;
 use crate::session::user::{MZ_SUPPORT_ROLE_ID, MZ_SYSTEM_ROLE_ID, SUPPORT_USER, SYSTEM_USER};
 use crate::session::vars::SystemVars;
@@ -398,6 +399,43 @@ pub fn check_usage(
     Ok(())
 }
 
+/// Authorizes a statement for purification, which runs before planning and so
+/// escapes the [`check_plan`] gate. `source` is the existing source the
+/// statement would drive, as resolved by [`crate::pure::statement_source`].
+///
+/// The requirements are the same ones planning enforces later, so a statement
+/// that passes here can still be rejected by [`check_plan`], never the reverse.
+pub fn check_purification(
+    catalog: &impl SessionCatalog,
+    session: &dyn SessionMetadata,
+    source: Option<StatementSource>,
+    resolved_ids: &ResolvedIds,
+) -> Result<(), UnauthorizedError> {
+    // Like `check_plan`: `validate` reads the current role's membership through
+    // the panicking `get_role`, so the concurrent-role-drop case must be turned
+    // into a clean error before anything else runs.
+    rbac_check_preamble(catalog, session)?;
+
+    let role_id = session.role_metadata().current_role;
+    let mut requirements = RbacRequirements {
+        item_usage: &CREATE_ITEM_USAGE,
+        ..Default::default()
+    };
+    match source {
+        Some(StatementSource::Altered(id)) => {
+            requirements.ownership = vec![ObjectId::Item(id)];
+        }
+        Some(StatementSource::Read(id)) => {
+            requirements.privileges = generate_read_privileges(catalog, iter::once(id), role_id);
+        }
+        // Statements that name their connection are covered by `item_usage`.
+        None => {}
+    }
+
+    let requirements = filter_requirements(catalog, session, requirements);
+    requirements.validate(catalog, session, resolved_ids)
+}
+
 /// Checks if a session is authorized to execute a plan. If not, an error is returned.
 ///
 /// `sql_impl_resolved_ids` contains resolved IDs discovered inside SQL-implemented function
@@ -645,7 +683,9 @@ fn generate_rbac_requirements(
                 role_id,
             )];
             // `CREATE TABLE ... FROM SOURCE` reads the source's data, so it
-            // requires `SELECT` on the source.
+            // requires `SELECT` on the source. `check_purification` enforces
+            // the same read requirement before purification; keep them in sync
+            // or its subset invariant inverts.
             if let TableDataSource::DataSource {
                 desc: DataSourceDesc::IngestionExport { ingestion_id, .. },
                 timeline: _,
@@ -1169,6 +1209,8 @@ fn generate_rbac_requirements(
             ownership: vec![ObjectId::Item(*id)],
             ..Default::default()
         },
+        // `check_purification` enforces the same ownership requirement before
+        // purification; keep them in sync or its subset invariant inverts.
         Plan::AlterSource(plan::AlterSourcePlan {
             item_id,
             ingestion_id: _,

test/sqllogictest/rbac_create_table_from_source.slt

@@ -10,6 +10,9 @@
 # `CREATE TABLE ... FROM SOURCE` reads the source's data, so it requires the
 # same read privileges as reading the source directly: `SELECT` on the source
 # and `USAGE` on its schema.
+#
+# Those privileges are also required *before* purification, which runs ahead of
+# planning and borrows the source's own connection to contact its upstream.
 
 mode cockroach
 
@@ -165,3 +168,115 @@ simple conn=other_team,user=other_team
 CREATE TABLE other_team_schema.mine FROM SOURCE victim_schema.victim_auction (REFERENCE "auction"."users");
 ----
 COMPLETE 0
+
+# Purification runs before planning and reaches the source's upstream, so the
+# read requirement is enforced before it, not only at plan time.
+#
+# An unresolvable reference is the discriminator: purification reports whether a
+# reference exists upstream, so if it ran first this would return
+# "reference to ... not found" and leak that fact to a role with no privilege on
+# the source. A permission error means the check fired first.
+
+simple conn=mz_system,user=mz_system
+REVOKE SELECT ON victim_schema.victim_auction FROM attacker;
+----
+COMPLETE 0
+
+simple conn=attacker,user=attacker
+CREATE TABLE attacker_schema.probe FROM SOURCE victim_schema.victim_auction (REFERENCE "auction"."does_not_exist");
+----
+db error: ERROR: permission denied for SOURCE "materialize.victim_schema.victim_auction"
+DETAIL: The 'attacker' role needs SELECT privileges on SOURCE "materialize.victim_schema.victim_auction"
+
+# `ALTER SOURCE` reaches the same purification path but carries its target as an
+# unresolved name, so it is not covered by the resolved-id check and needs the
+# source resolved from the statement. It mutates the source, so it requires
+# ownership.
+#
+# `ADD SUBSOURCE` is the discriminator here: purification rejects it on a load
+# generator with "does not support ALTER SOURCE", so an ownership error can only
+# come from the earlier gate. `REFRESH REFERENCES` would not discriminate, since
+# it purifies successfully on a load generator and plan-time ownership produces
+# the identical message.
+
+simple conn=attacker,user=attacker
+ALTER SOURCE victim_schema.victim_auction ADD SUBSOURCE t;
+----
+db error: ERROR: must be owner of SOURCE materialize.victim_schema.victim_auction
+
+# The owner is unaffected on both paths.
+
+simple conn=mz_system,user=mz_system
+ALTER SOURCE victim_schema.victim_auction OWNER TO attacker;
+----
+COMPLETE 0
+
+simple conn=attacker,user=attacker
+ALTER SOURCE victim_schema.victim_auction REFRESH REFERENCES;
+----
+COMPLETE 0
+
+# Ownership does not stand in for read privileges: an owner's `SELECT` is an
+# ordinary revocable grant, and schema `USAGE` is separate from ownership, so
+# the pre-purification check must require the read even from the owner. The
+# attacker owns the source here but has had `SELECT` revoked; the invalid
+# reference again discriminates, since a reference-not-found error would mean
+# purification ran and leaked reference existence to a role planning would
+# reject.
+
+simple conn=mz_system,user=mz_system
+REVOKE SELECT ON victim_schema.victim_auction FROM attacker;
+----
+COMPLETE 0
+
+simple conn=attacker,user=attacker
+CREATE TABLE attacker_schema.owner_probe FROM SOURCE victim_schema.victim_auction (REFERENCE "auction"."does_not_exist");
+----
+db error: ERROR: permission denied for SOURCE "materialize.victim_schema.victim_auction"
+DETAIL: The 'attacker' role needs SELECT privileges on SOURCE "materialize.victim_schema.victim_auction"
+
+# Schema `USAGE` is the other half of the read requirement. The attacker gets
+# `SELECT` back but loses `USAGE` on the schema; the denial must move to the
+# schema, again ahead of purification.
+
+simple conn=mz_system,user=mz_system
+GRANT SELECT ON victim_schema.victim_auction TO attacker;
+----
+COMPLETE 0
+
+simple conn=mz_system,user=mz_system
+REVOKE USAGE ON SCHEMA victim_schema FROM attacker;
+----
+COMPLETE 0
+
+simple conn=attacker,user=attacker
+CREATE TABLE attacker_schema.schema_probe FROM SOURCE victim_schema.victim_auction (REFERENCE "auction"."does_not_exist");
+----
+db error: ERROR: permission denied for SCHEMA "materialize.victim_schema"
+DETAIL: The 'attacker' role needs USAGE privileges on SCHEMA "materialize.victim_schema"
+
+# Superusers pass through the gate: mz_system does not own the source (the
+# attacker does), so reaching purification's "does not support ALTER SOURCE"
+# error proves the ownership requirement was filtered, not enforced.
+
+simple conn=mz_system,user=mz_system
+ALTER SOURCE victim_schema.victim_auction ADD SUBSOURCE t;
+----
+db error: ERROR: source victim_schema.victim_auction does not support ALTER SOURCE.
+
+# With RBAC checks disabled, a non-owner passes through the same way.
+
+simple conn=mz_system,user=mz_system
+ALTER SYSTEM SET enable_rbac_checks TO false;
+----
+COMPLETE 0
+
+simple conn=no_create,user=no_create
+ALTER SOURCE victim_schema.victim_auction ADD SUBSOURCE t;
+----
+db error: ERROR: source victim_schema.victim_auction does not support ALTER SOURCE.
+
+simple conn=mz_system,user=mz_system
+ALTER SYSTEM SET enable_rbac_checks TO true;
+----
+COMPLETE 0