security(authz): require capabilities at dispatch boundaries

Author: farhan-syah

.config/nextest.toml

@@ -51,6 +51,18 @@ retries = { backoff = "fixed", count = 2, delay = "1s" }
 # default 30s × 4 = 120s ceiling stays in force for unit tests.
 slow-timeout = { period = "30s", terminate-after = 6 }
 
+# Load-sensitive in-process integrations use full Data-Plane runtimes and
+# bounded request deadlines. Running them beside the unrestricted unit-test
+# fan-out can starve their executor threads long enough to turn a healthy local
+# dispatch into DeadlineExceeded. Claim every slot for only these measured
+# heavy cases; this preserves real request deadlines rather than weakening them.
+# The crash-recovery label test is included because its child server otherwise
+# competes with the same in-process graph fan-out during the full suite.
+[[profile.default.overrides]]
+filter = 'binary(graph_cross_core_bfs) | test(query_stream_returns_all_rows_across_chunks) | test(streaming_select_returns_all_rows_across_chunks) | test(graph_node_labels_survive_kill_9) | test(wal_replay_restart_keeps_governor_accounting_sane) | test(alter_add_column_does_not_stall_writes) | test(reindex_vector_concurrent_p99) | test(count_star_sees_flushed_partitions) | test(timeseries_flush_does_not_over_release_engine_budget) | test(timeseries_governor_reflects_resident_memtable_after_flush)'
+test-group = 'load-sensitive'
+threads-required = 'num-test-threads'
+
 # WAL-truncation crash tests wait on real wall-clock server behaviour that the
 # other real-server tests do not: a checkpoint cycle must fire AND actually
 # unlink a sealed segment before the test can crash the server, on top of two
@@ -96,6 +108,7 @@ test-group = 'server-process'
 
 [test-groups]
 cluster = { max-threads = 1 }
+load-sensitive = { max-threads = 1 }
 # Up to 2 real-server tests concurrently. Each picks ephemeral (OS-assigned)
 # ports + a private tempdir, so two running side by side never collide on
 # ports/files; 2 single-core servers fit any CI runner, and CI `retries`

.github/workflows/test.yml

@@ -68,6 +68,10 @@ jobs:
         run: |
           python3 scripts/ci/check_reconstructed_sql.py --self-test
           python3 scripts/ci/check_reconstructed_sql.py
+      - name: Authorized-dispatch gate
+        run: |
+          python3 scripts/ci/check_authorized_dispatch.py --self-test
+          python3 scripts/ci/check_authorized_dispatch.py
       - name: Install cargo-deny
         uses: taiki-e/install-action@v2
         with:

nodedb-cluster-tests/tests/array_raft_partition_recovery.rs

@@ -70,7 +70,19 @@ fn make_inbound(shared: &Arc<SharedState>) -> OriginArrayInbound {
         engine,
         Arc::clone(&shared.array_sync_schemas),
         Arc::clone(shared),
-        TenantId::new(0),
+        nodedb::control::security::identity::AuthenticatedIdentity {
+            user_id: 0,
+            username: "array-sync-test".into(),
+            tenant_id: TenantId::new(0),
+            auth_method: nodedb::control::security::identity::AuthMethod::Trust,
+            roles: Vec::new(),
+            is_superuser: true,
+            default_database: None,
+            accessible_databases:
+                nodedb::control::security::identity::AuthenticatedIdentity::default_database_set(
+                    true,
+                ),
+        },
     )
 }
 

nodedb-cluster-tests/tests/array_raft_replication.rs

@@ -79,7 +79,19 @@ fn make_inbound(shared: &Arc<SharedState>) -> OriginArrayInbound {
         engine,
         Arc::clone(&shared.array_sync_schemas),
         Arc::clone(shared),
-        TenantId::new(0),
+        nodedb::control::security::identity::AuthenticatedIdentity {
+            user_id: 0,
+            username: "array-sync-test".into(),
+            tenant_id: TenantId::new(0),
+            auth_method: nodedb::control::security::identity::AuthMethod::Trust,
+            roles: Vec::new(),
+            is_superuser: true,
+            default_database: None,
+            accessible_databases:
+                nodedb::control::security::identity::AuthenticatedIdentity::default_database_set(
+                    true,
+                ),
+        },
     )
 }
 

nodedb-cluster-tests/tests/array_raft_snapshot_install.rs

@@ -72,7 +72,19 @@ fn make_inbound(shared: &Arc<SharedState>) -> OriginArrayInbound {
         engine,
         Arc::clone(&shared.array_sync_schemas),
         Arc::clone(shared),
-        TenantId::new(0),
+        nodedb::control::security::identity::AuthenticatedIdentity {
+            user_id: 0,
+            username: "array-sync-test".into(),
+            tenant_id: TenantId::new(0),
+            auth_method: nodedb::control::security::identity::AuthMethod::Trust,
+            roles: Vec::new(),
+            is_superuser: true,
+            default_database: None,
+            accessible_databases:
+                nodedb::control::security::identity::AuthenticatedIdentity::default_database_set(
+                    true,
+                ),
+        },
     )
 }
 

