fix(query): prevent modification of columns with security policies (#18896)

* fix(query): prevent modification of columns with security policies

This patch addresses two critical security policy enforcement issues:

Bug-1: Prevent dropping or modifying columns with attached policies
  Columns that have masking policies or serve as row access policy
  parameters must not be dropped or modified, as this would break
  security policy enforcement. Added validation in:
  - interpreter_table_drop_column: Blocks DROP COLUMN operations
  - interpreter_table_modify_column: Blocks data type changes

Bug-2: Prevent duplicate policy assignment on columns
  A column cannot have multiple security policies attached
  simultaneously. Added checks in:
  - interpreter_table_modify_column: Validates before SET MASKING POLICY
  - interpreter_table_row_access_add: Validates before SET ROW ACCESS POLICY

Implementation:
  - Created check_column_has_policy() utility in interpreters/util.rs
  - Validates against both masking and row access policies
  - Returns descriptive error messages to prevent security violations

This ensures policy integrity and prevents accidental security
configuration corruption during table schema alterations.

* optimize

Author: TCeason

src/meta/app/src/schema/table/ops.rs

@@ -18,6 +18,7 @@ use std::sync::Arc;
 
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
+use databend_common_expression::ColumnId;
 use databend_common_expression::FieldIndex;
 use databend_common_expression::TableField;
 
@@ -83,6 +84,16 @@ impl TableMeta {
         };
         Ok(())
     }
+
+    /// Check if a column reference a security policy.
+    pub fn is_column_reference_policy(&self, column_id: &ColumnId) -> bool {
+        self.column_mask_policy_columns_ids.contains_key(column_id)
+            || self
+                .row_access_policy_columns_ids
+                .as_ref()
+                .map(|policy| policy.columns_ids.contains(column_id))
+                .unwrap_or(false)
+    }
 }
 
 #[cfg(test)]

src/query/service/src/interpreters/interpreter_table_drop_column.rs

@@ -85,13 +85,12 @@ impl Interpreter for DropTableColumnInterpreter {
 
         let table_schema = table_info.schema();
         let field = table_schema.field_with_name(self.plan.column.as_str())?;
-        if let Some(p) = &table_info.meta.row_access_policy_columns_ids {
-            if p.columns_ids.contains(&field.column_id) {
-                return Err(ErrorCode::AlterTableError(format!(
-                    "Cannot drop column '{}' which is associated with a Row access policy",
-                    self.plan.column.as_str()
-                )));
-            }
+
+        if table_info.meta.is_column_reference_policy(&field.column_id) {
+            return Err(ErrorCode::AlterTableError(format!(
+                "Cannot drop column '{}' which is associated with a security policy",
+                self.plan.column.as_str()
+            )));
         }
         if field.computed_expr().is_none() {
             let mut schema: DataSchema = table_info.schema().into();

src/query/service/src/interpreters/interpreter_table_modify_column.rs

@@ -146,6 +146,16 @@ impl ModifyTableColumnInterpreter {
                 ErrorCode::UnknownColumn(format!("Cannot find column {}", column))
             })?;
 
+            if table_info
+                .meta
+                .is_column_reference_policy(&data_field.column_id)
+            {
+                return Err(ErrorCode::AlterTableError(format!(
+                    "Column '{}' is already attached to a security policy. A column cannot be attached to multiple security policies",
+                    data_field.name
+                )));
+            }
+
             let column_type = data_field.data_type();
             if policy_data_type != column_type.remove_nullable() {
                 return Err(ErrorCode::UnmatchColumnDataType(format!(
@@ -241,6 +251,16 @@ impl ModifyTableColumnInterpreter {
         let mut modify_comment = false;
         for (field, comment) in field_and_comments {
             if let Some((i, old_field)) = schema.column_with_name(&field.name) {
+                if table_info
+                    .meta
+                    .is_column_reference_policy(&old_field.column_id)
+                {
+                    return Err(ErrorCode::AlterTableError(format!(
+                        "Cannot modify column '{}' which is associated with a security policy",
+                        old_field.name
+                    )));
+                }
+
                 if old_field.data_type != field.data_type {
                     // If the column is defined in bloom index columns,
                     // check whether the data type is supported for bloom index.

src/query/service/src/interpreters/interpreter_table_row_access_add.rs

@@ -120,6 +120,16 @@ impl Interpreter for AddTableRowAccessPolicyInterpreter {
 
         for (column, policy_data_type) in columns.iter().zip(policy_data_types.into_iter()) {
             if let Some((_, data_field)) = schema.column_with_name(column) {
+                if table
+                    .get_table_info()
+                    .meta
+                    .is_column_reference_policy(&data_field.column_id)
+                {
+                    return Err(ErrorCode::AlterTableError(format!(
+                        "Column '{}' is already attached to a security policy. A column cannot be attached to multiple security policies",
+                        data_field.name
+                    )));
+                }
                 let column_type = data_field.data_type();
                 if policy_data_type != column_type.remove_nullable() {
                     return Err(ErrorCode::UnmatchColumnDataType(format!(

tests/sqllogictests/suites/ee/05_ee_ddl/05_0004_ddl_security_policy.test

@@ -136,7 +136,10 @@ select id from t_1;
 ----
 2
 
-statement error
+statement error 1132
+alter table t_1 modify column cc string not null;
+
+statement error 1132
 alter table t_1 drop column cc;
 
 statement ok
@@ -301,6 +304,12 @@ CREATE MASKING POLICY maskc AS (val int) RETURNS int -> CASE WHEN current_role()
 statement ok
 alter table data_mask_test modify column b set masking policy maskb;
 
+statement error 1132
+alter table data_mask_test modify column b string not null;
+
+statement error 1132
+alter table data_mask_test drop column b;
+
 query ITT
 select * from data_mask_test;
 ----
@@ -309,6 +318,9 @@ select * from data_mask_test;
 statement ok
 alter table data_mask_test modify column a set masking policy maska;
 
+statement error 1132
+alter table data_mask_test modify column a set masking policy maskc;
+
 query ITT
 select * from data_mask_test;
 ----
@@ -394,6 +406,10 @@ COMMENT = 'Mask SSN conditionally based on role';
 statement ok
 alter table employees modify column ssn set masking policy mask_ssn_conditional using (ssn, role);
 
+## Test 2: Apply policy with USING clause - Column 'ssn' is already attached to a security policy.
+statement error 1132
+alter table employees modify column role set masking policy mask_ssn_conditional using (role, ssn);
+
 ## Query should mask SSN for all rows since current_role() won't match
 query ITTIT
 select id, name, ssn, salary, department from employees order by id;
@@ -464,7 +480,7 @@ drop masking policy if exists mask_int_policy;
 statement ok
 CREATE MASKING POLICY mask_int_policy AS (val STRING, num INT) RETURNS STRING -> val;
 
-## Should fail - department is STRING, not INT
+## Should fail - Column 'department' is already attached to a security policy. A column cannot be attached to multiple security policies
 statement error 1114
 alter table employees modify column name set masking policy mask_int_policy using (name, department);
 
@@ -810,6 +826,9 @@ SELECT array_transform(arr, x -> x + sensitive) FROM t_test;
 statement ok
 CREATE MASKING POLICY mask_conditional AS (val INT, threshold INT) RETURNS INT -> CASE WHEN val > threshold THEN val ELSE -999 END;
 
+statement ok
+ALTER TABLE t_test MODIFY COLUMN sensitive UNSET MASKING POLICY
+
 statement ok
 ALTER TABLE t_test MODIFY COLUMN sensitive SET MASKING POLICY mask_conditional USING (sensitive, public);