security: complete authorization denial matrix

Author: farhan-syah

nodedb/src/control/security/predicate_parser.rs

@@ -372,6 +372,7 @@ pub fn validate_auth_refs(predicate: &RlsPredicate) -> crate::Result<()> {
         "auth_method",
         "auth_time",
         "session_id",
+        "database_id",
     ];
 
     fn check(pred: &RlsPredicate, known: &[&str]) -> crate::Result<()> {

nodedb/src/control/server/http/routes/crdt.rs

@@ -11,11 +11,15 @@ use axum::http::HeaderMap;
 use axum::response::IntoResponse;
 
 use crate::bridge::envelope::PhysicalPlan;
-use crate::control::security::audit::ArcAuditEmitter;
-use crate::control::security::identity::Permission;
+use crate::control::security::audit::{ArcAuditEmitter, AuditEmitter};
+use crate::control::security::identity::{AuthenticatedIdentity, Permission};
+use crate::control::security::permission::PermissionStore;
+use crate::control::security::role::RoleStore;
 use crate::control::server::http::auth::{ApiError, AppState, resolve_identity};
 use crate::control::server::http::types::{HttpCrdtApplyRequest, HttpCrdtApplyResponse};
-use crate::control::server::shared::authorization::{authorize_collection, authorize_task_set};
+use crate::control::server::shared::authorization::{
+    AuthorizationError, authorize_collection, authorize_task_set,
+};
 use crate::control::server::shared::ddl::sql_parse::hex_decode;
 use nodedb_physical::physical_plan::CrdtOp;
 use nodedb_physical::physical_task::{PhysicalTask, PostSetOp};
@@ -45,11 +49,9 @@ pub async fn crdt_apply(
     let identity = resolve_identity(&headers, &state, "http")?;
 
     let audit = ArcAuditEmitter(std::sync::Arc::clone(&state.shared.audit));
-    authorize_collection(
+    preflight_crdt_apply(
         &identity,
-        crate::types::DatabaseId::DEFAULT,
         &collection,
-        Permission::Write,
         &state.shared.permissions,
         &state.shared.roles,
         &audit,
@@ -147,6 +149,25 @@ pub async fn crdt_apply(
     )))
 }
 
+/// Authorize the fixed DEFAULT-database write before any CRDT apply work.
+fn preflight_crdt_apply(
+    identity: &AuthenticatedIdentity,
+    collection: &str,
+    permissions: &PermissionStore,
+    roles: &RoleStore,
+    audit: &dyn AuditEmitter,
+) -> Result<(), AuthorizationError> {
+    authorize_collection(
+        identity,
+        crate::types::DatabaseId::DEFAULT,
+        collection,
+        Permission::Write,
+        permissions,
+        roles,
+        audit,
+    )
+}
+
 fn decode_bounded_delta(encoded: &str) -> Result<Vec<u8>, ApiError> {
     let delta = hex_decode(encoded)
         .ok_or_else(|| ApiError::BadRequest("invalid hex in 'delta' field".into()))?;
@@ -162,6 +183,95 @@ fn decode_bounded_delta(encoded: &str) -> Result<Vec<u8>, ApiError> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::control::security::audit::AuditEvent;
+    use crate::control::security::audit::emitter::test_helpers::CapturingEmitter;
+    use crate::control::security::identity::{AuthMethod, DatabaseSet};
+    use crate::types::TenantId;
+
+    fn regular_identity() -> AuthenticatedIdentity {
+        AuthenticatedIdentity::new_regular(
+            7,
+            "writer",
+            TenantId::new(9),
+            AuthMethod::ApiKey,
+            Vec::new(),
+            None,
+            DatabaseSet::Some(smallvec::smallvec![crate::types::DatabaseId::DEFAULT]),
+        )
+    }
+
+    fn assert_one_audit_denial(audit: &CapturingEmitter) {
+        let events = audit.recorded();
+        assert_eq!(events.len(), 1);
+        assert_eq!(events[0].0, AuditEvent::PermissionDenied);
+    }
+
+    #[test]
+    fn ungranted_regular_identity_is_denied_before_authorized_work_runs() {
+        // This endpoint has no read action or caller-selectable database, so
+        // those authorization-matrix cells are N/A.
+        let identity = regular_identity();
+        let permissions = PermissionStore::new();
+        let roles = RoleStore::new();
+        let audit = CapturingEmitter::new();
+        let mut authorized_work_runs = 0;
+        let preflight = preflight_crdt_apply(&identity, "orders", &permissions, &roles, &audit);
+
+        if preflight.is_ok() {
+            authorized_work_runs += 1;
+        }
+
+        assert!(preflight.is_err());
+        assert_eq!(authorized_work_runs, 0);
+        assert_one_audit_denial(&audit);
+    }
+
+    #[test]
+    fn regular_identity_cannot_write_system_audit_log_even_with_grant() {
+        // This endpoint has no read action or caller-selectable database, so
+        // those authorization-matrix cells are N/A.
+        let identity = regular_identity();
+        let permissions = PermissionStore::new();
+        permissions
+            .grant(
+                "collection:9:_system.audit_log",
+                "user:writer",
+                Permission::Write,
+                "admin",
+                None,
+            )
+            .expect("in-memory grant succeeds");
+        let roles = RoleStore::new();
+        let audit = CapturingEmitter::new();
+
+        assert!(
+            preflight_crdt_apply(&identity, "_system.audit_log", &permissions, &roles, &audit,)
+                .is_err()
+        );
+        assert_one_audit_denial(&audit);
+    }
+
+    #[test]
+    fn tenant_ten_grant_does_not_authorize_tenant_nine_identity() {
+        // This endpoint has no read action or caller-selectable database, so
+        // those authorization-matrix cells are N/A.
+        let identity = regular_identity();
+        let permissions = PermissionStore::new();
+        permissions
+            .grant(
+                "collection:10:orders",
+                "user:writer",
+                Permission::Write,
+                "admin",
+                None,
+            )
+            .expect("in-memory grant succeeds");
+        let roles = RoleStore::new();
+        let audit = CapturingEmitter::new();
+
+        assert!(preflight_crdt_apply(&identity, "orders", &permissions, &roles, &audit).is_err());
+        assert_one_audit_denial(&audit);
+    }
 
     #[test]
     fn decoded_delta_limit_is_exact_and_invalid_hex_is_bad_request() {

nodedb/src/control/server/ilp_batch.rs

@@ -306,18 +306,76 @@ fn build_ilp_calvin_tasks(
 
 #[cfg(test)]
 mod tests {
-    use super::{IlpPreflightFailure, preflight_ilp_batch};
-    use crate::control::security::audit::NoopAuditEmitter;
+    use super::{IlpPreflightFailure, flush_authenticated_ilp_batch, preflight_ilp_batch};
+    use std::sync::{Arc, Mutex};
+
+    use crate::bridge::dispatch::Dispatcher;
+    use crate::control::state::SharedState;
+
     use crate::control::security::audit::emitter::test_helpers::CapturingEmitter;
+    use crate::control::security::audit::{
+        AuditEmitContext, AuditEmitter, AuditEvent, NoopAuditEmitter,
+    };
     use crate::control::security::identity::{
         AuthMethod, AuthenticatedIdentity, DatabaseSet, Permission,
     };
     use crate::control::security::permission::PermissionStore;
     use crate::control::security::role::RoleStore;
     use crate::types::{DatabaseId, TenantId, VShardId};
+    use crate::wal::WalManager;
     use nodedb_physical::physical_plan::{PhysicalPlan, TimeseriesOp};
     use nodedb_types::Surrogate;
 
+    #[derive(Clone, Debug)]
+    struct CapturedAuditEvent {
+        event: AuditEvent,
+        source: String,
+        auth_user_id: String,
+        auth_user_name: String,
+        tenant_id: Option<TenantId>,
+    }
+
+    struct ContextCapturingEmitter {
+        events: Mutex<Vec<CapturedAuditEvent>>,
+    }
+
+    impl ContextCapturingEmitter {
+        fn new() -> Self {
+            Self {
+                events: Mutex::new(Vec::new()),
+            }
+        }
+
+        fn recorded(&self) -> Vec<CapturedAuditEvent> {
+            self.events
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .clone()
+        }
+    }
+
+    impl AuditEmitter for ContextCapturingEmitter {
+        fn emit(
+            &self,
+            event: AuditEvent,
+            source: &str,
+            _detail: &str,
+            context: AuditEmitContext<'_>,
+        ) {
+            let mut events = self
+                .events
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            events.push(CapturedAuditEvent {
+                event,
+                source: source.into(),
+                auth_user_id: context.auth_user_id.into(),
+                auth_user_name: context.auth_user_name.into(),
+                tenant_id: context.tenant_id,
+            });
+        }
+    }
+
     fn identity(database_id: DatabaseId) -> AuthenticatedIdentity {
         AuthenticatedIdentity::new_regular(
             7,
@@ -351,6 +409,42 @@ mod tests {
         )
     }
 
+    #[tokio::test]
+    async fn authenticated_flush_rejects_unwritable_measurement_before_dispatch_or_catalog_projection()
+     {
+        let directory = tempfile::tempdir().expect("create ILP batch test directory");
+        let wal = Arc::new(
+            WalManager::open_for_testing(&directory.path().join("ilp-batch.wal"))
+                .expect("open ILP batch test WAL"),
+        );
+        let (dispatcher, _data_sides) = Dispatcher::new(1, 64);
+        let state = SharedState::new(dispatcher, wal).expect("construct ILP batch state");
+        let database_id = DatabaseId::new(7);
+
+        let error = flush_authenticated_ilp_batch(
+            &state,
+            &identity(database_id),
+            database_id,
+            "cpu value=1i\n",
+        )
+        .await
+        .expect_err("regular identity without write permission is rejected during preflight");
+
+        assert!(matches!(
+            error,
+            crate::Error::BadRequest { detail } if detail == "ILP batch rejected"
+        ));
+        assert!(
+            state
+                .credentials
+                .catalog()
+                .get_collection(database_id, 9, "cpu")
+                .expect("read catalog collection projection")
+                .is_none(),
+            "early authorization denial must not create a collection/schema projection"
+        );
+    }
+
     #[test]
     fn groups_two_measurements_in_canonical_order_and_preserves_source_order() {
         let permissions = PermissionStore::new();
@@ -445,6 +539,7 @@ mod tests {
 
     #[test]
     fn read_only_collection_is_not_sufficient_for_ilp_ingest() {
+        // ILP is write-only; read access is not applicable to ingestion.
         let permissions = PermissionStore::new();
         permissions
             .grant(
@@ -462,13 +557,59 @@ mod tests {
         );
     }
 
+    #[test]
+    fn system_audit_log_measurement_is_denied_to_regular_ingester() {
+        let permissions = PermissionStore::new();
+        grant_write(&permissions, "_system.audit_log");
+        let audit = ContextCapturingEmitter::new();
+
+        assert_eq!(
+            preflight_ilp_batch(
+                &identity(DatabaseId::new(7)),
+                DatabaseId::new(7),
+                "_system.audit_log value=1i\n",
+                &permissions,
+                &RoleStore::new(),
+                &audit,
+            ),
+            Err(IlpPreflightFailure::PermissionDenied)
+        );
+        let events = audit.recorded();
+        assert_eq!(events.len(), 1);
+        assert_eq!(events[0].event, AuditEvent::PermissionDenied);
+        assert_eq!(events[0].source, "ingester");
+        assert_eq!(events[0].auth_user_id, "7");
+        assert_eq!(events[0].auth_user_name, "ingester");
+        assert_eq!(events[0].tenant_id, Some(TenantId::new(9)));
+    }
+
+    #[test]
+    fn tenant_ten_write_grant_does_not_authorize_tenant_nine_ilp_ingest() {
+        let permissions = PermissionStore::new();
+        permissions
+            .grant(
+                "collection:10:cpu",
+                "user:ingester",
+                Permission::Write,
+                "admin",
+                None,
+            )
+            .expect("in-memory grant succeeds");
+
+        assert_eq!(
+            preflight("cpu value=1i\n", &permissions),
+            Err(IlpPreflightFailure::PermissionDenied)
+        );
+    }
+
     #[test]
     fn non_default_database_is_used_for_collection_authorization() {
         let permissions = PermissionStore::new();
         grant_write(&permissions, "cpu");
         let database_id = DatabaseId::new(7);
         let roles = RoleStore::new();
 
+        // Database selection is handshake-bound; ILP payload does not select it.
         assert_eq!(
             preflight_ilp_batch(
                 &identity(DatabaseId::DEFAULT),

nodedb/src/control/server/native/dispatch/sql.rs

@@ -125,8 +125,15 @@ async fn handle_sql_inner(
         return resp(NativeResponse::status_row(seq, "RESET"));
     }
     if upper == "DISCARD ALL" {
+        // Recreate only disposable session state. The authenticated database
+        // binding belongs to the connection and must survive the reset.
+        let database_id = ctx.sessions.get_current_database(ctx.peer_addr);
         ctx.sessions.remove(ctx.peer_addr);
         ctx.sessions.ensure_session(*ctx.peer_addr);
+        if let Some(database_id) = database_id {
+            ctx.sessions
+                .set_current_database(ctx.peer_addr, database_id);
+        }
         return resp(NativeResponse::status_row(seq, "DISCARD ALL"));
     }