Skip to content

Commit 4b7ebab

Browse files
TCeasonTCeason
committed
feat(meta): prevent dropping in-use security policies
- raise constraint errors when masking/row access policies remain bound to tables - introduce dedicated masking/row-access policy error types and surface them via APIs - extend SQL logic tests to cover drop failures for active policies
1 parent b6afed8 commit 4b7ebab

File tree

8 files changed

+844
-14
lines changed

8 files changed

+844
-14
lines changed

‎src/meta/api/src/data_mask_api.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ use databend_common_meta_app::tenant::Tenant;
2121
use databend_common_meta_types::MetaError;
2222
use databend_common_meta_types::SeqV;
2323

24+
use crate::errors::MaskingPolicyError;
2425
use crate::kv_app_error::KVAppError;
2526

2627
#[async_trait::async_trait]
@@ -35,7 +36,7 @@ pub trait DatamaskApi: Send + Sync {
3536
async fn drop_data_mask(
3637
&self,
3738
name_ident: &DataMaskNameIdent,
38-
) -> Result<Option<(SeqV<DataMaskId>, SeqV<DatamaskMeta>)>, KVAppError>;
39+
) -> Result<Option<(SeqV<DataMaskId>, SeqV<DatamaskMeta>)>, MaskingPolicyError>;
3940

