feat(security): centralize SQL task authorization

Author: farhan-syah

nodedb/src/control/server/shared/authorization/error.rs

@@ -0,0 +1,36 @@
+//! Transport-neutral authorization failures.
+
+use crate::types::TenantId;
+
+/// A safe, transport-neutral authorization denial.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AuthorizationError {
+    tenant_id: TenantId,
+    resource: String,
+}
+
+impl AuthorizationError {
+    pub fn new(tenant_id: TenantId, resource: impl Into<String>) -> Self {
+        Self {
+            tenant_id,
+            resource: resource.into(),
+        }
+    }
+
+    pub fn tenant_id(&self) -> TenantId {
+        self.tenant_id
+    }
+
+    pub fn resource(&self) -> &str {
+        &self.resource
+    }
+}
+
+impl From<AuthorizationError> for crate::Error {
+    fn from(error: AuthorizationError) -> Self {
+        Self::RejectedAuthz {
+            tenant_id: error.tenant_id,
+            resource: error.resource,
+        }
+    }
+}

nodedb/src/control/server/shared/authorization/mod.rs

@@ -0,0 +1,9 @@
+//! Protocol-neutral authorization for SQL physical tasks.
+
+pub mod error;
+pub mod requirements;
+pub mod service;
+
+pub use error::AuthorizationError;
+pub use requirements::{AuthorizationRequirement, plan_requirements};
+pub use service::{authorize_database, authorize_task_set};

nodedb/src/control/server/shared/authorization/requirements.rs

@@ -0,0 +1,47 @@
+//! Physical-plan authorization requirements.
+//!
+//! This deliberately does not use `shared::plan_util::extract_collection`: a
+//! single collection cannot represent joins or source/target DML correctly.
+
+#![deny(clippy::wildcard_enum_match_arm)]
+
+use crate::control::security::identity::Permission;
+
+mod collect;
+mod order;
+mod query;
+
+pub use collect::plan_requirements;
+
+/// A protected resource and the permission needed to use it.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum AuthorizationRequirement {
+    /// A collection-scoped operation. The name is the physical-plan name and
+    /// may be database-qualified; the authorization service normalizes it to
+    /// the grant-store name before looking up grants.
+    Collection {
+        collection: String,
+        permission: Permission,
+    },
+    /// An operation with no collection-level resource, such as an array or a
+    /// tenant-wide maintenance action. It must still be authorized at tenant
+    /// scope rather than silently allowed.
+    Tenant { permission: Permission },
+}
+
+impl AuthorizationRequirement {
+    fn collection(collection: impl Into<String>, permission: Permission) -> Self {
+        Self::Collection {
+            collection: collection.into(),
+            permission,
+        }
+    }
+
+    fn tenant(permission: Permission) -> Self {
+        Self::Tenant { permission }
+    }
+}
+
+#[cfg(test)]
+#[path = "requirements/tests.rs"]
+mod tests;

nodedb/src/control/server/shared/authorization/requirements/collect.rs

