diff --git a/lib/storage/src/content_manager/claims.rs b/lib/storage/src/content_manager/claims.rs index 022d35c44d..2b989735a1 100644 --- a/lib/storage/src/content_manager/claims.rs +++ b/lib/storage/src/content_manager/claims.rs @@ -14,7 +14,7 @@ use itertools::{Either, Itertools as _}; use segment::types::{Condition, ExtendedPointId, FieldCondition, Filter, Match, Payload}; use super::errors::StorageError; -use crate::rbac::access::{Access, PayloadClaim}; +use crate::rbac::access::PayloadClaim; pub fn check_collection_name( collections: Option<&Vec>, @@ -28,53 +28,6 @@ pub fn check_collection_name( }) } -/// Check if a access object is allowed to manage collections. -pub fn check_manage_rights(access: &Access) -> Result<(), StorageError> { - let Access { - collections, - payload, - } = access; - if collections.is_some() { - return incompatible_with_collection_claim(); - } - if payload.is_some() { - return incompatible_with_payload_claim(); - } - Ok(()) -} - -/// Check if a claim object has full access to a collection. -pub fn check_full_access_to_collection( - access: &Access, - collection_name: &str, -) -> Result<(), StorageError> { - let Access { - collections, - payload, - } = access; - if let Some(collections) = collections { - check_collection_name(Some(collections), collection_name)?; - } - if payload.is_some() { - return incompatible_with_payload_claim(); - } - Ok(()) -} - -pub fn check_points_op( - collections: Option<&Vec>, - payload: Option<&PayloadClaim>, - op: &mut impl PointsOpClaimsChecker, -) -> Result<(), StorageError> { - for collection in op.collections_used() { - check_collection_name(collections, collection)?; - } - if let Some(payload) = payload { - op.apply_payload_claim(payload)?; - } - Ok(()) -} - pub trait PointsOpClaimsChecker { /// An iterator over the collection names used in the operation, for checking `collections` /// claim. diff --git a/lib/storage/src/content_manager/snapshots/mod.rs b/lib/storage/src/content_manager/snapshots/mod.rs index 73bbd02be3..6ce5468ca1 100644 --- a/lib/storage/src/content_manager/snapshots/mod.rs +++ b/lib/storage/src/content_manager/snapshots/mod.rs @@ -11,11 +11,9 @@ use tempfile::TempPath; use tokio::io::AsyncWriteExt; use tokio::task::JoinHandle; -use super::claims::check_manage_rights; -use crate::content_manager::claims::check_full_access_to_collection; use crate::content_manager::toc::FULL_SNAPSHOT_FILE_NAME; use crate::dispatcher::Dispatcher; -use crate::rbac::access::Access; +use crate::rbac::access::{Access, CollectionMultipass}; use crate::{StorageError, TableOfContent}; #[derive(Debug, Serialize, Deserialize, Clone)] @@ -61,7 +59,7 @@ pub async fn do_delete_full_snapshot( access: Access, snapshot_name: &str, ) -> Result>, StorageError> { - check_manage_rights(&access)?; + access.check_manage_rights()?; let dispatcher = dispatcher.clone(); let snapshot_manager = dispatcher.clone().toc().get_snapshots_storage_manager(); let snapshot_dir = get_full_snapshot_path(dispatcher.toc(), snapshot_name).await?; @@ -77,10 +75,9 @@ pub async fn do_delete_collection_snapshot( collection_name: &str, snapshot_name: &str, ) -> Result>, StorageError> { - check_full_access_to_collection(&access, collection_name)?; - let collection_name = collection_name.to_string(); + let collection_pass = access.check_whole_collection_rights(collection_name)?; let snapshot_name = snapshot_name.to_string(); - let collection = dispatcher.get_collection(&collection_name).await?; + let collection = dispatcher.get_collection_by_pass(&collection_pass).await?; let file_name = collection.get_snapshot_path(&snapshot_name).await?; let snapshot_manager = dispatcher.clone().toc().get_snapshots_storage_manager(); @@ -94,7 +91,7 @@ pub async fn do_list_full_snapshots( toc: &TableOfContent, access: Access, ) -> Result, StorageError> { - check_manage_rights(&access)?; + access.check_manage_rights()?; let snapshots_manager = toc.get_snapshots_storage_manager(); let snapshots_path = Path::new(toc.snapshots_path()); Ok(snapshots_manager.list_snapshots(snapshots_path).await?) @@ -104,15 +101,16 @@ pub fn do_create_full_snapshot( dispatcher: &Dispatcher, access: Access, ) -> Result>, StorageError> { - check_manage_rights(&access)?; + let multipass = access.check_manage_rights()?; let dispatcher = dispatcher.clone(); Ok(tokio::spawn(async move { - _do_create_full_snapshot(&dispatcher).await + _do_create_full_snapshot(&dispatcher, multipass).await })) } async fn _do_create_full_snapshot( dispatcher: &Dispatcher, + multipass: CollectionMultipass, ) -> Result { let dispatcher = dispatcher.clone(); @@ -121,7 +119,9 @@ async fn _do_create_full_snapshot( let all_collections = dispatcher.all_collections().await; let mut created_snapshots: Vec<(&str, SnapshotDescription)> = vec![]; for collection_name in &all_collections { - let snapshot_details = dispatcher.create_snapshot(collection_name).await?; + let snapshot_details = dispatcher + .create_snapshot(&multipass.issue_pass(collection_name)) + .await?; created_snapshots.push((collection_name, snapshot_details)); } let current_time = chrono::Utc::now().format("%Y-%m-%d-%H-%M-%S").to_string(); diff --git a/lib/storage/src/content_manager/snapshots/recover.rs b/lib/storage/src/content_manager/snapshots/recover.rs index fd8e6b66f8..e0aef5c4ed 100644 --- a/lib/storage/src/content_manager/snapshots/recover.rs +++ b/lib/storage/src/content_manager/snapshots/recover.rs @@ -8,13 +8,12 @@ use collection::shards::shard_config::ShardType; use collection::shards::shard_versioning::latest_shard_paths; use tokio::task::JoinHandle; -use crate::content_manager::claims::check_manage_rights; use crate::content_manager::collection_meta_ops::{ CollectionMetaOperations, CreateCollectionOperation, }; use crate::content_manager::snapshots::download::download_snapshot; use crate::dispatcher::Dispatcher; -use crate::rbac::access::Access; +use crate::rbac::access::{Access, CollectionPass}; use crate::{StorageError, TableOfContent}; pub async fn activate_shard( @@ -56,18 +55,18 @@ pub fn do_recover_from_snapshot( access: Access, client: reqwest::Client, ) -> Result>, StorageError> { - check_manage_rights(&access)?; + let multipass = access.check_manage_rights()?; let dispatch = dispatcher.clone(); - let collection_name = collection_name.to_string(); + let collection_pass = multipass.issue_pass(collection_name).into_static(); Ok(tokio::spawn(async move { - _do_recover_from_snapshot(dispatch, &collection_name, source, &client).await + _do_recover_from_snapshot(dispatch, collection_pass, source, &client).await })) } async fn _do_recover_from_snapshot( dispatcher: Dispatcher, - collection_name: &str, + collection_pass: CollectionPass<'static>, source: SnapshotRecover, client: &reqwest::Client, ) -> Result { @@ -106,11 +105,11 @@ async fn _do_recover_from_snapshot( let temp_storage_path = toc.optional_temp_or_storage_temp_path()?; let tmp_collection_dir = tempfile::Builder::new() - .prefix(&format!("col-{collection_name}-recovery-")) + .prefix(&format!("col-{collection_pass}-recovery-")) .tempdir_in(temp_storage_path)?; log::debug!( - "Recovering collection {collection_name} from snapshot {}", + "Recovering collection {collection_pass} from snapshot {}", snapshot_path.display(), ); @@ -134,19 +133,19 @@ async fn _do_recover_from_snapshot( let snapshot_config = CollectionConfig::load(tmp_collection_dir.path())?; snapshot_config.validate_and_warn(); - let collection = match toc.get_collection(collection_name).await.ok() { + let collection = match toc.get_collection_by_pass(&collection_pass).await.ok() { Some(collection) => collection, None => { - log::debug!("Collection {} does not exist, creating it", collection_name); + log::debug!("Collection {collection_pass} does not exist, creating it"); let operation = CollectionMetaOperations::CreateCollection(CreateCollectionOperation::new( - collection_name.to_string(), + collection_pass.to_string(), snapshot_config.clone().into(), )); dispatcher .submit_collection_meta_op(operation, Access::full(), None) .await?; - toc.get_collection(collection_name).await? + toc.get_collection_by_pass(&collection_pass).await? } }; @@ -176,7 +175,7 @@ async fn _do_recover_from_snapshot( Some(state) => { if state != &ReplicaState::Partial { toc.send_set_replica_state_proposal( - collection_name.to_string(), + collection_pass.to_string(), this_peer_id, *shard_id, ReplicaState::Partial, @@ -266,13 +265,13 @@ async fn _do_recover_from_snapshot( // Don't need more replicas, remove this one toc.request_remove_replica( - collection_name.to_string(), + collection_pass.to_string(), *shard_id, *peer_id, )?; } else { toc.send_set_replica_state_proposal( - collection_name.to_string(), + collection_pass.to_string(), *peer_id, *shard_id, ReplicaState::Dead, @@ -289,13 +288,13 @@ async fn _do_recover_from_snapshot( log::debug!( "Running synchronization for shard {} of collection {} from {}", shard_id, - collection_name, + collection_pass, replica_peer_id ); // assume that if there is another peers, the server is distributed toc.request_shard_transfer( - collection_name.to_string(), + collection_pass.to_string(), *shard_id, *replica_peer_id, this_peer_id, diff --git a/lib/storage/src/content_manager/toc/mod.rs b/lib/storage/src/content_manager/toc/mod.rs index cea7acdd9f..98421dd63e 100644 --- a/lib/storage/src/content_manager/toc/mod.rs +++ b/lib/storage/src/content_manager/toc/mod.rs @@ -44,6 +44,7 @@ use crate::content_manager::collections_ops::{Checker, Collections}; use crate::content_manager::consensus::operation_sender::OperationSender; use crate::content_manager::errors::StorageError; use crate::content_manager::shard_distribution::ShardDistributionProposal; +use crate::rbac::access::CollectionPass; use crate::types::{PeerAddressById, StorageConfig}; use crate::ConsensusOperations; @@ -242,6 +243,13 @@ impl TableOfContent { })) } + pub async fn get_collection_by_pass<'a>( + &self, + collection: &CollectionPass<'a>, + ) -> Result, StorageError> { + self.get_collection(collection.name()).await + } + async fn get_collection_opt( &self, collection_name: String, diff --git a/lib/storage/src/content_manager/toc/point_ops.rs b/lib/storage/src/content_manager/toc/point_ops.rs index e847901eb8..64866eff06 100644 --- a/lib/storage/src/content_manager/toc/point_ops.rs +++ b/lib/storage/src/content_manager/toc/point_ops.rs @@ -14,7 +14,6 @@ use futures::TryStreamExt as _; use segment::types::{ScoredPoint, ShardKey}; use super::TableOfContent; -use crate::content_manager::claims::{check_collection_name, check_points_op}; use crate::content_manager::errors::StorageError; use crate::rbac::access::Access; @@ -38,14 +37,10 @@ impl TableOfContent { access: Access, timeout: Option, ) -> Result, StorageError> { - let Access { - collections, - payload, - } = &access; - check_collection_name(collections.as_ref(), collection_name)?; - check_points_op(collections.as_ref(), payload.as_ref(), &mut request)?; + let collection_pass = access.check_partial_collection_rights(collection_name)?; + access.check_point_op(&mut request)?; - let collection = self.get_collection(collection_name).await?; + let collection = self.get_collection_by_pass(&collection_pass).await?; recommendations::recommend_by( request, &collection, @@ -76,16 +71,12 @@ impl TableOfContent { access: Access, timeout: Option, ) -> Result>, StorageError> { - let Access { - collections, - payload, - } = &access; - check_collection_name(collections.as_ref(), collection_name)?; + let collection_pass = access.check_partial_collection_rights(collection_name)?; for (request, _shard_selector) in &mut requests { - check_points_op(collections.as_ref(), payload.as_ref(), request)?; + access.check_point_op(request)?; } - let collection = self.get_collection(collection_name).await?; + let collection = self.get_collection_by_pass(&collection_pass).await?; recommendations::recommend_batch_by( requests, &collection, @@ -120,16 +111,12 @@ impl TableOfContent { access: Access, timeout: Option, ) -> Result>, StorageError> { - let Access { - collections, - payload, - } = &access; - check_collection_name(collections.as_ref(), collection_name)?; - for req in &mut request.searches { - check_points_op(collections.as_ref(), payload.as_ref(), req)?; + let collection_pass = access.check_partial_collection_rights(collection_name)?; + for request in &mut request.searches { + access.check_point_op(request)?; } - let collection = self.get_collection(collection_name).await?; + let collection = self.get_collection_by_pass(&collection_pass).await?; collection .core_search_batch(request, read_consistency, shard_selection, timeout) .await @@ -156,14 +143,10 @@ impl TableOfContent { shard_selection: ShardSelectorInternal, access: Access, ) -> Result { - let Access { - collections, - payload, - } = &access; - check_collection_name(collections.as_ref(), collection_name)?; - check_points_op(collections.as_ref(), payload.as_ref(), &mut request)?; + let collection_pass = access.check_partial_collection_rights(collection_name)?; + access.check_point_op(&mut request)?; - let collection = self.get_collection(collection_name).await?; + let collection = self.get_collection_by_pass(&collection_pass).await?; collection .count(request, read_consistency, &shard_selection) .await @@ -189,14 +172,10 @@ impl TableOfContent { shard_selection: ShardSelectorInternal, access: Access, ) -> Result, StorageError> { - let Access { - collections, - payload, - } = &access; - check_collection_name(collections.as_ref(), collection_name)?; - check_points_op(collections.as_ref(), payload.as_ref(), &mut request)?; + let collection_pass = access.check_partial_collection_rights(collection_name)?; + access.check_point_op(&mut request)?; - let collection = self.get_collection(collection_name).await?; + let collection = self.get_collection_by_pass(&collection_pass).await?; collection .retrieve(request, read_consistency, &shard_selection) .await @@ -212,14 +191,10 @@ impl TableOfContent { access: Access, timeout: Option, ) -> Result { - let Access { - collections, - payload, - } = &access; - check_collection_name(collections.as_ref(), collection_name)?; - check_points_op(collections.as_ref(), payload.as_ref(), &mut request)?; + let collection_pass = access.check_partial_collection_rights(collection_name)?; + access.check_point_op(&mut request)?; - let collection = self.get_collection(collection_name).await?; + let collection = self.get_collection_by_pass(&collection_pass).await?; let collection_by_name = |name| self.get_collection_opt(name); @@ -244,14 +219,10 @@ impl TableOfContent { access: Access, timeout: Option, ) -> Result, StorageError> { - let Access { - collections, - payload, - } = &access; - check_collection_name(collections.as_ref(), collection_name)?; - check_points_op(collections.as_ref(), payload.as_ref(), &mut request)?; + let collection_pass = access.check_partial_collection_rights(collection_name)?; + access.check_point_op(&mut request)?; - let collection = self.get_collection(collection_name).await?; + let collection = self.get_collection_by_pass(&collection_pass).await?; discovery::discover( request, &collection, @@ -272,15 +243,12 @@ impl TableOfContent { access: Access, timeout: Option, ) -> Result>, StorageError> { - let Access { - collections, - payload, - } = &access; - check_collection_name(collections.as_ref(), collection_name)?; + let collection_pass = access.check_partial_collection_rights(collection_name)?; for (request, _shard_selector) in &mut requests { - check_points_op(collections.as_ref(), payload.as_ref(), request)?; + access.check_point_op(request)?; } - let collection = self.get_collection(collection_name).await?; + + let collection = self.get_collection_by_pass(&collection_pass).await?; discovery::discover_batch( requests, @@ -312,14 +280,10 @@ impl TableOfContent { shard_selection: ShardSelectorInternal, access: Access, ) -> Result { - let Access { - collections, - payload, - } = &access; - check_collection_name(collections.as_ref(), collection_name)?; - check_points_op(collections.as_ref(), payload.as_ref(), &mut request)?; + let collection_pass = access.check_partial_collection_rights(collection_name)?; + access.check_point_op(&mut request)?; - let collection = self.get_collection(collection_name).await?; + let collection = self.get_collection_by_pass(&collection_pass).await?; collection .scroll_by(request, read_consistency, &shard_selection) .await @@ -369,21 +333,13 @@ impl TableOfContent { shard_selector: ShardSelectorInternal, access: Access, ) -> Result { - let Access { - collections, - payload, - } = &access; - check_collection_name(collections.as_ref(), collection_name)?; - check_points_op( - collections.as_ref(), - payload.as_ref(), - &mut operation.operation, - )?; + let collection_pass = access.check_partial_collection_rights(collection_name)?; + access.check_point_op(&mut operation.operation)?; // `TableOfContent::_update_shard_keys` and `Collection::update_from_*` are cancel safe, // so this method is cancel safe. - let collection = self.get_collection(collection_name).await?; + let collection = self.get_collection_by_pass(&collection_pass).await?; // Ordered operation flow: // diff --git a/lib/storage/src/content_manager/toc/snapshots.rs b/lib/storage/src/content_manager/toc/snapshots.rs index 6fc13179e1..aa15ca9e10 100644 --- a/lib/storage/src/content_manager/toc/snapshots.rs +++ b/lib/storage/src/content_manager/toc/snapshots.rs @@ -10,6 +10,7 @@ use super::TableOfContent; use crate::content_manager::consensus::operation_sender::OperationSender; use crate::content_manager::consensus_ops::ConsensusOperations; use crate::content_manager::errors::StorageError; +use crate::rbac::access::CollectionPass; impl TableOfContent { pub fn get_snapshots_storage_manager(&self) -> SnapshotStorageManager { @@ -47,11 +48,11 @@ impl TableOfContent { Ok(snapshots_path) } - pub async fn create_snapshot( + pub async fn create_snapshot<'a>( &self, - collection_name: &str, + collection: &CollectionPass<'a>, ) -> Result { - let collection = self.get_collection(collection_name).await?; + let collection = self.get_collection_by_pass(collection).await?; // We want to use temp dir inside the temp_path (storage if not specified), because it is possible, that // snapshot directory is mounted as network share and multiple writes to it could be slow let temp_dir = self.optional_temp_or_storage_temp_path()?; diff --git a/lib/storage/src/dispatcher.rs b/lib/storage/src/dispatcher.rs index 560a6f9ede..88d896681d 100644 --- a/lib/storage/src/dispatcher.rs +++ b/lib/storage/src/dispatcher.rs @@ -6,9 +6,6 @@ use std::time::{Duration, Instant}; use collection::config::ShardingMethod; use common::defaults::CONSENSUS_META_OP_WAIT; -use crate::content_manager::claims::{ - check_collection_name, incompatible_with_collection_claim, incompatible_with_payload_claim, -}; use crate::content_manager::collection_meta_ops::AliasOperations; use crate::content_manager::shard_distribution::ShardDistributionProposal; use crate::rbac::access::Access; @@ -50,41 +47,11 @@ impl Dispatcher { /// This function needs to be called from a runtime with timers enabled. pub async fn submit_collection_meta_op( &self, - mut operation: CollectionMetaOperations, + operation: CollectionMetaOperations, access: Access, wait_timeout: Option, ) -> Result { - let Access { - collections, - payload, - } = &access; - match &mut operation { - CollectionMetaOperations::CreateCollection(_) - | CollectionMetaOperations::UpdateCollection(_) - | CollectionMetaOperations::DeleteCollection(_) - | CollectionMetaOperations::ChangeAliases(_) - | CollectionMetaOperations::TransferShard(_, _) - | CollectionMetaOperations::SetShardReplicaState(_) - | CollectionMetaOperations::CreateShardKey(_) - | CollectionMetaOperations::DropShardKey(_) => { - if collections.is_some() { - return incompatible_with_collection_claim(); - } - } - CollectionMetaOperations::CreatePayloadIndex(op) => { - check_collection_name(collections.as_ref(), &op.collection_name)?; - if payload.is_some() { - return incompatible_with_payload_claim(); - } - } - CollectionMetaOperations::DropPayloadIndex(op) => { - check_collection_name(collections.as_ref(), &op.collection_name)?; - if payload.is_some() { - return incompatible_with_payload_claim(); - } - } - CollectionMetaOperations::Nop { token: _ } => (), - } + access.check_collection_meta_operation(&operation)?; // if distributed deployment is enabled if let Some(state) = self.consensus_state.as_ref() { diff --git a/lib/storage/src/rbac/access.rs b/lib/storage/src/rbac/access.rs index ba857d92e8..491c5f180b 100644 --- a/lib/storage/src/rbac/access.rs +++ b/lib/storage/src/rbac/access.rs @@ -1,9 +1,16 @@ +use std::borrow::Cow; use std::collections::HashMap; use segment::json_path::JsonPath; use segment::types::ValueVariants; +use serde::{Deserialize, Serialize}; -#[derive(Clone)] +use crate::content_manager::claims::{ + check_collection_name, incompatible_with_collection_claim, incompatible_with_payload_claim, +}; +use crate::content_manager::errors::StorageError; + +#[derive(Serialize, Deserialize, PartialEq, Clone, Debug)] pub struct Access { /// Collection names that are allowed to be accessed pub collections: Option>, @@ -21,6 +28,81 @@ impl Access { payload: None, } } + + /// Check if a access object is allowed to manage everything. + pub fn check_manage_rights(&self) -> Result { + let Access { + collections, + payload, + } = self; + if collections.is_some() { + return incompatible_with_collection_claim(); + } + if payload.is_some() { + return incompatible_with_payload_claim(); + } + Ok(CollectionMultipass) + } + + /// Check if a claim object has access to the whole collection. + pub fn check_whole_collection_rights<'a>( + &self, + collection_name: &'a str, + ) -> Result, StorageError> { + let Access { + collections, + payload, + } = self; + if let Some(collections) = collections { + check_collection_name(Some(collections), collection_name)?; + } + if payload.is_some() { + return incompatible_with_payload_claim(); + } + Ok(CollectionPass(Cow::Borrowed(collection_name))) + } + + /// Check if a claim object has read access to a collection. + pub fn check_partial_collection_rights<'a>( + &self, + collection_name: &'a str, + ) -> Result, StorageError> { + let Access { + collections, + payload: _, + } = self; + if let Some(collections) = collections { + check_collection_name(Some(collections), collection_name)?; + } + Ok(CollectionPass(Cow::Borrowed(collection_name))) + } +} + +pub struct CollectionMultipass; + +impl CollectionMultipass { + pub fn issue_pass<'a>(&self, name: &'a str) -> CollectionPass<'a> { + CollectionPass(Cow::Borrowed(name)) + } +} + +/// A pass that allows access to a specific collection. +pub struct CollectionPass<'a>(Cow<'a, str>); + +impl<'a> CollectionPass<'a> { + pub fn name(&'a self) -> &'a str { + &self.0 + } + + pub fn into_static(self) -> CollectionPass<'static> { + CollectionPass(Cow::Owned(self.0.into_owned())) + } +} + +impl std::fmt::Display for CollectionPass<'_> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } } pub type PayloadClaim = HashMap; diff --git a/lib/storage/src/rbac/mod.rs b/lib/storage/src/rbac/mod.rs index 9e1901e444..0c637eaab1 100644 --- a/lib/storage/src/rbac/mod.rs +++ b/lib/storage/src/rbac/mod.rs @@ -5,3 +5,5 @@ pub mod access_token; pub mod collection_access; #[allow(dead_code)] pub mod error; + +mod ops_checks; diff --git a/lib/storage/src/rbac/ops_checks.rs b/lib/storage/src/rbac/ops_checks.rs new file mode 100644 index 0000000000..da5439a684 --- /dev/null +++ b/lib/storage/src/rbac/ops_checks.rs @@ -0,0 +1,44 @@ +use super::access::Access; +use crate::content_manager::claims::{incompatible_with_collection_claim, PointsOpClaimsChecker}; +use crate::content_manager::collection_meta_ops::CollectionMetaOperations; +use crate::content_manager::errors::StorageError; + +impl Access { + pub fn check_point_op(&self, op: &mut impl PointsOpClaimsChecker) -> Result<(), StorageError> { + for collection in op.collections_used() { + self.check_partial_collection_rights(collection)?; + } + if let Some(payload) = self.payload.as_ref() { + op.apply_payload_claim(payload)?; + } + Ok(()) + } + + pub fn check_collection_meta_operation( + &self, + operation: &CollectionMetaOperations, + ) -> Result<(), StorageError> { + match operation { + CollectionMetaOperations::CreateCollection(_) + | CollectionMetaOperations::UpdateCollection(_) + | CollectionMetaOperations::DeleteCollection(_) + | CollectionMetaOperations::ChangeAliases(_) + | CollectionMetaOperations::TransferShard(_, _) + | CollectionMetaOperations::SetShardReplicaState(_) + | CollectionMetaOperations::CreateShardKey(_) + | CollectionMetaOperations::DropShardKey(_) => { + if self.collections.is_some() { + return incompatible_with_collection_claim(); + } + } + CollectionMetaOperations::CreatePayloadIndex(op) => { + self.check_whole_collection_rights(&op.collection_name)?; + } + CollectionMetaOperations::DropPayloadIndex(op) => { + self.check_whole_collection_rights(&op.collection_name)?; + } + CollectionMetaOperations::Nop { token: _ } => (), + } + Ok(()) + } +} diff --git a/src/actix/api/cluster_api.rs b/src/actix/api/cluster_api.rs index 2388118736..7b6d524951 100644 --- a/src/actix/api/cluster_api.rs +++ b/src/actix/api/cluster_api.rs @@ -3,7 +3,6 @@ use std::future::Future; use actix_web::{delete, get, post, web, HttpResponse}; use actix_web_validator::Query; use serde::Deserialize; -use storage::content_manager::claims::check_manage_rights; use storage::content_manager::consensus_ops::ConsensusOperations; use storage::content_manager::errors::StorageError; use storage::content_manager::toc::TableOfContent; @@ -28,7 +27,7 @@ fn cluster_status( ActixAccess(access): ActixAccess, ) -> impl Future { helpers::time(async move { - check_manage_rights(&access)?; + access.check_manage_rights()?; Ok(dispatcher.cluster_status()) }) } @@ -39,7 +38,7 @@ fn recover_current_peer( ActixAccess(access): ActixAccess, ) -> impl Future { helpers::time(async move { - check_manage_rights(&access)?; + access.check_manage_rights()?; toc.request_snapshot()?; Ok(true) }) @@ -53,7 +52,7 @@ fn remove_peer( ActixAccess(access): ActixAccess, ) -> impl Future { helpers::time(async move { - check_manage_rights(&access)?; + access.check_manage_rights()?; let dispatcher = dispatcher.into_inner(); let peer_id = peer_id.into_inner(); diff --git a/src/actix/api/service_api.rs b/src/actix/api/service_api.rs index 86bf57c567..f6487e53bc 100644 --- a/src/actix/api/service_api.rs +++ b/src/actix/api/service_api.rs @@ -11,7 +11,6 @@ use common::types::{DetailsLevel, TelemetryDetail}; use schemars::JsonSchema; use segment::common::anonymize::Anonymize; use serde::{Deserialize, Serialize}; -use storage::content_manager::claims::check_manage_rights; use storage::content_manager::toc::TableOfContent; use tokio::sync::Mutex; @@ -36,7 +35,7 @@ fn telemetry( ActixAccess(access): ActixAccess, ) -> impl Future { helpers::time(async move { - check_manage_rights(&access)?; + access.check_manage_rights()?; let anonymize = params.anonymize.unwrap_or(false); let details_level = params .details_level @@ -67,7 +66,7 @@ async fn metrics( params: Query, ActixAccess(access): ActixAccess, ) -> HttpResponse { - if let Err(err) = check_manage_rights(&access) { + if let Err(err) = access.check_manage_rights() { return process_response_error(err, Instant::now()); } @@ -97,7 +96,7 @@ fn put_locks( ActixAccess(access): ActixAccess, ) -> impl Future { helpers::time(async move { - check_manage_rights(&access)?; + access.check_manage_rights()?; let result = LocksOption { write: toc.get_ref().is_write_locked(), error_message: toc.get_ref().get_lock_error_message(), @@ -114,7 +113,7 @@ fn get_locks( ActixAccess(access): ActixAccess, ) -> impl Future { helpers::time(async move { - check_manage_rights(&access)?; + access.check_manage_rights()?; let result = LocksOption { write: toc.get_ref().is_write_locked(), error_message: toc.get_ref().get_lock_error_message(), @@ -126,7 +125,7 @@ fn get_locks( #[get("/stacktrace")] fn get_stacktrace(ActixAccess(access): ActixAccess) -> impl Future { helpers::time(async move { - check_manage_rights(&access)?; + access.check_manage_rights()?; Ok(get_stack_trace()) }) } diff --git a/src/actix/api/snapshot_api.rs b/src/actix/api/snapshot_api.rs index bbdd4b33be..6835142750 100644 --- a/src/actix/api/snapshot_api.rs +++ b/src/actix/api/snapshot_api.rs @@ -16,7 +16,6 @@ use futures::{FutureExt as _, TryFutureExt as _}; use reqwest::Url; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; -use storage::content_manager::claims::{check_full_access_to_collection, check_manage_rights}; use storage::content_manager::errors::StorageError; use storage::content_manager::snapshots::recover::do_recover_from_snapshot; use storage::content_manager::snapshots::{ @@ -70,7 +69,7 @@ pub async fn do_get_full_snapshot( access: Access, snapshot_name: &str, ) -> Result { - check_manage_rights(&access)?; + access.check_manage_rights()?; let file_name = get_full_snapshot_path(toc, snapshot_name).await?; Ok(NamedFile::open(file_name)?) } @@ -124,8 +123,8 @@ pub async fn do_get_snapshot( collection_name: &str, snapshot_name: &str, ) -> Result { - check_full_access_to_collection(&access, collection_name)?; - let collection = toc.get_collection(collection_name).await?; + let collection_pass = access.check_whole_collection_rights(collection_name)?; + let collection = toc.get_collection_by_pass(&collection_pass).await?; let file_name = collection.get_snapshot_path(snapshot_name).await?; Ok(NamedFile::open(file_name)?) } @@ -169,7 +168,7 @@ async fn upload_snapshot( helpers::time_or_accept_with_handle(params.wait.unwrap_or(true), async move { let snapshot = form.snapshot; - check_manage_rights(&access)?; + access.check_manage_rights()?; if let Some(checksum) = ¶ms.checksum { let snapshot_checksum = hash_file(snapshot.file.path()).await?; @@ -379,7 +378,7 @@ async fn upload_shard_snapshot( let future = cancel::future::spawn_cancel_on_drop(move |cancel| async move { // TODO: Run this check before the multipart blob is uploaded - check_manage_rights(&access)?; + let multipass = access.check_manage_rights()?; if let Some(checksum) = checksum { let snapshot_checksum = hash_file(form.snapshot.file.path()).await?; @@ -389,7 +388,9 @@ async fn upload_shard_snapshot( } let future = async { - let collection = toc.get_collection(&collection).await?; + let collection = toc + .get_collection_by_pass(&multipass.issue_pass(&collection)) + .await?; collection.assert_shard_exists(shard).await?; Result::<_, StorageError>::Ok(collection) @@ -422,8 +423,8 @@ async fn download_shard_snapshot( ActixAccess(access): ActixAccess, ) -> Result { let (collection, shard, snapshot) = path.into_inner(); - check_full_access_to_collection(&access, &collection)?; - let collection = toc.get_collection(&collection).await?; + let collection_pass = access.check_whole_collection_rights(&collection)?; + let collection = toc.get_collection_by_pass(&collection_pass).await?; let snapshot_path = collection.get_shard_snapshot_path(shard, &snapshot).await?; Ok(NamedFile::open(snapshot_path)) diff --git a/src/common/auth/claims.rs b/src/common/auth/claims.rs index 120d3fd3fa..99fe34b9ce 100644 --- a/src/common/auth/claims.rs +++ b/src/common/auth/claims.rs @@ -1,7 +1,7 @@ use segment::json_path::JsonPath; use segment::types::{Condition, FieldCondition, Filter, Match, ValueVariants}; use serde::{Deserialize, Serialize}; -use storage::rbac::access::PayloadClaim; +use storage::rbac::access::Access; #[derive(Serialize, Deserialize, PartialEq, Clone, Debug)] pub struct Claims { @@ -11,12 +11,8 @@ pub struct Claims { /// Write access, default is false. Read access is always enabled with a valid token. pub w: Option, - /// Collection names that are allowed to be accessed - pub collections: Option>, - - /// Payload constraints. - /// An object where each key is a JSON path, and each value is JSON value. - pub payload: Option, + #[serde(flatten)] + pub access: Access, /// Validate this token by looking for a value inside a collection. pub value_exists: Option, diff --git a/src/common/auth/jwt_parser.rs b/src/common/auth/jwt_parser.rs index 781e23a54f..1b68493880 100644 --- a/src/common/auth/jwt_parser.rs +++ b/src/common/auth/jwt_parser.rs @@ -39,6 +39,7 @@ impl JwtParser { #[cfg(test)] mod tests { use segment::types::ValueVariants; + use storage::rbac::access::Access; use super::*; @@ -59,19 +60,21 @@ mod tests { let claims = Claims { exp: Some(exp), w: Some(true), - collections: Some(vec!["collection".to_string()]), - payload: Some( - vec![ - ( - "field1".parse().unwrap(), - ValueVariants::Keyword("value".to_string()), - ), - ("field2".parse().unwrap(), ValueVariants::Integer(42)), - ("field2".parse().unwrap(), ValueVariants::Bool(true)), - ] - .into_iter() - .collect(), - ), + access: Access { + collections: Some(vec!["collection".to_string()]), + payload: Some( + vec![ + ( + "field1".parse().unwrap(), + ValueVariants::Keyword("value".to_string()), + ), + ("field2".parse().unwrap(), ValueVariants::Integer(42)), + ("field2".parse().unwrap(), ValueVariants::Bool(true)), + ] + .into_iter() + .collect(), + ), + }, value_exists: None, }; let token = create_token(&claims); @@ -94,8 +97,10 @@ mod tests { let mut claims = Claims { exp: Some(exp), w: Some(false), - collections: None, - payload: None, + access: Access { + collections: None, + payload: None, + }, value_exists: None, }; diff --git a/src/common/auth/mod.rs b/src/common/auth/mod.rs index 84b88807d1..91db04d358 100644 --- a/src/common/auth/mod.rs +++ b/src/common/auth/mod.rs @@ -90,9 +90,8 @@ impl AuthKeys { let Claims { exp: _, // already validated on decoding w: write_access, + access, value_exists, - collections, - payload, } = claims; if !write_access.unwrap_or(false) && !is_read_only { @@ -103,10 +102,7 @@ impl AuthKeys { self.validate_value_exists(&value_exists).await?; } - return Ok(Some(Access { - collections, - payload, - })); + return Ok(Some(access)); } Err("Invalid API key or JWT".to_string()) diff --git a/src/common/collections.rs b/src/common/collections.rs index f33cd721bb..110fa68bc8 100644 --- a/src/common/collections.rs +++ b/src/common/collections.rs @@ -18,9 +18,7 @@ use collection::shards::shard::{PeerId, ShardId, ShardsPlacement}; use collection::shards::transfer::{ShardTransfer, ShardTransferKey, ShardTransferRestart}; use itertools::Itertools; use rand::prelude::SliceRandom; -use storage::content_manager::claims::{ - check_collection_name, check_full_access_to_collection, incompatible_with_collection_claim, -}; +use storage::content_manager::claims::check_collection_name; use storage::content_manager::collection_meta_ops::ShardTransferOperations::{Abort, Start}; use storage::content_manager::collection_meta_ops::{ CollectionMetaOperations, CreateShardKey, DropShardKey, ShardTransferOperations, @@ -37,15 +35,11 @@ pub async fn do_collection_exists( access: Access, name: &str, ) -> Result { - let Access { - collections, - payload: _, - } = &access; - check_collection_name(collections.as_ref(), name)?; + let collection_pass = access.check_partial_collection_rights(name)?; // if this returns Ok, it means the collection exists. // if not, we check that the error is NotFound - let Err(error) = toc.get_collection(name).await else { + let Err(error) = toc.get_collection_by_pass(&collection_pass).await else { return Ok(CollectionExists { exists: true }); }; match error { @@ -60,13 +54,9 @@ pub async fn do_get_collection( name: &str, shard_selection: Option, ) -> Result { - let Access { - collections, - payload: _, - } = &access; - check_collection_name(collections.as_ref(), name)?; + let collection_pass = access.check_partial_collection_rights(name)?; - let collection = toc.get_collection(name).await?; + let collection = toc.get_collection_by_pass(&collection_pass).await?; let shard_selection = match shard_selection { None => ShardSelectorInternal::All, @@ -77,21 +67,11 @@ pub async fn do_get_collection( } pub async fn do_list_collections(toc: &TableOfContent, access: Access) -> CollectionsResponse { - let access_collections = { - let Access { - collections, - payload: _, - } = &access; - collections.as_ref() - }; - let collections = toc .all_collections() .await .into_iter() - .filter(|c| { - access_collections.map_or(true, |access_collections| access_collections.contains(c)) - }) + .filter(|c| access.check_partial_collection_rights(c).is_ok()) .map(|name| CollectionDescription { name }) .collect_vec(); @@ -196,9 +176,9 @@ pub async fn do_list_snapshots( access: Access, collection_name: &str, ) -> Result, StorageError> { - check_full_access_to_collection(&access, collection_name)?; + let collection_pass = access.check_whole_collection_rights(collection_name)?; Ok(toc - .get_collection(collection_name) + .get_collection_by_pass(&collection_pass) .await? .list_snapshots() .await?) @@ -209,11 +189,12 @@ pub fn do_create_snapshot( access: Access, collection_name: &str, ) -> Result>, StorageError> { - check_full_access_to_collection(&access, collection_name)?; - let collection = collection_name.to_string(); + let collection_pass = access + .check_whole_collection_rights(collection_name)? + .into_static(); let dispatcher = dispatcher.clone(); Ok(tokio::spawn(async move { - dispatcher.create_snapshot(&collection).await + dispatcher.create_snapshot(&collection_pass).await })) } @@ -222,13 +203,8 @@ pub async fn do_get_collection_cluster( access: Access, name: &str, ) -> Result { - let Access { - collections, - payload: _, - } = &access; - check_collection_name(collections.as_ref(), name)?; - - let collection = toc.get_collection(name).await?; + let collection_pass = access.check_partial_collection_rights(name)?; + let collection = toc.get_collection_by_pass(&collection_pass).await?; Ok(collection.cluster_info(toc.this_peer_id).await?) } @@ -239,34 +215,26 @@ pub async fn do_update_collection_cluster( access: Access, wait_timeout: Option, ) -> Result { - let Access { - collections, - payload: _, - } = &access; - - check_collection_name(collections.as_ref(), &collection_name)?; + let collection_pass = access.check_whole_collection_rights(&collection_name)?; match &operation { ClusterOperations::MoveShard(_) | ClusterOperations::ReplicateShard(_) | ClusterOperations::AbortTransfer(_) | ClusterOperations::DropReplica(_) | ClusterOperations::RestartTransfer(_) => { - if collections.is_some() { - return incompatible_with_collection_claim(); - } + access.check_manage_rights()?; } ClusterOperations::CreateShardingKey(CreateShardingKeyOperation { create_sharding_key, }) => { - if collections.is_some() && !create_sharding_key.has_default_params() { - return incompatible_with_collection_claim(); + if !create_sharding_key.has_default_params() { + access.check_manage_rights()?; } } ClusterOperations::DropShardingKey(DropShardingKeyOperation { drop_sharding_key: DropShardingKey { shard_key: _ }, }) => (), } - let full_access = Access::full(); // We already checked the request above if dispatcher.consensus_state().is_none() { @@ -302,7 +270,7 @@ pub async fn do_update_collection_cluster( Ok(()) }; - let collection = dispatcher.get_collection(&collection_name).await?; + let collection = dispatcher.get_collection_by_pass(&collection_pass).await?; match operation { ClusterOperations::MoveShard(MoveShardOperation { move_shard }) => { diff --git a/src/common/snapshots.rs b/src/common/snapshots.rs index f4baba1e7e..1696137084 100644 --- a/src/common/snapshots.rs +++ b/src/common/snapshots.rs @@ -9,7 +9,6 @@ use collection::operations::snapshot_ops::{ }; use collection::shards::replica_set::ReplicaState; use collection::shards::shard::ShardId; -use storage::content_manager::claims::check_full_access_to_collection; use storage::content_manager::errors::StorageError; use storage::content_manager::snapshots; use storage::content_manager::toc::TableOfContent; @@ -26,8 +25,8 @@ pub async fn create_shard_snapshot( collection_name: String, shard_id: ShardId, ) -> Result { - check_full_access_to_collection(&access, &collection_name)?; - let collection = toc.get_collection(&collection_name).await?; + let collection_pass = access.check_whole_collection_rights(&collection_name)?; + let collection = toc.get_collection_by_pass(&collection_pass).await?; let snapshot = collection .create_shard_snapshot(shard_id, &toc.optional_temp_or_snapshot_temp_path()?) @@ -45,8 +44,8 @@ pub async fn list_shard_snapshots( collection_name: String, shard_id: ShardId, ) -> Result, StorageError> { - check_full_access_to_collection(&access, &collection_name)?; - let collection = toc.get_collection(&collection_name).await?; + let collection_pass = access.check_whole_collection_rights(&collection_name)?; + let collection = toc.get_collection_by_pass(&collection_pass).await?; let snapshots = collection.list_shard_snapshots(shard_id).await?; Ok(snapshots) } @@ -61,8 +60,8 @@ pub async fn delete_shard_snapshot( shard_id: ShardId, snapshot_name: String, ) -> Result<(), StorageError> { - check_full_access_to_collection(&access, &collection_name)?; - let collection = toc.get_collection(&collection_name).await?; + let collection_pass = access.check_whole_collection_rights(&collection_name)?; + let collection = toc.get_collection_by_pass(&collection_pass).await?; let snapshot_path = collection .get_shard_snapshot_path(shard_id, &snapshot_name) .await?; @@ -88,7 +87,9 @@ pub async fn recover_shard_snapshot( checksum: Option, client: HttpClient, ) -> Result<(), StorageError> { - check_full_access_to_collection(&access, &collection_name)?; + let collection_pass = access + .check_whole_collection_rights(&collection_name)? + .into_static(); // - `download_dir` handled by `tempfile` and would be deleted, if request is cancelled // - remote snapshot is downloaded into `download_dir` and would be deleted with it @@ -97,7 +98,7 @@ pub async fn recover_shard_snapshot( cancel::future::spawn_cancel_on_drop(move |cancel| async move { let future = async { - let collection = toc.get_collection(&collection_name).await?; + let collection = toc.get_collection_by_pass(&collection_pass).await?; collection.assert_shard_exists(shard_id).await?; let download_dir = toc.snapshots_download_tempdir()?; diff --git a/src/migrations/single_to_cluster.rs b/src/migrations/single_to_cluster.rs index 91d7f12cfb..c9b087880f 100644 --- a/src/migrations/single_to_cluster.rs +++ b/src/migrations/single_to_cluster.rs @@ -24,10 +24,16 @@ pub async fn handle_existing_collections( collections: Vec, ) { let full_access = Access::full(); + let multipass = full_access + .check_manage_rights() + .expect("Full access should have manage rights"); consensus_state.is_leader_established.await_ready(); for collection_name in collections { - let collection_obj = match toc_arc.get_collection(&collection_name).await { + let collection_obj = match toc_arc + .get_collection_by_pass(&multipass.issue_pass(&collection_name)) + .await + { Ok(collection_obj) => collection_obj, Err(_) => break, }; diff --git a/src/tonic/auth.rs b/src/tonic/auth.rs index ad41822de2..7d1219caf3 100644 --- a/src/tonic/auth.rs +++ b/src/tonic/auth.rs @@ -106,10 +106,9 @@ impl Layer for AuthLayer { } pub fn extract_access(req: &mut tonic::Request) -> Access { - req.extensions_mut().remove::().unwrap_or(Access { - collections: None, - payload: None, - }) + req.extensions_mut() + .remove::() + .unwrap_or(Access::full()) } fn is_read_only(req: &tonic::codegen::http::Request) -> bool {