Teams-based authorization for SQL / subscriptions

Author: kim

crates/client-api/src/lib.rs

@@ -490,6 +490,20 @@ pub trait Authorization {
         database: Identity,
         action: Action,
     ) -> impl Future<Output = Result<(), Unauthorized>> + Send;
+
+    /// Obtain an attenuated [AuthCtx] for `subject` to evaluate SQL against
+    /// `database`.
+    ///
+    /// "SQL" includes the sql endpoint, pg wire connections, as well as
+    /// subscription queries.
+    ///
+    /// If any SQL should be rejected outright, or the authorization database
+    /// is not available, return `Err(Unauthorized)`.
+    fn authorize_sql(
+        &self,
+        subject: Identity,
+        database: Identity,
+    ) -> impl Future<Output = Result<AuthCtx, Unauthorized>> + Send;
 }
 
 impl<T: Authorization> Authorization for Arc<T> {
@@ -501,6 +515,14 @@ impl<T: Authorization> Authorization for Arc<T> {
     ) -> impl Future<Output = Result<(), Unauthorized>> + Send {
         (**self).authorize_action(subject, database, action)
     }
+
+    fn authorize_sql(
+        &self,
+        subject: Identity,
+        database: Identity,
+    ) -> impl Future<Output = Result<AuthCtx, Unauthorized>> + Send {
+        (**self).authorize_sql(subject, database)
+    }
 }
 
 pub fn log_and_500(e: impl std::fmt::Display) -> ErrorResponse {

crates/client-api/src/routes/database.rs

@@ -27,7 +27,7 @@ use spacetimedb::host::module_host::ClientConnectedError;
 use spacetimedb::host::ReducerOutcome;
 use spacetimedb::host::{FunctionArgs, MigratePlanResult};
 use spacetimedb::host::{ReducerCallError, UpdateDatabaseResult};
-use spacetimedb::identity::{AuthCtx, Identity};
+use spacetimedb::identity::Identity;
 use spacetimedb::messages::control_db::{Database, HostType};
 use spacetimedb_client_api_messages::http::SqlStmtResult;
 use spacetimedb_client_api_messages::name::{
@@ -420,7 +420,7 @@ pub async fn sql_direct<S>(
     sql: String,
 ) -> axum::response::Result<Vec<SqlStmtResult<ProductValue>>>
 where
-    S: NodeDelegate + ControlStateDelegate,
+    S: NodeDelegate + ControlStateDelegate + Authorization,
 {
     // Anyone is authorized to execute SQL queries. The SQL engine will determine
     // which queries this identity is allowed to execute against the database.
@@ -430,8 +430,9 @@ where
         .await?
         .ok_or(NO_SUCH_DATABASE)?;
 
-    let auth = AuthCtx::new(database.owner_identity, caller_identity);
-    log::debug!("auth: {auth:?}");
+    let auth = worker_ctx
+        .authorize_sql(caller_identity, database.database_identity)
+        .await?;
 
     let host = worker_ctx
         .leader(database.id)
@@ -450,7 +451,7 @@ pub async fn sql<S>(
     body: String,
 ) -> axum::response::Result<impl IntoResponse>
 where
-    S: NodeDelegate + ControlStateDelegate,
+    S: NodeDelegate + ControlStateDelegate + Authorization,
 {
     let json = sql_direct(worker_ctx, name_or_identity, params, auth.claims.identity, body).await?;
 

crates/client-api/src/routes/subscribe.rs

@@ -49,7 +49,7 @@ use crate::util::websocket::{
     CloseCode, CloseFrame, Message as WsMessage, WebSocketConfig, WebSocketStream, WebSocketUpgrade, WsError,
 };
 use crate::util::{NameOrIdentity, XForwardedFor};
-use crate::{log_and_500, ControlStateDelegate, NodeDelegate};
+use crate::{log_and_500, Authorization, ControlStateDelegate, NodeDelegate};
 
 #[allow(clippy::declare_interior_mutable_const)]
 pub const TEXT_PROTOCOL: HeaderValue = HeaderValue::from_static(ws_api::TEXT_PROTOCOL);
@@ -106,7 +106,7 @@ pub async fn handle_websocket<S>(
     ws: WebSocketUpgrade,
 ) -> axum::response::Result<impl IntoResponse>
 where
-    S: NodeDelegate + ControlStateDelegate + HasWebSocketOptions,
+    S: NodeDelegate + ControlStateDelegate + HasWebSocketOptions + Authorization,
 {
     if connection_id.is_some() {
         // TODO: Bump this up to `log::warn!` after removing the client SDKs' uses of that parameter.
@@ -125,6 +125,7 @@ where
     }
 
     let db_identity = name_or_identity.resolve(&ctx).await?;
+    let sql_auth = ctx.authorize_sql(auth.claims.identity, db_identity).await?;
 
     let (res, ws_upgrade, protocol) =
         ws.select_protocol([(BIN_PROTOCOL, Protocol::Binary), (TEXT_PROTOCOL, Protocol::Text)]);
@@ -218,6 +219,7 @@ where
         let client = ClientConnection::spawn(
             client_id,
             auth.into(),
+            sql_auth,
             client_config,
             leader.replica_id,
             module_rx,

crates/core/src/client/client_connection.rs

@@ -28,7 +28,7 @@ use spacetimedb_client_api_messages::websocket::{
     UnsubscribeMulti,
 };
 use spacetimedb_durability::{DurableOffset, TxOffset};
-use spacetimedb_lib::identity::RequestId;
+use spacetimedb_lib::identity::{AuthCtx, RequestId};
 use spacetimedb_lib::metrics::ExecutionMetrics;
 use spacetimedb_lib::Identity;
 use tokio::sync::mpsc::error::{SendError, TrySendError};
@@ -423,6 +423,7 @@ pub struct ClientConnection {
     sender: Arc<ClientConnectionSender>,
     pub replica_id: u64,
     module_rx: watch::Receiver<ModuleHost>,
+    auth: AuthCtx,
 }
 
 impl Deref for ClientConnection {
@@ -674,9 +675,11 @@ impl ClientConnection {
     /// to verify that the database at `module_rx` approves of this connection,
     /// and should not invoke this method if that call returns an error,
     /// and pass the returned [`Connected`] as `_proof_of_client_connected_call`.
+    #[allow(clippy::too_many_arguments)]
     pub async fn spawn<Fut>(
         id: ClientActorId,
         auth: ConnectionAuthCtx,
+        sql_auth: AuthCtx,
         config: ClientConfig,
         replica_id: u64,
         mut module_rx: watch::Receiver<ModuleHost>,
@@ -734,6 +737,7 @@ impl ClientConnection {
             sender,
             replica_id,
             module_rx,
+            auth: sql_auth,
         };
 
         let actor_fut = actor(this.clone(), receiver);
@@ -749,10 +753,12 @@ impl ClientConnection {
         replica_id: u64,
         module_rx: watch::Receiver<ModuleHost>,
     ) -> Self {
+        let auth = AuthCtx::new(module_rx.borrow().database_info().database_identity, id.identity);
         Self {
             sender: Arc::new(ClientConnectionSender::dummy(id, config, module_rx.clone())),
             replica_id,
             module_rx,
+            auth,
         }
     }
 
@@ -842,9 +848,13 @@ impl ClientConnection {
         let me = self.clone();
         self.module()
             .on_module_thread("subscribe_single", move || {
-                me.module()
-                    .subscriptions()
-                    .add_single_subscription(me.sender, subscription, timer, None)
+                me.module().subscriptions().add_single_subscription(
+                    me.sender,
+                    me.auth.clone(),
+                    subscription,
+                    timer,
+                    None,
+                )
             })
             .await?
     }
@@ -854,7 +864,7 @@ impl ClientConnection {
         asyncify(move || {
             me.module()
                 .subscriptions()
-                .remove_single_subscription(me.sender, request, timer)
+                .remove_single_subscription(me.sender, me.auth.clone(), request, timer)
         })
         .await
     }
@@ -869,7 +879,7 @@ impl ClientConnection {
             .on_module_thread("subscribe_multi", move || {
                 me.module()
                     .subscriptions()
-                    .add_multi_subscription(me.sender, request, timer, None)
+                    .add_multi_subscription(me.sender, me.auth.clone(), request, timer, None)
             })
             .await?
     }
@@ -884,7 +894,7 @@ impl ClientConnection {
             .on_module_thread("unsubscribe_multi", move || {
                 me.module()
                     .subscriptions()
-                    .remove_multi_subscription(me.sender, request, timer)
+                    .remove_multi_subscription(me.sender, me.auth.clone(), request, timer)
             })
             .await?
     }
@@ -894,7 +904,7 @@ impl ClientConnection {
         asyncify(move || {
             me.module()
                 .subscriptions()
-                .add_legacy_subscriber(me.sender, subscription, timer, None)
+                .add_legacy_subscriber(me.sender, me.auth.clone(), subscription, timer, None)
         })
         .await
     }
@@ -907,7 +917,7 @@ impl ClientConnection {
     ) -> Result<(), anyhow::Error> {
         self.module()
             .one_off_query::<JsonFormat>(
-                self.id.identity,
+                self.auth.clone(),
                 query.to_owned(),
                 self.sender.clone(),
                 message_id.to_owned(),
@@ -925,7 +935,7 @@ impl ClientConnection {
     ) -> Result<(), anyhow::Error> {
         self.module()
             .one_off_query::<BsatnFormat>(
-                self.id.identity,
+                self.auth.clone(),
                 query.to_owned(),
                 self.sender.clone(),
                 message_id.to_owned(),

crates/core/src/db/relational_db.rs

@@ -624,6 +624,10 @@ impl RelationalDB {
         self.database_identity
     }
 
+    pub fn owner_identity(&self) -> Identity {
+        self.owner_identity
+    }
+
     /// The number of bytes on disk occupied by the durability layer.
     ///
     /// If this is an in-memory instance, `Ok(0)` is returned.

crates/core/src/host/host_controller.rs

@@ -545,12 +545,7 @@ async fn make_replica_ctx(
         send_worker_queue.clone(),
     )));
     let downgraded = Arc::downgrade(&subscriptions);
-    let subscriptions = ModuleSubscriptions::new(
-        relational_db.clone(),
-        subscriptions,
-        send_worker_queue,
-        database.owner_identity,
-    );
+    let subscriptions = ModuleSubscriptions::new(relational_db.clone(), subscriptions, send_worker_queue);
 
     // If an error occurs when evaluating a subscription,
     // we mark each client that was affected,

crates/core/src/host/module_host.rs

@@ -1329,7 +1329,7 @@ impl ModuleHost {
     #[tracing::instrument(level = "trace", skip_all)]
     pub async fn one_off_query<F: BuildableWebsocketFormat>(
         &self,
-        caller_identity: Identity,
+        auth: AuthCtx,
         query: String,
         client: Arc<ClientConnectionSender>,
         message_id: Vec<u8>,
@@ -1340,7 +1340,6 @@ impl ModuleHost {
         let replica_ctx = self.replica_ctx();
         let db = replica_ctx.relational_db.clone();
         let subscriptions = replica_ctx.subscriptions.clone();
-        let auth = AuthCtx::new(replica_ctx.owner_identity, caller_identity);
         log::debug!("One-off query: {query}");
         let metrics = self
             .on_module_thread("one_off_query", move || {

crates/core/src/sql/ast.rs

@@ -6,7 +6,6 @@ use spacetimedb_datastore::locking_tx_datastore::state_view::StateView;
 use spacetimedb_datastore::system_tables::{StRowLevelSecurityFields, ST_ROW_LEVEL_SECURITY_ID};
 use spacetimedb_expr::check::SchemaView;
 use spacetimedb_expr::statement::compile_sql_stmt;
-use spacetimedb_lib::db::auth::StAccess;
 use spacetimedb_lib::identity::AuthCtx;
 use spacetimedb_primitives::{ColId, TableId};
 use spacetimedb_sats::{AlgebraicType, AlgebraicValue};
@@ -492,22 +491,20 @@ impl<T> Deref for SchemaViewer<'_, T> {
 
 impl<T: StateView> SchemaView for SchemaViewer<'_, T> {
     fn table_id(&self, name: &str) -> Option<TableId> {
-        let AuthCtx { owner, caller } = self.auth;
         // Get the schema from the in-memory state instead of fetching from the database for speed
         self.tx
             .table_id_from_name(name)
             .ok()
             .flatten()
             .and_then(|table_id| self.schema_for_table(table_id))
-            .filter(|schema| schema.table_access == StAccess::Public || caller == owner)
+            .filter(|schema| self.auth.has_read_access(schema.table_access))
             .map(|schema| schema.table_id)
     }
 
     fn schema_for_table(&self, table_id: TableId) -> Option<Arc<TableSchema>> {
-        let AuthCtx { owner, caller } = self.auth;
         self.tx
             .get_schema(table_id)
-            .filter(|schema| schema.table_access == StAccess::Public || caller == owner)
+            .filter(|schema| self.auth.has_read_access(schema.table_access))
             .cloned()
     }
 

crates/core/src/sql/execute.rs

@@ -122,15 +122,15 @@ pub fn execute_sql(
         let mut tx = db.begin_mut_tx(IsolationLevel::Serializable, Workload::Sql);
         let mut updates = Vec::with_capacity(ast.len());
         let res = execute(
-            &mut DbProgram::new(db, &mut (&mut tx).into(), auth),
+            &mut DbProgram::new(db, &mut (&mut tx).into(), auth.clone()),
             ast,
             sql,
             &mut updates,
         );
         if res.is_ok() && !updates.is_empty() {
             let event = ModuleEvent {
                 timestamp: Timestamp::now(),
-                caller_identity: auth.caller,
+                caller_identity: auth.caller(),
                 caller_connection_id: None,
                 function_call: ModuleFunctionCall {
                     reducer: String::new(),
@@ -249,7 +249,7 @@ pub fn run(
         }
         Statement::DML(stmt) => {
             // An extra layer of auth is required for DML
-            if auth.caller != auth.owner {
+            if !auth.has_write_access() {
                 return Err(anyhow!("Only owners are authorized to run SQL DML statements").into());
             }
 
@@ -287,7 +287,7 @@ pub fn run(
                     None,
                     ModuleEvent {
                         timestamp: Timestamp::now(),
-                        caller_identity: auth.caller,
+                        caller_identity: auth.caller(),
                         caller_connection_id: None,
                         function_call: ModuleFunctionCall {
                             reducer: String::new(),
@@ -510,7 +510,7 @@ pub(crate) mod tests {
         expected: impl IntoIterator<Item = ProductValue>,
     ) {
         assert_eq!(
-            run(db, sql, *auth, None, &mut vec![])
+            run(db, sql, auth.clone(), None, &mut vec![])
                 .unwrap()
                 .rows
                 .into_iter()
@@ -1270,19 +1270,25 @@ pub(crate) mod tests {
         let run = |db, sql, auth, subs| run(db, sql, auth, subs, &mut vec![]);
 
         // No row limit, both queries pass.
-        assert!(run(&db, "SELECT * FROM T", internal_auth, None).is_ok());
-        assert!(run(&db, "SELECT * FROM T", external_auth, None).is_ok());
+        assert!(run(&db, "SELECT * FROM T", internal_auth.clone(), None).is_ok());
+        assert!(run(&db, "SELECT * FROM T", external_auth.clone(), None).is_ok());
 
         // Set row limit.
-        assert!(run(&db, "SET row_limit = 4", internal_auth, None).is_ok());
+        assert!(run(&db, "SET row_limit = 4", internal_auth.clone(), None).is_ok());
 
         // External query fails.
-        assert!(run(&db, "SELECT * FROM T", internal_auth, None).is_ok());
-        assert!(run(&db, "SELECT * FROM T", external_auth, None).is_err());
+        assert!(run(&db, "SELECT * FROM T", internal_auth.clone(), None).is_ok());
+        assert!(run(&db, "SELECT * FROM T", external_auth.clone(), None).is_err());
 
         // Increase row limit.
-        assert!(run(&db, "DELETE FROM st_var WHERE name = 'row_limit'", internal_auth, None).is_ok());
-        assert!(run(&db, "SET row_limit = 5", internal_auth, None).is_ok());
+        assert!(run(
+            &db,
+            "DELETE FROM st_var WHERE name = 'row_limit'",
+            internal_auth.clone(),
+            None
+        )
+        .is_ok());
+        assert!(run(&db, "SET row_limit = 5", internal_auth.clone(), None).is_ok());
 
         // Both queries pass.
         assert!(run(&db, "SELECT * FROM T", internal_auth, None).is_ok());
@@ -1333,10 +1339,10 @@ pub(crate) mod tests {
             ..ExecutionMetrics::default()
         };
 
-        check(&db, "INSERT INTO T (a) VALUES (5)", internal_auth, ins)?;
-        check(&db, "UPDATE T SET a = 2", internal_auth, upd)?;
+        check(&db, "INSERT INTO T (a) VALUES (5)", internal_auth.clone(), ins)?;
+        check(&db, "UPDATE T SET a = 2", internal_auth.clone(), upd)?;
         assert_eq!(
-            run(&db, "SELECT * FROM T", internal_auth, None)?.rows,
+            run(&db, "SELECT * FROM T", internal_auth.clone(), None)?.rows,
             vec![product!(2u8)]
         );
         check(&db, "DELETE FROM T", internal_auth, del)?;