chore(bors): merge pull request #962

962: feat(encryption): restrict different devlink of same device for pool creation r=Abhinandan-Purkait a=Abhinandan-Purkait

## Issue

Since we can't read the metadata of an encrypted pool, there isn't any definitive way in data-plane to stop a encrypted pool from getting recreated with a different devlink of a same device with a different pool name.

## Solution

Compare the devlinks across all pools on the specified node, if any pool shares a devlink with the requested disk's devlink/devpath/devname then reject the creation.

## Test

Adds a test to validate the scenario by creating an lvm lv on a loop device and using it's devlinks across multiple creations.

Co-authored-by: Abhinandan Purkait <[email protected]>

Author: mayastor-bors

Cargo.lock

@@ -58,6 +58,7 @@ tracing = "0.1.40"
 nix = { version = "0.29.0", default-features = false }
 prost-types = "0.13.3"
 url = "2.5.4"
+regex = "1.11.1"
 
 grpc = { path = "../grpc" }
 shutdown = { path = "../../utils/shutdown" }

control-plane/agents/Cargo.toml

@@ -85,10 +85,6 @@ impl GrpcContext {
         self.lock.clone().lock_owned().await
     }
     /// Use this context to connect a grpc client and return it.
-    pub(crate) async fn connect(&self) -> Result<GrpcClient, SvcError> {
-        GrpcClient::new(self).await
-    }
-    /// Use this context to connect a grpc client and return it.
     /// This client differs from `GrpcClient` from `Self::connect` in that it ensures only 1 request
     /// is sent at a time among all `GrpcClientLocked` clients.
     pub(crate) async fn connect_locked(

control-plane/agents/src/bin/core/controller/io_engine/client.rs

@@ -114,6 +114,8 @@ pub(crate) struct RegistryInner<S: Store> {
     ha_disabled: bool,
     /// Etcd max page size.
     etcd_max_page_size: i64,
+    /// Allow pool creation using non-persistent devlinks.
+    allow_non_persistent_devlinks: bool,
 }
 
 impl Registry {
@@ -140,6 +142,7 @@ impl Registry {
         ha_enabled: bool,
         etcd_max_page_size: i64,
         no_volume_health: bool,
+        allow_non_persistent_devlinks: bool,
     ) -> Result<Self, SvcError> {
         let store_endpoint = Self::format_store_endpoint(&store_url);
         tracing::info!("Connecting to persistent store at {}", store_endpoint);
@@ -199,6 +202,7 @@ impl Registry {
                 thin_args,
                 ha_disabled: ha_enabled,
                 etcd_max_page_size,
+                allow_non_persistent_devlinks,
             }),
         };
         registry.init().await?;
@@ -230,6 +234,11 @@ impl Registry {
         self.ha_disabled
     }
 
+    /// Check if pool creation using non-persistent devlink is allowed.
+    pub(crate) fn allow_non_persistent_devlinks(&self) -> bool {
+        self.allow_non_persistent_devlinks
+    }
+
     /// Check if the partial rebuilds are disabled.
     pub(crate) fn partial_rebuild_disabled(&self) -> bool {
         self.disable_partial_rebuild

control-plane/agents/src/bin/core/controller/registry.rs

@@ -142,6 +142,10 @@ pub(crate) struct CliArgs {
     /// Disable volume health reporting which uses pstor watchers.
     #[clap(long)]
     no_volume_health: bool,
+
+    /// Allow pools to be created using non persistent devlinks.
+    #[clap(long)]
+    allow_non_persistent_devlink: bool,
 }
 impl CliArgs {
     fn args() -> Self {
@@ -219,6 +223,7 @@ async fn server(cli_args: CliArgs) -> anyhow::Result<()> {
         cli_args.disable_ha,
         cli_args.etcd_page_limit as i64,
         cli_args.no_volume_health,
+        cli_args.allow_non_persistent_devlink,
     )
     .await?;
 

control-plane/agents/src/bin/core/main.rs

@@ -13,6 +13,7 @@ use stor_port::types::v0::transport::{
     Deregister, Filter, Node, NodeId, NodeState, NodeStatus, Register,
 };
 
+use crate::controller::io_engine::HostApi;
 use crate::controller::wrapper::InternalOps;
 use grpc::{
     context::Context,
@@ -367,9 +368,7 @@ impl Service {
     ) -> Result<BlockDevices, SvcError> {
         let node = self.registry.node_wrapper(&request.node).await?;
 
-        let grpc = node.read().await.grpc_context()?;
-        let client = grpc.connect().await?;
-        client.list_blockdevices(request).await
+        node.list_blockdevices(request).await
     }
 
     /// Cordon the specified node.

control-plane/agents/src/bin/core/node/service.rs

@@ -1,3 +1,4 @@
+use crate::controller::io_engine::HostApi;
 use crate::{
     controller::{
         io_engine::{
@@ -15,8 +16,11 @@ use crate::{
     NumRebuilds,
 };
 use agents::{errors::SvcError, eventing::EventWithMeta};
+use async_trait::async_trait;
 use events_api::event::{EventAction, EventCategory, EventMessage, EventMeta, EventSource};
 use grpc::operations::snapshot::SnapshotInfo;
+use stor_port::transport_api::v0::BlockDevices;
+use stor_port::types::v0::transport::GetBlockDevices;
 use stor_port::{
     transport_api::{Message, MessageId, ResourceKind},
     types::v0::{
@@ -36,7 +40,6 @@ use stor_port::{
     },
 };
 
-use async_trait::async_trait;
 use parking_lot::RwLock;
 use std::{future::Future, ops::DerefMut, sync::Arc};
 use tracing::{debug, trace, warn};
@@ -1754,3 +1757,16 @@ impl From<&NodeWrapper> for NodeState {
         node.node_state().clone()
     }
 }
+
+#[async_trait]
+impl HostApi for Arc<tokio::sync::RwLock<NodeWrapper>> {
+    async fn liveness_probe(&self) -> Result<Register, SvcError> {
+        let dataplane = self.read().await.grpc_client().await?;
+        dataplane.liveness_probe().await
+    }
+
+    async fn list_blockdevices(&self, request: &GetBlockDevices) -> Result<BlockDevices, SvcError> {
+        let dataplane = self.read().await.grpc_client().await?;
+        dataplane.list_blockdevices(request).await
+    }
+}

control-plane/agents/src/bin/core/node/wrapper.rs

@@ -1,3 +1,5 @@
+use crate::controller::io_engine::HostApi;
+use crate::node::wrapper::NodeWrapper;
 use crate::{
     controller::{
         io_engine::PoolApi,
@@ -11,8 +13,8 @@ use crate::{
     node::wrapper::GetterOps,
 };
 use agents::{errors, errors::SvcError};
-use snafu::OptionExt;
-use std::ops::Deref;
+use stor_port::transport_api::ResourceKind;
+use stor_port::types::v0::transport::{GetBlockDevices, PoolDeviceUri};
 use stor_port::types::v0::{
     store::{
         pool::PoolSpec,
@@ -21,6 +23,14 @@ use stor_port::types::v0::{
     transport::{CreatePool, DestroyReplica, NodeId, ReplicaOwners},
 };
 
+use itertools::Itertools;
+use regex::Regex;
+use snafu::OptionExt;
+use std::collections::HashSet;
+use std::ops::Deref;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+
 impl OperationGuardArc<PoolSpec> {
     /// Retries the creation of the pool which is being done in the background by the io-engine.
     /// This may happen if the pool create gRPC times out, for very large pools.
@@ -130,3 +140,102 @@ impl OperationGuardArc<ReplicaSpec> {
         }
     }
 }
+
+pub(crate) async fn devlink_preflight_checks(
+    request: &CreatePool,
+    node: Arc<RwLock<NodeWrapper>>,
+    registry: &Registry,
+) -> Result<(), SvcError> {
+    let request_disks: HashSet<String> = request
+        .disks
+        .iter()
+        .map(|disk| utils::disk::normalize_disk(disk.as_str()))
+        .collect();
+
+    if !registry.allow_non_persistent_devlinks() {
+        fn is_persistent_devlink(pattern: &str) -> Result<bool, SvcError> {
+            let re = Regex::new(utils::DEVLINK_REGEX).map_err(|_| SvcError::InvalidArguments {})?;
+            Ok(re.is_match(pattern))
+        }
+
+        if request_disks
+            .iter()
+            // Only attempt to validate if it starts with "/dev".
+            .filter(|disk| disk.starts_with("/dev"))
+            .any(|disk| !is_persistent_devlink(disk).is_ok_and(|val| val))
+        {
+            return Err(SvcError::InvalidDevlink {});
+        }
+    }
+
+    let node_pools = registry
+        .get_node_opt_pools(Some(request.node.clone()))
+        .await?;
+
+    if !node_pools.is_empty() {
+        let node_pools_disks: Vec<PoolDeviceUri> = node_pools
+            .into_iter()
+            .filter_map(|pool| pool.spec().map(|spec| spec.disks))
+            .flatten()
+            .collect();
+
+        let node_pools_disks_normalized: HashSet<String> = node_pools_disks
+            .iter()
+            .map(|disk| utils::disk::normalize_disk(disk.as_str()))
+            .collect();
+
+        let common_disks: HashSet<_> = request_disks
+            .intersection(&node_pools_disks_normalized)
+            .cloned()
+            .collect();
+
+        // Same devpaths or devlinks should be rejected.
+        if !common_disks.is_empty() {
+            return Err(SvcError::InUse {
+                kind: ResourceKind::Block,
+                id: common_disks.iter().join(","),
+            });
+        }
+
+        let node_block_devices = node
+            .list_blockdevices(&GetBlockDevices {
+                node: request.node.clone(),
+                all: true,
+            })
+            .await?
+            .into_inner();
+
+        let matched_devices: Vec<_> = node_block_devices
+            .iter()
+            .filter(|device| {
+                request_disks.contains(&device.devname)
+                    || request_disks.contains(&device.devpath)
+                    || device
+                        .devlinks
+                        .iter()
+                        .any(|link| request_disks.contains(link))
+            })
+            .collect();
+
+        // If the requested disk was not found, that could be because it could be a malloc or a file,
+        // in that case ignore and move ahead to allow tests. If it is an actual disk that was not
+        // detected by blockdevice api then the disk might not be visible to io-engine, so let it fail from io-engine
+        // rather than bailing out from control-plane to keep the behaviour as before.
+        if !matched_devices.is_empty() {
+            if let Some(conflict) = matched_devices.iter().find(|bd| {
+                node_pools_disks_normalized.contains(&bd.devname)
+                    || node_pools_disks_normalized.contains(&bd.devpath)
+                    || bd
+                        .devlinks
+                        .iter()
+                        .any(|link| node_pools_disks_normalized.contains(link))
+            }) {
+                return Err(SvcError::InUse {
+                    kind: ResourceKind::Block,
+                    id: conflict.devname.clone(),
+                });
+            }
+        }
+    }
+    Ok(())
+}

control-plane/agents/src/bin/core/pool/operations_helper.rs

@@ -7,8 +7,8 @@ use crate::controller::{
         OperationGuardArc,
     },
 };
+use crate::pool::operations_helper::devlink_preflight_checks;
 use agents::errors::{SvcError, SvcError::CordonedNode};
-use std::collections::HashMap;
 use stor_port::{
     transport_api::ResourceKind,
     types::v0::{
@@ -18,6 +18,8 @@ use stor_port::{
 };
 use utils::dsp_created_by_key;
 
+use std::collections::HashMap;
+
 #[async_trait::async_trait]
 impl ResourceLifecycle for OperationGuardArc<PoolSpec> {
     type Create = CreatePool;
@@ -35,13 +37,15 @@ impl ResourceLifecycle for OperationGuardArc<PoolSpec> {
                 node_id: request.node.to_string(),
             });
         }
+
         if request.disks.len() != 1 {
             return Err(SvcError::InvalidPoolDeviceNum {
                 disks: request.disks.clone(),
             });
         }
 
-        if let Ok(pool) = registry.specs().pool(&request.id) {
+        let pool_get_result = registry.specs().pool(&request.id);
+        if let Ok(pool) = &pool_get_result {
             if pool.status.created() {
                 return Err(SvcError::AlreadyExists {
                     kind: ResourceKind::Pool,
@@ -58,6 +62,11 @@ impl ResourceLifecycle for OperationGuardArc<PoolSpec> {
             });
         }
 
+        if pool_get_result.is_err() {
+            // If the devlink for a device is used for multiple pools, reject new creation.
+            devlink_preflight_checks(request, node.clone(), registry).await?
+        }
+
         let mut pool = specs
             .get_or_create_pool(request)
             .operation_guard_wait()