@@ -0,0 +1,221 @@
+use crate::bridge::envelope::PhysicalPlan;
+use crate::control::gateway::version_set::touched_collections;
+use crate::control::security::identity::{Permission, required_permission};
+
+use super::AuthorizationRequirement;
+use super::order::requirement_order;
+use super::query::collect_query_requirements;
+
+/// Return every collection requirement for `plan`.
+///
+/// The general plan permission applies to ordinary single-resource plans. The
+/// multi-resource cases below intentionally override it so sources require
+/// `Read` while targets require `Write`. Nested query inputs are traversed
+/// iteratively in addition to their named collection fields.
+pub fn plan_requirements(plan: &PhysicalPlan) -> Vec<AuthorizationRequirement> {
+    let mut requirements = Vec::new();
+    collect_requirements(plan, &mut requirements);
+    requirements.sort_by(requirement_order);
+    requirements.dedup();
+    requirements
+}
+
+fn collect_requirements(plan: &PhysicalPlan, out: &mut Vec<AuthorizationRequirement>) {
+    use nodedb_physical::physical_plan::{CrdtOp, DocumentOp, GraphOp, KvOp, MetaOp, SpatialOp};
+
+    let mut pending = vec![plan];
+    while let Some(plan) = pending.pop() {
+        let initial_len = out.len();
+        match plan {
+            PhysicalPlan::Document(DocumentOp::InsertSelect {
+                target_collection,
+                source_collection,
+                ..
+            })
+            | PhysicalPlan::Document(DocumentOp::UpdateFromJoin {
+                target_collection,
+                source_collection,
+                ..
+            })
+            | PhysicalPlan::Document(DocumentOp::Merge {
+                target_collection,
+                source_collection,
+                ..
+            }) => {
+                out.push(AuthorizationRequirement::collection(
+                    target_collection,
+                    Permission::Write,
+                ));
+                out.push(AuthorizationRequirement::collection(
+                    source_collection,
+                    Permission::Read,
+                ));
+            }
+            PhysicalPlan::Kv(KvOp::TransferItem {
+                source_collection,
+                dest_collection,
+                ..
+            }) => {
+                out.push(AuthorizationRequirement::collection(
+                    source_collection,
+                    Permission::Read,
+                ));
+                out.push(AuthorizationRequirement::collection(
+                    dest_collection,
+                    Permission::Write,
+                ));
+            }
+            PhysicalPlan::Query(_) | PhysicalPlan::Vector(_) => {
+                if !collect_query_requirements(plan, &mut pending, out) {
+                    add_general_requirements(plan, out);
+                }
+            }
+            PhysicalPlan::Spatial(SpatialOp::Insert { collection, .. })
+            | PhysicalPlan::Spatial(SpatialOp::Delete { collection, .. })
+            | PhysicalPlan::Crdt(
+                CrdtOp::ImportSnapshot { collection, .. }
+                | CrdtOp::SetConstraints { collection, .. }
+                | CrdtOp::DropConstraints { collection, .. }
+                | CrdtOp::ReadConstraints { collection, .. }
+                | CrdtOp::GetVersionVector { collection, .. }
+                | CrdtOp::ExportDelta { collection, .. }
+                | CrdtOp::CompactAtVersion { collection, .. },
+            ) => add_collection_requirement(collection, required_permission(plan), out),
+            PhysicalPlan::Graph(GraphOp::EdgePut { collection, .. })
+            | PhysicalPlan::Graph(GraphOp::EdgeDelete { collection, .. }) => {
+                add_collection_requirement(collection, required_permission(plan), out);
+            }
+            PhysicalPlan::Graph(GraphOp::EdgePutBatch { edges })
+            | PhysicalPlan::Graph(GraphOp::EdgeDeleteBatch { edges }) => {
+                let permission = required_permission(plan);
+                for edge in edges {
+                    add_collection_requirement(&edge.collection, permission, out);
+                }
+            }
+            PhysicalPlan::Meta(
+                MetaOp::ConvertCollection { collection, .. }
+                | MetaOp::EnforceTimeseriesRetention { collection, .. }
+                | MetaOp::TemporalPurgeEdgeStore { collection, .. }
+                | MetaOp::TemporalPurgeDocumentStrict { collection, .. }
+                | MetaOp::TemporalPurgeColumnar { collection, .. }
+                | MetaOp::TemporalPurgeCrdt { collection, .. }
+                | MetaOp::QueryLastValues { collection }
+                | MetaOp::QueryLastValue { collection, .. }
+                | MetaOp::RebuildIndex { collection, .. },
+            ) => add_collection_requirement(collection, required_permission(plan), out),
+            PhysicalPlan::Meta(
+                MetaOp::UnregisterCollection { name, .. }
+                | MetaOp::UnregisterMaterializedView { name, .. }
+                | MetaOp::QueryCollectionSize { name, .. },
+            ) => add_collection_requirement(name, required_permission(plan), out),
+            PhysicalPlan::Meta(MetaOp::RenameCollection {
+                old_collection,
+                new_collection,
+                ..
+            }) => {
+                let permission = required_permission(plan);
+                add_collection_requirement(old_collection, permission, out);
+                add_collection_requirement(new_collection, permission, out);
+            }
+            PhysicalPlan::Meta(MetaOp::TransactionBatch { plans, .. })
+            | PhysicalPlan::Meta(MetaOp::ResolveTxn { plans, .. })
+            | PhysicalPlan::Meta(MetaOp::RecordCalvinWriteVersions { plans, .. })
+            | PhysicalPlan::Meta(MetaOp::CalvinExecuteStatic { plans, .. })
+            | PhysicalPlan::Meta(MetaOp::CalvinExecuteActive { plans, .. }) => {
+                add_tenant_requirement(required_permission(plan), out);
+                for nested in plans {
+                    pending.push(nested);
+                }
+            }
+            PhysicalPlan::Meta(MetaOp::StageWrite { plan: nested }) => {
+                add_tenant_requirement(required_permission(plan), out);
+                pending.push(nested);
+            }
+            PhysicalPlan::Kv(KvOp::Transfer { collection, .. }) => {
+                out.push(AuthorizationRequirement::collection(
+                    collection,
+                    Permission::Write,
+                ));
+            }
+            PhysicalPlan::Kv(
+                KvOp::Get { .. }
+                | KvOp::Put { .. }
+                | KvOp::Insert { .. }
+                | KvOp::InsertIfAbsent { .. }
+                | KvOp::InsertOnConflictUpdate { .. }
+                | KvOp::Delete { .. }
+                | KvOp::Scan { .. }
+                | KvOp::Expire { .. }
+                | KvOp::Persist { .. }
+                | KvOp::GetTtl { .. }
+                | KvOp::BatchGet { .. }
+                | KvOp::BatchPut { .. }
+                | KvOp::RegisterIndex { .. }
+                | KvOp::DropIndex { .. }
+                | KvOp::FieldGet { .. }
+                | KvOp::FieldSet { .. }
+                | KvOp::Truncate { .. }
+                | KvOp::Incr { .. }
+                | KvOp::IncrFloat { .. }
+                | KvOp::Cas { .. }
+                | KvOp::GetSet { .. }
+                | KvOp::RegisterSortedIndex { .. }
+                | KvOp::DropSortedIndex { .. }
+                | KvOp::SortedIndexRank { .. }
+                | KvOp::SortedIndexTopK { .. }
+                | KvOp::SortedIndexRange { .. }
+                | KvOp::SortedIndexCount { .. }
+                | KvOp::SortedIndexScore { .. }
+                | KvOp::MaterializeScan { .. },
+            ) => add_general_requirements(plan, out),
+            PhysicalPlan::Document(_)
+            | PhysicalPlan::Graph(_)
+            | PhysicalPlan::Text(_)
+            | PhysicalPlan::Columnar(_)
+            | PhysicalPlan::Timeseries(_)
+            | PhysicalPlan::Spatial(_)
+            | PhysicalPlan::Crdt(_)
+            | PhysicalPlan::Meta(_)
+            | PhysicalPlan::Array(_)
+            | PhysicalPlan::ClusterArray(_) => add_general_requirements(plan, out),
+        }
+
+        // A plan without a collection name is not public: it still needs the
+        // permission implied by its physical operation at tenant scope. This also
+        // covers provider-materialized and array-backed plans.
+        if out.len() == initial_len {
+            add_tenant_requirement(required_permission(plan), out);
+        }
+    }
+}
+
+pub(super) fn add_general_requirements(
+    plan: &PhysicalPlan,
+    out: &mut Vec<AuthorizationRequirement>,
+) {
+    let permission = required_permission(plan);
+    for collection in touched_collections(plan) {
+        out.push(AuthorizationRequirement::collection(collection, permission));
+    }
+}
+
+pub(super) fn add_collection_requirement(
+    collection: &str,
+    permission: Permission,
+    out: &mut Vec<AuthorizationRequirement>,
+) {
+    if !collection.is_empty() {
+        out.push(AuthorizationRequirement::collection(collection, permission));
+    }
+}
+
+pub(super) fn add_tenant_requirement(
+    permission: Permission,
+    out: &mut Vec<AuthorizationRequirement>,
+) {
+    out.push(AuthorizationRequirement::tenant(permission));
+}
+
+pub(super) fn add_read(collection: &str, out: &mut Vec<AuthorizationRequirement>) {
+    add_collection_requirement(collection, Permission::Read, out);
+}

