security(auth): bind external identity authority

Author: farhan-syah

.config/nextest.toml

@@ -59,7 +59,7 @@ slow-timeout = { period = "30s", terminate-after = 6 }
 # 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)'
+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(spatial_document_geometry_survives_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'
 

nodedb-cluster-tests/tests/array_raft_partition_recovery.rs

@@ -21,7 +21,6 @@ use nodedb_array::sync::op::{ArrayOp, ArrayOpHeader, ArrayOpKind};
 use nodedb_array::sync::op_codec;
 use nodedb_array::types::cell_value::value::CellValue;
 use nodedb_array::types::coord::value::CoordValue;
-use nodedb_types::TenantId;
 use nodedb_types::sync::wire::array::ArrayDeltaMsg;
 
 use common::array_sync::{hlc, register_schema_on_all};
@@ -70,19 +69,7 @@ fn make_inbound(shared: &Arc<SharedState>) -> OriginArrayInbound {
         engine,
         Arc::clone(&shared.array_sync_schemas),
         Arc::clone(shared),
-        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_test_support::pgwire_auth_helpers::superuser(),
     )
 }
 

nodedb-cluster-tests/tests/array_raft_replication.rs

@@ -25,7 +25,6 @@ use nodedb_array::sync::op::{ArrayOp, ArrayOpHeader, ArrayOpKind};
 use nodedb_array::sync::op_codec;
 use nodedb_array::types::cell_value::value::CellValue;
 use nodedb_array::types::coord::value::CoordValue;
-use nodedb_types::TenantId;
 use nodedb_types::sync::wire::array::{ArrayDeltaMsg, ArraySchemaSyncMsg};
 
 use common::array_sync::{
@@ -79,19 +78,7 @@ fn make_inbound(shared: &Arc<SharedState>) -> OriginArrayInbound {
         engine,
         Arc::clone(&shared.array_sync_schemas),
         Arc::clone(shared),
-        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_test_support::pgwire_auth_helpers::superuser(),
     )
 }
 

nodedb-cluster-tests/tests/array_raft_snapshot_install.rs

@@ -21,7 +21,6 @@ use nodedb_array::sync::op::{ArrayOp, ArrayOpHeader, ArrayOpKind};
 use nodedb_array::sync::op_codec;
 use nodedb_array::types::cell_value::value::CellValue;
 use nodedb_array::types::coord::value::CoordValue;
-use nodedb_types::TenantId;
 use nodedb_types::sync::wire::array::{ArrayDeltaMsg, ArraySchemaSyncMsg};
 
 use common::array_sync::{
@@ -72,19 +71,7 @@ fn make_inbound(shared: &Arc<SharedState>) -> OriginArrayInbound {
         engine,
         Arc::clone(&shared.array_sync_schemas),
         Arc::clone(shared),
-        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_test_support::pgwire_auth_helpers::superuser(),
     )
 }
 

nodedb-cluster-tests/tests/array_vshard_migration.rs

@@ -22,7 +22,6 @@ use nodedb_array::sync::op::{ArrayOp, ArrayOpHeader, ArrayOpKind};
 use nodedb_array::sync::op_codec;
 use nodedb_array::types::cell_value::value::CellValue;
 use nodedb_array::types::coord::value::CoordValue;
-use nodedb_types::TenantId;
 use nodedb_types::sync::wire::array::ArrayDeltaMsg;
 
 use common::array_sync::{hlc, register_schema_on_all};
@@ -71,19 +70,7 @@ fn make_inbound(shared: &Arc<SharedState>) -> OriginArrayInbound {
         engine,
         Arc::clone(&shared.array_sync_schemas),
         Arc::clone(shared),
-        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_test_support::pgwire_auth_helpers::superuser(),
     )
 }
 

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

