upstream-archive/qdrant
1
0
Fork 0
mirror of https://github.com/qdrant/qdrant.git synced 2026-08-04 17:10:59 -05:00
Code Issues Packages Projects Releases Wiki Activity

RBAC: introduce CollectionPass (#3932)

Browse Source
This commit is contained in:
xzfc
2024-03-28 12:46:12 +00:00
committed by GitHub
parent 99db4b8ca0
commit 70ba0471a3
20 changed files with 285 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
lib/storage/src/content_manager/claims.rs
View File

@@ -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<String>>,
@@ -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<String>>,
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.

22
lib/storage/src/content_manager/snapshots/mod.rs
View File

@@ -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<JoinHandle<Result<bool, StorageError>>, 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<JoinHandle<Result<bool, StorageError>>, 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<Vec<SnapshotDescription>, 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<JoinHandle<Result<SnapshotDescription, StorageError>>, 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<SnapshotDescription, StorageError> {
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();

33
lib/storage/src/content_manager/snapshots/recover.rs
View File

@@ -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<JoinHandle<Result<bool, StorageError>>, 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<bool, StorageError> {
@@ -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,

8
lib/storage/src/content_manager/toc/mod.rs
View File

@@ -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<RwLockReadGuard<Collection>, StorageError> {
self.get_collection(collection.name()).await
}
async fn get_collection_opt(
&self,
collection_name: String,

108
lib/storage/src/content_manager/toc/point_ops.rs
View File

@@ -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<Duration>,
) -> Result<Vec<ScoredPoint>, 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<Duration>,
) -> Result<Vec<Vec<ScoredPoint>>, 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<Duration>,
) -> Result<Vec<Vec<ScoredPoint>>, 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<CountResult, 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
.count(request, read_consistency, &shard_selection)
.await
@@ -189,14 +172,10 @@ impl TableOfContent {
shard_selection: ShardSelectorInternal,
access: Access,
) -> Result<Vec<Record>, 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<Duration>,
) -> Result<GroupsResult, 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?;
let collection_by_name = |name| self.get_collection_opt(name);
@@ -244,14 +219,10 @@ impl TableOfContent {
access: Access,
timeout: Option<Duration>,
) -> Result<Vec<ScoredPoint>, 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<Duration>,
) -> Result<Vec<Vec<ScoredPoint>>, 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<ScrollResult, 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
.scroll_by(request, read_consistency, &shard_selection)
.await
@@ -369,21 +333,13 @@ impl TableOfContent {
shard_selector: ShardSelectorInternal,
access: Access,
) -> Result<UpdateResult, StorageError> {
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:
//

7
lib/storage/src/content_manager/toc/snapshots.rs
View File

@@ -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<SnapshotDescription, StorageError> {
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()?;

37
lib/storage/src/dispatcher.rs
View File

@@ -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<Duration>,
) -> Result<bool, StorageError> {
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() {

84
lib/storage/src/rbac/access.rs
View File

@@ -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<Vec<String>>,
@@ -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<CollectionMultipass, StorageError> {
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<CollectionPass<'a>, 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<CollectionPass<'a>, 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<JsonPath, ValueVariants>;

2
lib/storage/src/rbac/mod.rs
View File

@@ -5,3 +5,5 @@ pub mod access_token;
pub mod collection_access;
#[allow(dead_code)]
pub mod error;
mod ops_checks;

44
lib/storage/src/rbac/ops_checks.rs Normal file
View File

@@ -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(())
}
}

7
src/actix/api/cluster_api.rs
View File

@@ -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<Output = HttpResponse> {
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<Output = HttpResponse> {
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<Output = HttpResponse> {
helpers::time(async move {
check_manage_rights(&access)?;
access.check_manage_rights()?;
let dispatcher = dispatcher.into_inner();
let peer_id = peer_id.into_inner();

11
src/actix/api/service_api.rs
View File

@@ -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<Output = HttpResponse> {
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<MetricsParam>,
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<Output = HttpResponse> {
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<Output = HttpResponse> {
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<Output = HttpResponse> {
helpers::time(async move {
check_manage_rights(&access)?;
access.check_manage_rights()?;
Ok(get_stack_trace())
})
}

19
src/actix/api/snapshot_api.rs
View File

@@ -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<NamedFile, HttpError> {
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<NamedFile, HttpError> {
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) = &params.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<impl Responder, HttpError> {
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))

10
src/common/auth/claims.rs
View File

@@ -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<bool>,
/// Collection names that are allowed to be accessed
pub collections: Option<Vec<String>>,
/// Payload constraints.
/// An object where each key is a JSON path, and each value is JSON value.
pub payload: Option<PayloadClaim>,
#[serde(flatten)]
pub access: Access,
/// Validate this token by looking for a value inside a collection.
pub value_exists: Option<ValueExists>,

35
src/common/auth/jwt_parser.rs
View File

@@ -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,
};

8
src/common/auth/mod.rs
View File

@@ -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())

70
src/common/collections.rs
View File

@@ -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<CollectionExists, StorageError> {
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<ShardId>,
) -> Result<CollectionInfo, StorageError> {
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<Vec<SnapshotDescription>, 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<JoinHandle<Result<SnapshotDescription, StorageError>>, 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<CollectionClusterInfo, StorageError> {
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<Duration>,
) -> Result<bool, StorageError> {
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 }) => {

19
src/common/snapshots.rs
View File

@@ -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<SnapshotDescription, 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 = 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<Vec<SnapshotDescription>, 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<String>,
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()?;

8
src/migrations/single_to_cluster.rs
View File

@@ -24,10 +24,16 @@ pub async fn handle_existing_collections(
collections: Vec<String>,
) {
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,
};

7
src/tonic/auth.rs
View File

@@ -106,10 +106,9 @@ impl<S> Layer<S> for AuthLayer {
}
pub fn extract_access<R>(req: &mut tonic::Request<R>) -> Access {
req.extensions_mut().remove::<Access>().unwrap_or(Access {
collections: None,
payload: None,
})
req.extensions_mut()
.remove::<Access>()
.unwrap_or(Access::full())
}
fn is_read_only<R>(req: &tonic::codegen::http::Request<R>) -> bool {