nodedb-cluster-tests/tests/array_vshard_migration.rs

@@ -71,7 +71,19 @@ fn make_inbound(shared: &Arc<SharedState>) -> OriginArrayInbound {
         engine,
         Arc::clone(&shared.array_sync_schemas),
         Arc::clone(shared),
-        TenantId::new(0),
+        nodedb::control::security::identity::AuthenticatedIdentity {
+            user_id: 0,
+            username: "array-sync-test".into(),
+            tenant_id: TenantId::new(0),
+            auth_method: nodedb::control::security::identity::AuthMethod::Trust,
+            roles: Vec::new(),
+            is_superuser: true,
+            default_database: None,
+            accessible_databases:
+                nodedb::control::security::identity::AuthenticatedIdentity::default_database_set(
+                    true,
+                ),
+        },
     )
 }
 

nodedb-cluster-tests/tests/common/mod.rs

@@ -9,3 +9,48 @@ pub use nodedb_test_support::{
     array_sync, cluster_harness, make_cdc_event, now_ms, occ_shuffle, pgwire_auth_helpers,
     pgwire_harness, tx_batch_helpers,
 };
+
+/// Mint the same exact-plan authorization capability required by production
+/// gateway entry points, using a trusted superuser identity scoped to `ctx`.
+#[allow(dead_code)]
+pub fn authorize_gateway_plan(
+    shared: &nodedb::control::state::SharedState,
+    ctx: &nodedb::control::gateway::core::QueryContext,
+    plan: nodedb_physical::physical_plan::PhysicalPlan,
+) -> nodedb::control::server::shared::authorization::AuthorizedTask {
+    use nodedb::control::security::audit::NoopAuditEmitter;
+    use nodedb::control::security::identity::{AuthMethod, AuthenticatedIdentity};
+    use nodedb::control::server::shared::authorization::authorize_task_set;
+    use nodedb_physical::physical_task::{PhysicalTask, PostSetOp};
+
+    let identity = AuthenticatedIdentity {
+        user_id: 0,
+        username: "cluster-gateway-test".into(),
+        tenant_id: ctx.tenant_id,
+        auth_method: AuthMethod::Trust,
+        roles: Vec::new(),
+        is_superuser: true,
+        default_database: Some(ctx.database_id),
+        accessible_databases: AuthenticatedIdentity::default_database_set(true),
+    };
+    let task = PhysicalTask {
+        tenant_id: ctx.tenant_id,
+        vshard_id: nodedb::types::VShardId::new(0),
+        database_id: ctx.database_id,
+        plan,
+        post_set_op: PostSetOp::None,
+        txn_id: ctx.txn_id,
+    };
+    authorize_task_set(
+        &identity,
+        std::slice::from_ref(&task),
+        &shared.permissions,
+        &shared.roles,
+        &NoopAuditEmitter,
+    )
+    .expect("authorize cluster gateway plan")
+    .into_tasks()
+    .into_iter()
+    .next()
+    .expect("one authorized cluster gateway plan")
+}

nodedb-cluster-tests/tests/gateway_execute.rs

@@ -74,7 +74,8 @@ async fn gateway_execute_kv_put_get_single_node() {
         ttl_ms: 0,
         surrogate: nodedb_types::Surrogate::ZERO,
     });
-    let put_result = gateway.execute(&ctx, put_plan).await;
+    let put_authorized = common::authorize_gateway_plan(&node.shared, &ctx, put_plan);
+    let put_result = gateway.execute(&ctx, put_authorized).await;
     assert!(
         put_result.is_ok(),
         "KvOp::Put failed: {:?}",
@@ -88,7 +89,8 @@ async fn gateway_execute_kv_put_get_single_node() {
         rls_filters: vec![],
         surrogate_ceiling: None,
     });
