From 2927fa339ec677fb632a8940bcd0d43eea78881a Mon Sep 17 00:00:00 2001 From: shuiyisong <113876041+shuiyisong@users.noreply.github.com> Date: Wed, 12 Aug 2026 03:12:04 +0000 Subject: [PATCH] fix: harden permission checks and process visibility (#8852) Signed-off-by: shuiyisong (cherry picked from commit 154f90b3652bf9b1b3e7eccc5f277f856a2f3079) --- src/auth/src/lib.rs | 4 +- src/auth/src/permission.rs | 2 + src/auth/src/user_info.rs | 75 +++++++----- src/auth/src/user_provider.rs | 10 +- .../src/user_provider/static_user_provider.rs | 4 + src/frontend/src/instance.rs | 112 +++++++++++++++++- src/frontend/src/instance/grpc.rs | 24 ++-- src/query/src/sql.rs | 6 +- 8 files changed, 195 insertions(+), 42 deletions(-) diff --git a/src/auth/src/lib.rs b/src/auth/src/lib.rs index 44382737ee..5e76fe956f 100644 --- a/src/auth/src/lib.rs +++ b/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, }; diff --git a/src/auth/src/permission.rs b/src/auth/src/permission.rs index 60a8922573..61c25a4911 100644 --- a/src/auth/src/permission.rs +++ b/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, diff --git a/src/auth/src/user_info.rs b/src/auth/src/user_info.rs index 6ce735c4e2..ab7c7f32f4 100644 --- a/src/auth/src/user_info.rs +++ b/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 { 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); } } diff --git a/src/auth/src/user_provider.rs b/src/auth/src/user_provider.rs index 344e666901..b503c669ef 100644 --- a/src/auth/src/user_provider.rs +++ b/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!( diff --git a/src/auth/src/user_provider/static_user_provider.rs b/src/auth/src/user_provider/static_user_provider.rs index 4ea7efb434..b55caf9faf 100644 --- a/src/auth/src/user_provider/static_user_provider.rs +++ b/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; } } diff --git a/src/frontend/src/instance.rs b/src/frontend/src/instance.rs index e8e39bebdf..8a5f6c84f3 100644 --- a/src/frontend/src/instance.rs +++ b/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::(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 diff --git a/src/frontend/src/instance/grpc.rs b/src/frontend/src/instance/grpc.rs index 3caac10b84..0327809239 100644 --- a/src/frontend/src/instance/grpc.rs +++ b/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::() - .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::() + .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?, diff --git a/src/query/src/sql.rs b/src/query/src/sql.rs index fc755deb8a..b2d979c666 100644 --- a/src/query/src/sql.rs +++ b/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(