4041
async fn get_data_mask(
4142
&self,

‎src/meta/api/src/data_mask_api_impl.rs‎

Lines changed: 130 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,20 +19,28 @@ use databend_common_meta_app::data_mask::DataMaskId;
1919
use databend_common_meta_app::data_mask::DataMaskIdIdent;
2020
use databend_common_meta_app::data_mask::DataMaskNameIdent;
2121
use databend_common_meta_app::data_mask::DatamaskMeta;
22+
use databend_common_meta_app::data_mask::MaskPolicyIdTableId;
23+
use databend_common_meta_app::data_mask::MaskPolicyTableIdIdent;
2224
use databend_common_meta_app::data_mask::MaskPolicyTableIdListIdent;
2325
use databend_common_meta_app::data_mask::MaskpolicyTableIdList;
2426
use databend_common_meta_app::id_generator::IdGenerator;
2527
use databend_common_meta_app::schema::CreateOption;
28+
use databend_common_meta_app::schema::TableId;
29+
use databend_common_meta_app::schema::TableMeta;
2630
use databend_common_meta_app::tenant::Tenant;
2731
use databend_common_meta_app::KeyWithTenant;
2832
use databend_common_meta_kvapi::kvapi;
33+
use databend_common_meta_kvapi::kvapi::DirName;
2934
use databend_common_meta_types::MetaError;
3035
use databend_common_meta_types::SeqV;
3136
use databend_common_meta_types::TxnRequest;
3237
use fastrace::func_name;
3338
use log::debug;
3439

3540
use crate::data_mask_api::DatamaskApi;
41+
use crate::errors::MaskingPolicyError;
42+
use crate::errors::SecurityPolicyError;
43+
use crate::errors::SecurityPolicyKind;
3644
use crate::fetch_id;
3745
use crate::kv_app_error::KVAppError;
3846
use crate::kv_pb_api::KVPbApi;
@@ -135,12 +143,16 @@ impl<KV: kvapi::KVApi<Error = MetaError>> DatamaskApi for KV {
135143
async fn drop_data_mask(
136144
&self,
137145
name_ident: &DataMaskNameIdent,
138-
) -> Result<Option<(SeqV<DataMaskId>, SeqV<DatamaskMeta>)>, KVAppError> {
146+
) -> Result<Option<(SeqV<DataMaskId>, SeqV<DatamaskMeta>)>, MaskingPolicyError> {
139147
debug!(name_ident :? =(name_ident); "DatamaskApi: {}", func_name!());
140148

141149
let mut trials = txn_backoff(None, func_name!());
142150
loop {
143-
trials.next().unwrap()?.await;
151+
trials
152+
.next()
153+
.unwrap()
154+
.map_err(MaskingPolicyError::from)?
155+
.await;
144156
let mut txn = TxnRequest::default();
145157

146158
let res = self.get_id_and_value(name_ident).await?;
@@ -150,10 +162,33 @@ impl<KV: kvapi::KVApi<Error = MetaError>> DatamaskApi for KV {
150162
return Ok(None);
151163
};
152164

165+
let policy_id = *seq_id.data;
166+
let usage = collect_mask_policy_usage(self, name_ident.tenant(), policy_id).await?;
167+
if !usage.active_tables.is_empty() {
168+
let tenant = name_ident.tenant().tenant_name().to_string();
169+
let policy_name = name_ident.data_mask_name().to_string();
170+
let err = SecurityPolicyError::policy_in_use(
171+
tenant,
172+
SecurityPolicyKind::Masking,
173+
policy_name,
174+
);
175+
return Err(err.into());
176+
}
177+
153178
let id_ident = seq_id.data.into_t_ident(name_ident.tenant());
154179

155180
txn_delete_exact(&mut txn, name_ident, seq_id.seq);
156181
txn_delete_exact(&mut txn, &id_ident, seq_meta.seq);
182+
for binding in &usage.stale_bindings {
183+
txn_delete_exact(&mut txn, &binding.ident, binding.seq);
184+
}
185+
for update in &usage.table_updates {
186+
txn.condition
187+
.push(txn_cond_eq_seq(&update.table_id, update.seq));
188+
let op = txn_op_put_pb(&update.table_id, &update.meta, None)
189+
.map_err(|e| MaskingPolicyError::from(MetaError::from(e)))?;
190+
txn.if_then.push(op);
191+
}
157192
// TODO: Tentative retention for compatibility MaskPolicyTableIdListIdent related logic. It can be directly deleted later
158193
clear_table_column_mask_policy(self, name_ident, &mut txn).await?;
159194
let (succ, _responses) = send_txn(self, txn).await?;
@@ -195,7 +230,7 @@ async fn clear_table_column_mask_policy(
195230
kv_api: &(impl kvapi::KVApi<Error = MetaError> + ?Sized),
196231
name_ident: &DataMaskNameIdent,
197232
txn: &mut TxnRequest,
198-
) -> Result<(), MetaError> {
233+
) -> Result<(), MaskingPolicyError> {
199234
let id_list_key = MaskPolicyTableIdListIdent::new_from(name_ident.clone());
200235

201236
let seq_id_list = kv_api.get_pb(&id_list_key).await?;
@@ -207,3 +242,95 @@ async fn clear_table_column_mask_policy(
207242
txn_delete_exact(txn, &id_list_key, seq_id_list.seq);
208243
Ok(())
209244
}
245+
246+
#[derive(Default)]
247+
struct MaskPolicyUsage {
248+
active_tables: Vec<u64>,
249+
stale_bindings: Vec<BindingEntry>,
250+
table_updates: Vec<TableMetaUpdate>,
251+
}
252+
253+
struct BindingEntry {
254+
ident: MaskPolicyTableIdIdent,
255+
seq: u64,
256+
}
257+
258+
struct TableMetaUpdate {
259+
table_id: TableId,
260+
seq: u64,
261+
meta: TableMeta,
262+
}
263+
264+
fn strip_mask_policy_from_table_meta(table_meta: &mut TableMeta, policy_id: u64) -> bool {
265+
let mut removed = false;
266+
table_meta
267+
.column_mask_policy_columns_ids
268+
.retain(|_, policy| {
269+
let keep = policy.policy_id != policy_id;
270+
if !keep {
271+
removed = true;
272+
}
273+
keep
274+
});
275+
276+
if removed {
277+
table_meta.column_mask_policy = None;
278+
}
279+
280+
removed
281+
}
282+
283+
async fn collect_mask_policy_usage(
284+
kv_api: &(impl kvapi::KVApi<Error = MetaError> + ?Sized),
285+
tenant: &Tenant,
286+
policy_id: u64,
287+
) -> Result<MaskPolicyUsage, MaskingPolicyError> {
288+
let binding_prefix = DirName::new(MaskPolicyTableIdIdent::new_generic(
289+
tenant.clone(),
290+
MaskPolicyIdTableId {
291+
policy_id,
292+
table_id: 0,
293+
},
294+
));
295+
let bindings = kv_api
296+
.list_pb_vec(&binding_prefix)
297+
.await
298+
.map_err(MaskingPolicyError::from)?;
299+
300+
let mut usage = MaskPolicyUsage::default();
301+
for (binding_ident, seqv) in bindings {
302+
let table_id = binding_ident.name().table_id;
303+
let table_key = TableId::new(table_id);
304+
match kv_api
305+
.get_pb(&table_key)
306+
.await
307+
.map_err(MaskingPolicyError::from)?
308+
{
309+
Some(mut table_meta_seqv) => {
310+
if table_meta_seqv.data.drop_on.is_none() {
311+
usage.active_tables.push(table_id);
312+
} else {
313+
if strip_mask_policy_from_table_meta(&mut table_meta_seqv.data, policy_id) {
314+
usage.table_updates.push(TableMetaUpdate {
315+
table_id: table_key,
316+
seq: table_meta_seqv.seq,
317+
meta: table_meta_seqv.data.clone(),
318+
});
319+
}
320+
usage.stale_bindings.push(BindingEntry {
321+
ident: binding_ident,
322+
seq: seqv.seq,
323+
});
324+
}
325+
}
326+
None => {
327+
usage.stale_bindings.push(BindingEntry {
328+
ident: binding_ident,
329+
seq: seqv.seq,
330+
});
331+
}
332+
}
333+
}
334+
335+
Ok(usage)
336+
}

‎src/meta/api/src/errors.rs‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,11 @@
1313
// limitations under the License.
1414

1515
use databend_common_exception::ErrorCode;
16+
use databend_common_meta_app::app_error::AppErrorMessage;
17+
use databend_common_meta_app::app_error::TxnRetryMaxTimes;
1618
use databend_common_meta_app::principal::AutoIncrementKey;
19+
use databend_common_meta_types::InvalidArgument;
20+
use databend_common_meta_types::MetaError;
1721

1822
/// Table logic error, unrelated to the backend service providing Table management, or dependent component.
1923
#[derive(Clone, Debug, thiserror::Error)]
@@ -39,6 +43,104 @@ impl From<TableError> for ErrorCode {
3943
}
4044
}
4145

46+
#[derive(Clone, Debug, PartialEq, Eq)]
47+
pub enum SecurityPolicyKind {
48+
Masking,
49+
RowAccess,
50+
}
51+
52+
impl std::fmt::Display for SecurityPolicyKind {
53+
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
54+
match self {
55+
SecurityPolicyKind::Masking => write!(f, "MASKING POLICY"),
56+
SecurityPolicyKind::RowAccess => write!(f, "ROW ACCESS POLICY"),
57+
}
58+
}
59+
}
60+
61+
/// Security policy related business error.
62+
#[derive(Clone, Debug, thiserror::Error)]
63+
pub enum SecurityPolicyError {
64+
#[error(
65+
"{policy_kind} `{policy_name}` is still in use. Unset it from all tables before dropping."
66+
)]
67+
PolicyInUse {
68+
tenant: String,
69+
policy_kind: SecurityPolicyKind,
70+
policy_name: String,
71+
},
72+
}
73+
74+
impl SecurityPolicyError {
75+
pub fn policy_in_use(
76+
tenant: impl Into<String>,
77+
policy_kind: SecurityPolicyKind,
78+
policy_name: impl Into<String>,
79+
) -> Self {
80+
Self::PolicyInUse {
81+
tenant: tenant.into(),
82+
policy_kind,
83+
policy_name: policy_name.into(),
84+
}
85+
}
86+
}
87+
88+
impl From<SecurityPolicyError> for ErrorCode {
89+
fn from(value: SecurityPolicyError) -> Self {
90+
let s = value.to_string();
91+
match value {
92+
SecurityPolicyError::PolicyInUse { .. } => ErrorCode::ConstraintError(s),
93+
}
94+
}
95+
}
96+
97+
impl From<SecurityPolicyError> for InvalidArgument {
98+
fn from(value: SecurityPolicyError) -> Self {
99+
let msg = value.to_string();
100+
InvalidArgument::new(value, msg)
101+
}
102+
}
103+
104+
#[derive(Clone, Debug, thiserror::Error)]
105+
pub enum MaskingPolicyError {
106+
#[error(transparent)]
107+
TxnRetry(#[from] TxnRetryMaxTimes),
108+
#[error(transparent)]
109+
Meta(#[from] MetaError),
110+
#[error(transparent)]
111+
PolicyInUse(#[from] SecurityPolicyError),
112+
}
113+
114+
impl From<MaskingPolicyError> for ErrorCode {
115+
fn from(value: MaskingPolicyError) -> Self {
116+
match value {
117+
MaskingPolicyError::TxnRetry(err) => ErrorCode::TxnRetryMaxTimes(err.message()),
118+
MaskingPolicyError::Meta(err) => ErrorCode::from(err),
119+
MaskingPolicyError::PolicyInUse(err) => err.into(),
120+
}
121+
}
122+
}
123+
124+
#[derive(Clone, Debug, thiserror::Error)]
125+
pub enum RowAccessPolicyError {
126+
#[error(transparent)]
127+
TxnRetry(#[from] TxnRetryMaxTimes),
128+
#[error(transparent)]
129+
Meta(#[from] MetaError),
130+
#[error(transparent)]
131+
PolicyInUse(#[from] SecurityPolicyError),
132+
}
133+
134+
impl From<RowAccessPolicyError> for ErrorCode {
135+
fn from(value: RowAccessPolicyError) -> Self {
136+
match value {
137+
RowAccessPolicyError::TxnRetry(err) => ErrorCode::TxnRetryMaxTimes(err.message()),
138+
RowAccessPolicyError::Meta(err) => ErrorCode::from(err),
139+
RowAccessPolicyError::PolicyInUse(err) => err.into(),
140+
}
141+
}
142+
}
143+
42144
#[derive(thiserror::Error, Debug, Clone, PartialEq, Eq)]
43145
pub enum AutoIncrementError {
44146
#[error("OutOfAutoIncrementRange: `{key}` while `{context}`")]

‎src/meta/api/src/lib.rs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,10 @@ pub use error_util::db_has_to_not_exist;
7878
pub use error_util::db_id_has_to_exist;
7979
pub use error_util::table_has_to_not_exist;
8080
pub use error_util::unknown_database_error;
81+
pub use errors::MaskingPolicyError;
82+
pub use errors::RowAccessPolicyError;
83+
pub use errors::SecurityPolicyError;
84+
pub use errors::SecurityPolicyKind;
8185
pub use garbage_collection_api::GarbageCollectionApi;
8286
pub use index_api::IndexApi;
8387
// Re-export from new kv_fetch_util module for backward compatibility

‎src/meta/api/src/row_access_policy_api.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ use databend_common_meta_app::tenant_key::errors::ExistError;
2323
use databend_common_meta_types::MetaError;
2424
use databend_common_meta_types::SeqV;
2525

26+
use crate::errors::RowAccessPolicyError;
2627
use crate::meta_txn_error::MetaTxnError;
2728

2829
#[async_trait::async_trait]
@@ -40,7 +41,7 @@ pub trait RowAccessPolicyApi: Send + Sync {
4041
async fn drop_row_access(
4142
&self,
4243
name_ident: &RowAccessPolicyNameIdent,
43-
) -> Result<Option<(SeqV<RowAccessPolicyId>, SeqV<RowAccessPolicyMeta>)>, MetaTxnError>;
44+
) -> Result<Option<(SeqV<RowAccessPolicyId>, SeqV<RowAccessPolicyMeta>)>, RowAccessPolicyError>;
4445

4546
async fn get_row_access(
4647
&self,

0 commit comments

Comments
 (0)