-    let get_result = gateway.execute(&ctx, get_plan).await;
+    let get_authorized = common::authorize_gateway_plan(&node.shared, &ctx, get_plan);
+    let get_result = gateway.execute(&ctx, get_authorized).await;
     assert!(
         get_result.is_ok(),
         "KvOp::Get failed: {:?}",
@@ -143,7 +145,13 @@ async fn gateway_execute_sql_plan_cache_populated() {
 
     // First call: cache miss — plan_fn is invoked; cache grows to 1.
     let _ = gateway
-        .execute_sql(&ctx, sql, &[], make_plan)
+        .execute_sql(&ctx, sql, &[], make_plan, |plan| {
+            Ok(common::authorize_gateway_plan(
+                &node.shared,
+                &ctx,
+                plan.clone(),
+            ))
+        })
         .await
         .expect("first execute_sql");
 
@@ -156,7 +164,13 @@ async fn gateway_execute_sql_plan_cache_populated() {
     // Second call with same SQL + same descriptor versions: the actual key is
     // identical, so insert is a no-op and len stays 1.
     let _ = gateway
-        .execute_sql(&ctx, sql, &[], make_plan)
+        .execute_sql(&ctx, sql, &[], make_plan, |plan| {
+            Ok(common::authorize_gateway_plan(
+                &node.shared,
+                &ctx,
+                plan.clone(),
+            ))
+        })
         .await
         .expect("second execute_sql");
 

nodedb-cluster-tests/tests/http_gateway_migration.rs

@@ -72,7 +72,8 @@ async fn http_gateway_migration_single_node_query() {
         ttl_ms: 0,
         surrogate: nodedb_types::Surrogate::ZERO,
     });
-    let put_result = gateway.execute(&ctx, put_plan).await;
+    let put_authorized = common::authorize_gateway_plan(&node.shared, &ctx, put_plan);
+    let put_result = gateway.execute(&ctx, put_authorized).await;
     assert!(
         put_result.is_ok(),
         "PUT via gateway failed: {:?}",
@@ -86,7 +87,8 @@ async fn http_gateway_migration_single_node_query() {
         rls_filters: vec![],
         surrogate_ceiling: None,
     });
-    let get_result = gateway.execute(&ctx, get_plan).await;
+    let get_authorized = common::authorize_gateway_plan(&node.shared, &ctx, get_plan);
+    let get_result = gateway.execute(&ctx, get_authorized).await;
     assert!(
         get_result.is_ok(),
         "GET via gateway failed: {:?}",
@@ -141,7 +143,8 @@ async fn http_gateway_migration_cross_node_query() {
         ttl_ms: 0,
         surrogate: nodedb_types::Surrogate::ZERO,
     });
-    let put_result = gateway.execute(&ctx, put_plan).await;
+    let put_authorized = common::authorize_gateway_plan(&follower.shared, &ctx, put_plan);
+    let put_result = gateway.execute(&ctx, put_authorized).await;
     assert!(
         put_result.is_ok(),
         "cross-node PUT via gateway failed: {:?}",
@@ -162,14 +165,26 @@ async fn http_gateway_migration_cross_node_query() {
 
     for i in 0..3u32 {
         let result = gateway
-            .execute_sql(&ctx, get_sql, &[], || {
-                Ok(PhysicalPlan::Kv(KvOp::Get {
-                    collection: "http_gw_cross_node".into(),
-                    key: b"cross-key".to_vec(),
-                    rls_filters: vec![],
-                    surrogate_ceiling: None,
-                }))
-            })
+            .execute_sql(
+                &ctx,
+                get_sql,
+                &[],
+                || {
+                    Ok(PhysicalPlan::Kv(KvOp::Get {
+                        collection: "http_gw_cross_node".into(),
+                        key: b"cross-key".to_vec(),
+                        rls_filters: vec![],
+                        surrogate_ceiling: None,
+                    }))
+                },
+                |plan| {
+                    Ok(common::authorize_gateway_plan(
+                        &follower.shared,
+                        &ctx,
+                        plan.clone(),
+                    ))
+                },
+            )
             .await;
         assert!(
             result.is_ok(),

nodedb-cluster-tests/tests/ilp_gateway_migration.rs

@@ -77,7 +77,8 @@ async fn ilp_gateway_migration_single_node_ingest() {
         surrogates: Vec::new(),
         provenance: None,
     });
-    let result = gw.execute(&ctx, plan).await;
+    let authorized = common::authorize_gateway_plan(&node.shared, &ctx, plan);
+    let result = gw.execute(&ctx, authorized).await;
     assert!(
         result.is_ok(),
         "gateway ILP ingest failed: {:?}",
@@ -122,7 +123,8 @@ async fn ilp_gateway_migration_cross_node_ingest() {
         surrogates: Vec::new(),
         provenance: None,
     });
-    let result1 = leader_gw.execute(&ctx, plan1).await;
+    let authorized1 = common::authorize_gateway_plan(&cluster.nodes[0].shared, &ctx, plan1);
+    let result1 = leader_gw.execute(&ctx, authorized1).await;
     assert!(
         result1.is_ok(),
         "node 1 (leader) ILP gateway ingest failed: {:?}",
@@ -145,7 +147,8 @@ async fn ilp_gateway_migration_cross_node_ingest() {
     });
     // Retry once on RetryableSchemaChanged: the descriptor may not yet be in
     // the follower catalog when the gateway snapshot was taken.
-    let result2 = match follower_gw.execute(&ctx, plan2).await {
+    let authorized2 = common::authorize_gateway_plan(&cluster.nodes[1].shared, &ctx, plan2);
+    let result2 = match follower_gw.execute(&ctx, authorized2).await {
         Err(nodedb::Error::RetryableSchemaChanged { .. }) => {
             tokio::time::sleep(Duration::from_millis(150)).await;
             let plan2b = PhysicalPlan::Timeseries(TimeseriesOp::Ingest {
@@ -156,7 +159,9 @@ async fn ilp_gateway_migration_cross_node_ingest() {
                 surrogates: Vec::new(),
                 provenance: None,
             });
-            follower_gw.execute(&ctx, plan2b).await
+            let authorized2b =
+                common::authorize_gateway_plan(&cluster.nodes[1].shared, &ctx, plan2b);
+            follower_gw.execute(&ctx, authorized2b).await
         }
         other => other,
     };