@@ -19,20 +19,10 @@ pub fn authorize_gateway_plan(
     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 identity = nodedb_test_support::pgwire_auth_helpers::superuser();
     let task = PhysicalTask {
         tenant_id: ctx.tenant_id,
         vshard_id: nodedb::types::VShardId::new(0),

nodedb-cluster-tests/tests/sync_failover.rs

@@ -27,7 +27,6 @@ use common::cluster_harness::{TestCluster, wait::wait_for};
 use std::sync::Arc;
 use std::time::Duration;
 
-use nodedb::control::security::identity::{AuthMethod, AuthenticatedIdentity};
 use nodedb::control::server::sync::columnar_handler::{
     ColumnarDispatcher, SharedStateColumnarDispatcher,
 };
@@ -185,16 +184,7 @@ async fn dispatch_columnar_seq(
     seq: u64,
     val: &str,
 ) -> Result<AckStatus, String> {
-    let identity = AuthenticatedIdentity {
-        user_id: 0,
-        username: "sync-failover-test".into(),
-        tenant_id: ctx.tenant,
-        auth_method: AuthMethod::Trust,
-        roles: Vec::new(),
-        is_superuser: true,
-        default_database: Some(DatabaseId::DEFAULT),
-        accessible_databases: AuthenticatedIdentity::default_database_set(true),
-    };
+    let identity = nodedb_test_support::pgwire_auth_helpers::superuser();
     let dispatcher =
         SharedStateColumnarDispatcher::new(ctx.shared.as_ref(), &identity, DatabaseId::DEFAULT);
     let rows = vec![vec![

nodedb-test-support/src/pgwire_auth_helpers.rs

@@ -49,32 +49,27 @@ pub fn make_state_with_catalog() -> Arc<SharedState> {
 
 /// Superuser identity for DDL tests.
 pub fn superuser() -> AuthenticatedIdentity {
-    AuthenticatedIdentity {
-        user_id: 0,
-        username: "nodedb".into(),
-        tenant_id: TenantId::new(0),
-        auth_method: AuthMethod::Trust,
-        roles: vec![Role::Superuser],
-        is_superuser: true,
-        default_database: None,
-        accessible_databases: DatabaseSet::All,
-    }
+    let store = nodedb::control::security::credential::store::CredentialStore::new()
+        .expect("create test credential store");
+    store
+        .bootstrap_superuser("nodedb", "test-only-password")
+        .expect("bootstrap test superuser");
+    store
+        .to_identity("nodedb", AuthMethod::Trust)
+        .expect("build catalog-authenticated test superuser")
 }
 
 /// Readonly identity for permission tests.
 pub fn readonly_user() -> AuthenticatedIdentity {
-    AuthenticatedIdentity {
-        user_id: 99,
-        username: "viewer".into(),
-        tenant_id: TenantId::new(1),
-        auth_method: AuthMethod::Trust,
-        roles: vec![Role::ReadOnly],
-        is_superuser: false,
-        default_database: None,
-        accessible_databases: DatabaseSet::Some(smallvec::smallvec![
-            nodedb_types::id::DatabaseId::DEFAULT
-        ]),
-    }
+    AuthenticatedIdentity::new_regular(
+        99,
+        "viewer",
+        TenantId::new(1),
+        AuthMethod::Trust,
+        vec![Role::ReadOnly],
+        None,
+        DatabaseSet::Some(smallvec::smallvec![nodedb_types::id::DatabaseId::DEFAULT,]),
+    )
 }
 
 /// Run DDL, expect success.
@@ -144,32 +139,28 @@ pub async fn assert_readonly_denied(state: &SharedState, sql: &str) {
 
 /// Cluster-admin identity (no implicit RLS bypass, no cross-DB data access).
 pub fn cluster_admin_user() -> AuthenticatedIdentity {
-    AuthenticatedIdentity {
-        user_id: 100,
-        username: "cluster_admin".into(),
-        tenant_id: nodedb::types::TenantId::new(1),
-        auth_method: AuthMethod::Trust,
-        roles: vec![Role::ClusterAdmin],
-        is_superuser: false,
-        default_database: None,
-        accessible_databases: DatabaseSet::Some(smallvec::smallvec![
-            nodedb_types::id::DatabaseId::DEFAULT
-        ]),
-    }
+    AuthenticatedIdentity::new_regular(
+        100,
+        "cluster_admin",
+        nodedb::types::TenantId::new(1),
+        AuthMethod::Trust,
+        vec![Role::ClusterAdmin],
+        None,
+        DatabaseSet::Some(smallvec::smallvec![nodedb_types::id::DatabaseId::DEFAULT,]),
+    )
 }
 
 /// Database-owner identity for `db_id`.
 pub fn database_owner_user(db_id: nodedb_types::id::DatabaseId) -> AuthenticatedIdentity {
-    AuthenticatedIdentity {
-        user_id: 101,
-        username: "db_owner".into(),
-        tenant_id: nodedb::types::TenantId::new(1),
-        auth_method: AuthMethod::Trust,
-        roles: vec![Role::DatabaseOwner(db_id)],
-        is_superuser: false,
-        default_database: None,
-        accessible_databases: DatabaseSet::Some(smallvec::smallvec![db_id]),
-    }
+    AuthenticatedIdentity::new_regular(
+        101,
+        "db_owner",
+        nodedb::types::TenantId::new(1),
+        AuthMethod::Trust,
+        vec![Role::DatabaseOwner(db_id)],
+        None,
+        DatabaseSet::Some(smallvec::smallvec![db_id]),
+    )
 }
 
 /// Assert that the audit log contains at least one entry with `event` and `db_id`.

nodedb/src/config/auth/config.rs

@@ -119,10 +119,14 @@ pub struct JwtAuthConfig {
     #[serde(default = "default_allowed_algorithms")]
     pub allowed_algorithms: Vec<String>,
 
-    /// Clock skew tolerance in seconds for `exp`/`nbf` validation.
+    /// Clock skew tolerance in seconds for `exp`/`nbf`/`iat` validation.
     #[serde(default = "default_clock_skew")]
     pub clock_skew_secs: u64,
 
+    /// Maximum accepted JWT lifetime (`exp - iat`) in seconds.
+    #[serde(default = "default_max_token_lifetime")]
+    pub max_token_lifetime_secs: u64,
+
     /// Path to cache JWKS on disk for offline fallback.
     /// If set, JWKS responses are persisted and used when providers are unreachable.
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -252,6 +256,11 @@ impl JwtAuthConfig {
     /// Validate all providers. Called by the server-config loader and JWKS
     /// registry construction so misconfiguration fails before authentication.
     pub fn validate(&self) -> crate::Result<()> {
+        if self.max_token_lifetime_secs == 0 {
+            return Err(crate::Error::Config {
+                detail: "auth.jwt max_token_lifetime_secs must be greater than zero".into(),
+            });
+        }
         let policy = self.jwks_policy().map_err(|e| crate::Error::Config {
             detail: format!("auth.jwt allow-list is invalid: {e}"),
         })?;
@@ -307,6 +316,9 @@ fn default_jwks_min_refetch() -> u64 {
 fn default_clock_skew() -> u64 {
     60
 }
+fn default_max_token_lifetime() -> u64 {
+    86_400
+}
 fn default_allowed_algorithms() -> Vec<String> {
     vec!["RS256".into(), "ES256".into()]
 }
@@ -321,6 +333,7 @@ impl Default for JwtAuthConfig {
             jwks_min_refetch_secs: default_jwks_min_refetch(),
             allowed_algorithms: default_allowed_algorithms(),
             clock_skew_secs: default_clock_skew(),
+            max_token_lifetime_secs: default_max_token_lifetime(),
             jwks_cache_path: None,
             providers: Vec::new(),
             jit_provisioning: false,

nodedb/src/control/crdt_admission.rs

@@ -953,19 +953,18 @@ mod tests {
             post_set_op: nodedb_physical::physical_task::PostSetOp::None,
             txn_id: None,
         };
-        let identity = crate::control::security::identity::AuthenticatedIdentity {
-            user_id: 1,
-            username: "crdt-test".into(),
-            tenant_id,
-            auth_method: crate::control::security::identity::AuthMethod::Trust,
-            roles: Vec::new(),
-            is_superuser: true,
-            default_database: None,
-            accessible_databases:
+        let identity =
+            crate::control::security::identity::AuthenticatedIdentity::new_internal_service(
+                1,
+                "crdt-test",
+                tenant_id,
+                Vec::new(),
+                true,
+                None,
                 crate::control::security::identity::AuthenticatedIdentity::default_database_set(
                     true,
                 ),
-        };
+            );
         let authorized = crate::control::server::shared::authorization::authorize_task_set(
             &identity,
             std::slice::from_ref(&task),