fix: harden permission checks and process visibility (#8852)

Signed-off-by: shuiyisong <xixing.sys@gmail.com>
(cherry picked from commit 154f90b365)

src/auth/src/lib.rs

@@ -31,8 +31,8 @@ pub use common::{
 };
 pub use permission::{
     ALL_ACTIONS, AccessMode, DASHBOARD_DELETE, DASHBOARD_QUERY, DASHBOARD_SAVE,
-    DefaultPermissionChecker, INFLUXDB_WRITE, JAEGER_QUERY, LOG_QUERY, LOG_WRITE, OPENTSDB_WRITE,
-    OTLP_WRITE, PIPELINE_DELETE, PIPELINE_INSERT, PIPELINE_QUERY, PROM_STORE_READ,
+    DefaultPermissionChecker, ICEBERG_READ, INFLUXDB_WRITE, JAEGER_QUERY, LOG_QUERY, LOG_WRITE,
+    OPENTSDB_WRITE, OTLP_WRITE, PIPELINE_DELETE, PIPELINE_INSERT, PIPELINE_QUERY, PROM_STORE_READ,
     PROM_STORE_WRITE, PROMQL_QUERY, PermissionAction, PermissionChecker, PermissionReq,
     PermissionResp, PermissionTableTarget, PermissionTableTargets,
 };

src/auth/src/permission.rs

@@ -139,6 +139,7 @@ pub const PROMQL_QUERY: PermissionAction = PermissionAction::read("promql.query"
 pub const LOG_QUERY: PermissionAction = PermissionAction::read("log.query");
 pub const OPENTSDB_WRITE: PermissionAction = PermissionAction::write("opentsdb.write");
 pub const INFLUXDB_WRITE: PermissionAction = PermissionAction::write("influxdb.write");
+pub const ICEBERG_READ: PermissionAction = PermissionAction::read("iceberg.read");
 pub const PROM_STORE_WRITE: PermissionAction = PermissionAction::write("prom_store.write");
 pub const PROM_STORE_READ: PermissionAction = PermissionAction::read("prom_store.read");
 pub const OTLP_WRITE: PermissionAction = PermissionAction::write("otlp.write");
@@ -160,6 +161,7 @@ pub const ALL_ACTIONS: &[PermissionAction] = &[
     LOG_QUERY,
     OPENTSDB_WRITE,
     INFLUXDB_WRITE,
+    ICEBERG_READ,
     PROM_STORE_WRITE,
     PROM_STORE_READ,
     OTLP_WRITE,

src/auth/src/user_info.rs

@@ -21,6 +21,10 @@ use crate::UserInfoRef;
 pub trait UserInfo: Debug + Sync + Send {
     fn as_any(&self) -> &dyn Any;
     fn username(&self) -> &str;
+
+    fn is_admin(&self) -> bool {
+        false
+    }
 }
 
 /// The user permission mode
@@ -38,13 +42,14 @@ impl PermissionMode {
     /// - "rw", "readwrite", "read_write" => ReadWrite
     /// - "ro", "readonly", "read_only" => ReadOnly
     /// - "wo", "writeonly", "write_only" => WriteOnly
-    ///     Returns None if the input string is not a valid permission mode.
-    pub fn from_str(s: &str) -> Self {
+    ///
+    /// Returns `None` if the input string is not a valid permission mode.
+    pub fn from_str(s: &str) -> Option<Self> {
         match s.to_lowercase().as_str() {
-            "readwrite" | "read_write" | "rw" => PermissionMode::ReadWrite,
-            "readonly" | "read_only" | "ro" => PermissionMode::ReadOnly,
-            "writeonly" | "write_only" | "wo" => PermissionMode::WriteOnly,
-            _ => PermissionMode::ReadWrite,
+            "readwrite" | "read_write" | "rw" => Some(PermissionMode::ReadWrite),
+            "readonly" | "read_only" | "ro" => Some(PermissionMode::ReadOnly),
+            "writeonly" | "write_only" | "wo" => Some(PermissionMode::WriteOnly),
+            _ => None,
         }
     }
 
@@ -123,58 +128,72 @@ mod tests {
         // Test ReadWrite variants
         assert_eq!(
             PermissionMode::from_str("readwrite"),
-            PermissionMode::ReadWrite
+            Some(PermissionMode::ReadWrite)
         );
         assert_eq!(
             PermissionMode::from_str("read_write"),
-            PermissionMode::ReadWrite
+            Some(PermissionMode::ReadWrite)
+        );
+        assert_eq!(
+            PermissionMode::from_str("rw"),
+            Some(PermissionMode::ReadWrite)
         );
-        assert_eq!(PermissionMode::from_str("rw"), PermissionMode::ReadWrite);
         assert_eq!(
             PermissionMode::from_str("ReadWrite"),
-            PermissionMode::ReadWrite
+            Some(PermissionMode::ReadWrite)
+        );
+        assert_eq!(
+            PermissionMode::from_str("RW"),
+            Some(PermissionMode::ReadWrite)
         );
-        assert_eq!(PermissionMode::from_str("RW"), PermissionMode::ReadWrite);
 
         // Test ReadOnly variants
         assert_eq!(
             PermissionMode::from_str("readonly"),
-            PermissionMode::ReadOnly
+            Some(PermissionMode::ReadOnly)
         );
         assert_eq!(
             PermissionMode::from_str("read_only"),
-            PermissionMode::ReadOnly
+            Some(PermissionMode::ReadOnly)
+        );
+        assert_eq!(
+            PermissionMode::from_str("ro"),
+            Some(PermissionMode::ReadOnly)
         );
-        assert_eq!(PermissionMode::from_str("ro"), PermissionMode::ReadOnly);
         assert_eq!(
             PermissionMode::from_str("ReadOnly"),
-            PermissionMode::ReadOnly
+            Some(PermissionMode::ReadOnly)
+        );
+        assert_eq!(
+            PermissionMode::from_str("RO"),
+            Some(PermissionMode::ReadOnly)
         );
-        assert_eq!(PermissionMode::from_str("RO"), PermissionMode::ReadOnly);
 
         // Test WriteOnly variants
         assert_eq!(
             PermissionMode::from_str("writeonly"),
-            PermissionMode::WriteOnly
+            Some(PermissionMode::WriteOnly)
         );
         assert_eq!(
             PermissionMode::from_str("write_only"),
-            PermissionMode::WriteOnly
+            Some(PermissionMode::WriteOnly)
+        );
+        assert_eq!(
+            PermissionMode::from_str("wo"),
+            Some(PermissionMode::WriteOnly)
         );
-        assert_eq!(PermissionMode::from_str("wo"), PermissionMode::WriteOnly);
         assert_eq!(
             PermissionMode::from_str("WriteOnly"),
-            PermissionMode::WriteOnly
+            Some(PermissionMode::WriteOnly)
         );
-        assert_eq!(PermissionMode::from_str("WO"), PermissionMode::WriteOnly);
-
-        // Test invalid inputs default to ReadWrite
         assert_eq!(
-            PermissionMode::from_str("invalid"),
-            PermissionMode::ReadWrite
+            PermissionMode::from_str("WO"),
+            Some(PermissionMode::WriteOnly)
         );
-        assert_eq!(PermissionMode::from_str(""), PermissionMode::ReadWrite);
-        assert_eq!(PermissionMode::from_str("xyz"), PermissionMode::ReadWrite);
+
+        for invalid in ["readonyl", "", "xyz"] {
+            assert_eq!(PermissionMode::from_str(invalid), None);
+        }
     }
 
     #[test]
@@ -200,7 +219,7 @@ mod tests {
         for mode in modes {
             let str_repr = mode.as_str();
             let parsed = PermissionMode::from_str(str_repr);
-            assert_eq!(mode, parsed);
+            assert_eq!(Some(mode), parsed);
         }
     }
 

src/auth/src/user_provider.rs

@@ -371,7 +371,7 @@ pub(crate) fn parse_credential_line(
 
     let (username_part, password) = (parts[0], parts[1]);
     let (username, permission_mode) = if let Some((user, perm)) = username_part.split_once(':') {
-        (user, PermissionMode::from_str(perm))
+        (user, PermissionMode::from_str(perm)?)
     } else {
         (username_part, PermissionMode::default())
     };
@@ -674,6 +674,14 @@ mod tests {
         let result = parse_credential_line("user=pass=word");
         assert_eq!(result, None);
 
+        for line in [
+            "user:readonyl=password",
+            "user:=password",
+            "user:arbitrary=password",
+        ] {
+            assert_eq!(parse_credential_line(line), None);
+        }
+
         // Empty password
         let result = parse_credential_line("user=");
         assert_eq!(

src/auth/src/user_provider/static_user_provider.rs

@@ -144,6 +144,8 @@ pub mod test {
         let provider = StaticUserProvider::new("cmd:root=123456,admin=654321").unwrap();
         test_authenticate(&provider, "root", "123456").await;
         test_authenticate(&provider, "admin", "654321").await;
+
+        assert!(StaticUserProvider::new("cmd:user:readonyl=password").is_err());
     }
 
     #[tokio::test]
@@ -168,6 +170,7 @@ pub mod test {
             assert!(
                 lw.write_all(
                     b"root=123456
+invalid:readonyl=password
 admin=654321",
                 )
                 .is_ok()
@@ -179,5 +182,6 @@ admin=654321",
         let provider = StaticUserProvider::new(param.as_str()).unwrap();
         test_authenticate(&provider, "root", "123456").await;
         test_authenticate(&provider, "admin", "654321").await;
+        test_authenticate_fails(&provider, "invalid", "password").await;
     }
 }

src/frontend/src/instance.rs

@@ -1666,6 +1666,7 @@ fn should_track_plan_process(stmt: Option<&Statement>, plan: &LogicalPlan) -> bo
 
 #[cfg(test)]
 mod tests {
+    use std::any::Any;
     use std::collections::HashMap;
     use std::future::Future;
     use std::pin::Pin;
@@ -1677,10 +1678,12 @@ mod tests {
     use api::prom_store::remote::{
         Label, LabelMatcher, Query as RemoteQuery, ReadRequest, ReadResponse, Sample,
     };
+    use api::v1::greptime_request::Request;
     use api::v1::meta::{ProcedureDetailResponse, ReconcileRequest, ReconcileResponse};
+    use api::v1::query_request::Query;
     use auth::{
         DASHBOARD_DELETE, DASHBOARD_QUERY, DASHBOARD_SAVE, JAEGER_QUERY, PIPELINE_DELETE,
-        PIPELINE_INSERT, PIPELINE_QUERY, PermissionAction, PermissionResp, UserInfoRef,
+        PIPELINE_INSERT, PIPELINE_QUERY, PermissionAction, PermissionResp, UserInfo, UserInfoRef,
     };
     use catalog::process_manager::{ProcessManager, QueryStatement, SlowQueryTimer};
     use common_base::Plugins;
@@ -1943,6 +1946,23 @@ mod tests {
         )
     }
 
+    #[derive(Debug)]
+    struct AdminUserInfo;
+
+    impl UserInfo for AdminUserInfo {
+        fn as_any(&self) -> &dyn Any {
+            self
+        }
+
+        fn username(&self) -> &str {
+            "admin"
+        }
+
+        fn is_admin(&self) -> bool {
+            true
+        }
+    }
+
     struct RejectUnresolvedPermissionChecker;
 
     impl PermissionChecker for RejectUnresolvedPermissionChecker {
@@ -3103,6 +3123,45 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_write_only_grpc_sql_is_checked_after_parsing() -> TestResult<()> {
+        let plugins = Plugins::new();
+        plugins.insert::<PermissionCheckerRef>(Arc::new(WriteOnlyPermissionChecker));
+        let instance = test_instance_with_plugins(
+            test_table(1024, "source")?,
+            test_table(1025, "target")?,
+            plugins,
+        )
+        .await?;
+
+        let insert = Request::Query(api::v1::QueryRequest {
+            query: Some(Query::Sql(
+                "INSERT INTO target SELECT * FROM source".to_string(),
+            )),
+        });
+        servers::query_handler::grpc::GrpcQueryHandler::do_query(
+            &instance,
+            insert,
+            QueryContext::arc(),
+        )
+        .await
+        .unwrap();
+
+        let select = Request::Query(api::v1::QueryRequest {
+            query: Some(Query::Sql("SELECT * FROM source".to_string())),
+        });
+        assert_permission_denied(
+            servers::query_handler::grpc::GrpcQueryHandler::do_query(
+                &instance,
+                select,
+                QueryContext::arc(),
+            )
+            .await,
+        );
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_target_independent_checker_skips_target_resolution() -> TestResult<()> {
         let physical_table = "physical_metric";
@@ -3439,6 +3498,57 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_show_processlist_catalog_scope() -> TestResult<()> {
+        let instance =
+            test_instance_with_tables(test_table(1024, "source")?, test_table(1025, "target")?)
+                .await?;
+        let _current_catalog = instance.process_manager().register_query(
+            "greptime".to_string(),
+            vec!["public".to_string()],
+            "current_catalog_query".to_string(),
+            String::new(),
+            None,
+            None,
+        );
+        let _other_catalog = instance.process_manager().register_query(
+            "other".to_string(),
+            vec!["public".to_string()],
+            "other_catalog_query".to_string(),
+            String::new(),
+            None,
+            None,
+        );
+
+        for sql in ["SHOW PROCESSLIST", "SHOW FULL PROCESSLIST"] {
+            let output = execute_one_sql(&instance, sql, test_query_ctx(43)).await?;
+            let process_list = output.data.pretty_print().await;
+            assert!(
+                process_list.contains("current_catalog_query"),
+                "{process_list}"
+            );
+            assert!(
+                !process_list.contains("other_catalog_query"),
+                "{process_list}"
+            );
+
+            let admin_ctx = test_query_ctx(44);
+            admin_ctx.set_current_user(Arc::new(AdminUserInfo));
+            let output = execute_one_sql(&instance, sql, admin_ctx).await?;
+            let process_list = output.data.pretty_print().await;
+            assert!(
+                process_list.contains("current_catalog_query"),
+                "{process_list}"
+            );
+            assert!(
+                process_list.contains("other_catalog_query"),
+                "{process_list}"
+            );
+        }
+
+        Ok(())
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn test_kill_query_cancels_insert_select() -> TestResult<()> {
         assert_kill_cancels_insert_select("KILL QUERY 4242").await

src/frontend/src/instance/grpc.rs

@@ -70,15 +70,21 @@ impl GrpcQueryHandler for Instance {
             let interceptor = interceptor_ref.as_ref();
             interceptor.pre_execute(&request, ctx.clone())?;
 
-            self.plugins
-                .get::<PermissionCheckerRef>()
-                .as_ref()
-                .check_permission_with_context(
-                    ctx.current_user(),
-                    PermissionReq::GrpcRequest(&request),
-                    Some(&ctx.current_schema()),
-                )
-                .context(PermissionSnafu)?;
+            if !matches!(
+                &request,
+                Request::Query(query_request)
+                    if matches!(&query_request.query, Some(Query::Sql(_)))
+            ) {
+                self.plugins
+                    .get::<PermissionCheckerRef>()
+                    .as_ref()
+                    .check_permission_with_context(
+                        ctx.current_user(),
+                        PermissionReq::GrpcRequest(&request),
+                        Some(&ctx.current_schema()),
+                    )
+                    .context(PermissionSnafu)?;
+            }
 
             let output = match request {
                 Request::Inserts(requests) => self.handle_inserts(requests, ctx.clone()).await?,

src/query/src/sql.rs

@@ -1360,7 +1360,11 @@ pub async fn show_processlist(
         ]
     };
 
-    let filters = vec![];
+    let filters = if query_ctx.current_user().is_admin() {
+        vec![]
+    } else {
+        vec![col(process_list::CATALOG).eq(lit(query_ctx.current_catalog()))]
+    };
     let like_field = None;
     let sort = vec![col("id").sort(true, true)];
     query_from_information_schema_table(