[teams 4/5] SQL authorization (#3525)

Permissions for evaluating SQL/DML are not generally "actions", but more
a set of permissions that are checked during evaluation.

To make this work with the teams feature, this patch extends `AuthCtx`
to allow checking a set of permissions as mandated by the spec. This set
is a bit more fine-grained than "is owner", so as to avoid baking in the
concept of teams/collaborators, or assumptions about what a role might
entail. Both are likely to evolve in the future, so evaluation of
permissions / capabilities should be confined to the impl of the
`Authorization` trait.

Unlike "actions", the `AuthCtx` must be able to evaluate permission
checks quickly and without side-effects, nor can it enter an `async`
context. In that sense, it is precomputed (if you will), and stored as a
closure in the `AuthCtx` for external authorization.

A challenge posed is how to thread through the constructed `AuthCtx` for
subscriptions.

A tempting approach would have been to equip the `HostController` with
the ability to summon an `AuthCtx`. That, however, would have created a
gnarly circular dependency, because the `HostController` also controls
the controldb, which itself demands an `AuthCtx`.

Instead, the `AuthCtx` is obtained in the endpoint handler and passed to
each method call that requires one. That's less pretty, but more
effective.

---------

Signed-off-by: Kim Altintop <kim@eagain.io>
Co-authored-by: Phoebe Goldman <phoebe@clockworklabs.io>

Author: kim

crates/client-api/src/lib.rs

@@ -505,6 +505,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> {
@@ -516,6 +530,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

@@ -29,7 +29,7 @@ use spacetimedb::host::UpdateDatabaseResult;
 use spacetimedb::host::{FunctionArgs, MigratePlanResult};
 use spacetimedb::host::{ModuleHost, ReducerOutcome};
 use spacetimedb::host::{ProcedureCallError, ReducerCallError};
-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::{
@@ -530,15 +530,16 @@ 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.
 
     let (host, database) = find_leader_and_database(&worker_ctx, name_or_identity).await?;
 
-    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?;
 
     host.exec_sql(auth, database, confirmed, sql).await
 }
@@ -551,7 +552,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

@@ -29,7 +29,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};
@@ -424,6 +424,7 @@ pub struct ClientConnection {
     sender: Arc<ClientConnectionSender>,
     pub replica_id: u64,
     module_rx: watch::Receiver<ModuleHost>,
+    auth: AuthCtx,
 }
 
 impl Deref for ClientConnection {
@@ -675,9 +676,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>,
@@ -735,6 +738,7 @@ impl ClientConnection {
             sender,
             replica_id,
             module_rx,
+            auth: sql_auth,
         };
 
         let actor_fut = actor(this.clone(), receiver);
@@ -750,10 +754,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,
         }
     }
 
@@ -868,7 +874,7 @@ impl ClientConnection {
             .on_module_thread_async("subscribe_single", async move || {
                 let host = me.module();
                 host.subscriptions()
-                    .add_single_subscription(Some(&host), me.sender, subscription, timer, None)
+                    .add_single_subscription(Some(&host), me.sender, me.auth.clone(), subscription, timer, None)
                     .await
             })
             .await?
@@ -879,7 +885,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
     }
@@ -894,7 +900,7 @@ impl ClientConnection {
             .on_module_thread_async("subscribe_multi", async move || {
                 let host = me.module();
                 host.subscriptions()
-                    .add_multi_subscription(Some(&host), me.sender, request, timer, None)
+                    .add_multi_subscription(Some(&host), me.sender, me.auth.clone(), request, timer, None)
                     .await
             })
             .await?
@@ -910,7 +916,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?
     }
@@ -921,7 +927,7 @@ impl ClientConnection {
             .on_module_thread_async("subscribe", async move || {
                 let host = me.module();
                 host.subscriptions()
-                    .add_legacy_subscriber(Some(&host), me.sender, subscription, timer, None)
+                    .add_legacy_subscriber(Some(&host), me.sender, me.auth.clone(), subscription, timer, None)
                     .await
             })
             .await?
@@ -935,7 +941,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(),
@@ -953,7 +959,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

@@ -629,6 +629,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

@@ -553,12 +553,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

@@ -1715,7 +1715,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>,
@@ -1726,7 +1726,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<TableOrViewSchema>> {
-        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))
             .map(Arc::clone)
             .map(TableOrViewSchema::from)
             .map(Arc::new)