nodedb/src/control/server/shared/authorization/requirements/order.rs

@@ -0,0 +1,47 @@
+use crate::control::security::identity::Permission;
+
+use super::AuthorizationRequirement;
+
+pub(super) fn requirement_order(
+    left: &AuthorizationRequirement,
+    right: &AuthorizationRequirement,
+) -> std::cmp::Ordering {
+    match (left, right) {
+        (
+            AuthorizationRequirement::Tenant { permission: left },
+            AuthorizationRequirement::Tenant { permission: right },
+        ) => permission_rank(*left).cmp(&permission_rank(*right)),
+        (AuthorizationRequirement::Tenant { .. }, AuthorizationRequirement::Collection { .. }) => {
+            std::cmp::Ordering::Less
+        }
+        (AuthorizationRequirement::Collection { .. }, AuthorizationRequirement::Tenant { .. }) => {
+            std::cmp::Ordering::Greater
+        }
+        (
+            AuthorizationRequirement::Collection {
+                collection: left_collection,
+                permission: left_permission,
+            },
+            AuthorizationRequirement::Collection {
+                collection: right_collection,
+                permission: right_permission,
+            },
+        ) => left_collection.cmp(right_collection).then_with(|| {
+            permission_rank(*left_permission).cmp(&permission_rank(*right_permission))
+        }),
+    }
+}
+
+fn permission_rank(permission: Permission) -> u8 {
+    match permission {
+        Permission::Read => 0,
+        Permission::Write => 1,
+        Permission::Create => 2,
+        Permission::Drop => 3,
+        Permission::Alter => 4,
+        Permission::Admin => 5,
+        Permission::Monitor => 6,
+        Permission::Execute => 7,
+        Permission::Backup => 8,
+    }
+}