fix(frontend): harden SHOW object visibility

e2e_test/ddl/show_object_privilege.slt

@@ -0,0 +1,86 @@
+# Verify SHOW object-list commands don't expose objects that are only schema-visible.
+# Connector-backed SHOW cases live in their docker-backed source/sink/connection suites.
+
+statement ok
+CREATE USER show_acl_user WITH PASSWORD 'password';
+
+statement ok
+CREATE SCHEMA show_acl;
+
+statement ok
+GRANT USAGE ON SCHEMA show_acl TO show_acl_user;
+
+statement ok
+CREATE TABLE show_acl.hidden_table (v1 int, v2 varchar);
+
+statement ok
+CREATE INDEX hidden_idx ON show_acl.hidden_table (v1);
+
+statement ok
+CREATE MATERIALIZED VIEW show_acl.hidden_mv AS SELECT v1 FROM show_acl.hidden_table;
+
+statement ok
+CREATE VIEW show_acl.hidden_view AS SELECT v1 FROM show_acl.hidden_table;
+
+statement ok
+CREATE SECRET show_acl.hidden_secret WITH (backend = 'meta') AS 'http://example.invalid';
+
+# Schema USAGE alone should not reveal object names.
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW SECRETS FROM show_acl;"); ! grep -q "hidden_secret" <<<"$out"'
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW SECRETS FROM show_acl LIKE '\''hidden_secret'\'';"); ! grep -q "hidden_secret" <<<"$out"'
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW TABLES FROM show_acl;"); ! grep -q "hidden_table" <<<"$out"'
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW VIEWS FROM show_acl;"); ! grep -q "hidden_view" <<<"$out"'
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW MATERIALIZED VIEWS FROM show_acl;"); ! grep -q "hidden_mv" <<<"$out"'
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW INTERNAL TABLES FROM show_acl;"); test -z "$out"'
+
+# Known-name probes should not return metadata for hidden tables.
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW COLUMNS FROM show_acl.hidden_table;" 2>&1 || true); ! grep -q "v1.*integer" <<<"$out"'
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW INDEXES FROM show_acl.hidden_table;" 2>&1 || true); ! grep -q "hidden_idx" <<<"$out"'
+
+# Once the object itself becomes visible, SHOW should include it.
+statement ok
+GRANT USAGE ON SECRET show_acl.hidden_secret TO show_acl_user;
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW SECRETS FROM show_acl;"); grep -q "show_acl.hidden_secret" <<<"$out"'
+
+statement ok
+GRANT SELECT ON TABLE show_acl.hidden_table TO show_acl_user;
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW TABLES FROM show_acl;"); grep -q "show_acl.hidden_table" <<<"$out"'
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW COLUMNS FROM show_acl.hidden_table;"); grep -q "v1.*integer" <<<"$out"'
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW INDEXES FROM show_acl.hidden_table;"); grep -q "hidden_idx" <<<"$out"'
+
+statement ok
+GRANT SELECT ON VIEW show_acl.hidden_view TO show_acl_user;
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_user -d "$SLT_DB" -Atc "SHOW VIEWS FROM show_acl;"); grep -q "show_acl.hidden_view" <<<"$out"'
+
+statement ok
+DROP SECRET show_acl.hidden_secret;
+
+statement ok
+DROP SCHEMA show_acl CASCADE;
+
+statement ok
+DROP USER show_acl_user;

e2e_test/sink/blackhole_sink.slt

@@ -36,4 +36,43 @@ statement ok
 DROP TABLE t5;
 
 statement ok
-FLUSH;
+FLUSH;
+
+# Verify SHOW SINKS visibility in the sink e2e environment.
+statement ok
+CREATE USER show_acl_sink_user WITH PASSWORD 'password';
+
+statement ok
+CREATE SCHEMA show_acl_sink;
+
+statement ok
+GRANT USAGE ON SCHEMA show_acl_sink TO show_acl_sink_user;
+
+statement ok
+CREATE TABLE show_acl_sink.hidden_table (v1 int primary key, v2 int);
+
+statement ok
+CREATE SINK show_acl_sink.hidden_sink AS SELECT * FROM show_acl_sink.hidden_table WITH (
+    connector = 'blackhole'
+);
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_sink_user -d "$SLT_DB" -Atc "SHOW SINKS FROM show_acl_sink;"); ! grep -q "hidden_sink" <<<"$out"'
+
+statement ok
+GRANT SELECT ON SINK show_acl_sink.hidden_sink TO show_acl_sink_user;
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_sink_user -d "$SLT_DB" -Atc "SHOW SINKS FROM show_acl_sink;"); grep -q "hidden_sink" <<<"$out"'
+
+statement ok
+DROP SINK show_acl_sink.hidden_sink;
+
+statement ok
+DROP TABLE show_acl_sink.hidden_table;
+
+statement ok
+DROP SCHEMA show_acl_sink;
+
+statement ok
+DROP USER show_acl_sink_user;

e2e_test/source_inline/connection/schema_registry.slt

@@ -53,4 +53,48 @@ drop table t;
 statement ok
 drop connection schema_registry_conn;
 
+# Verify connection SHOW visibility and secret-ref redaction in the schema-registry docker environment.
+statement ok
+CREATE USER show_acl_conn_user WITH PASSWORD 'password';
+
+statement ok
+CREATE SCHEMA show_acl_conn;
+
+statement ok
+GRANT USAGE ON SCHEMA show_acl_conn TO show_acl_conn_user;
+
+statement ok
+CREATE SECRET show_acl_conn.hidden_registry WITH (backend = 'meta') AS '${RISEDEV_SCHEMA_REGISTRY_URL}';
+
+statement ok
+CREATE CONNECTION show_acl_conn.visible_conn WITH (
+    type = 'schema_registry',
+    schema.registry = secret show_acl_conn.hidden_registry
+);
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_conn_user -d "$SLT_DB" -Atc "SHOW CONNECTIONS FROM show_acl_conn;"); ! grep -q "visible_conn" <<<"$out"'
+
+statement ok
+GRANT USAGE ON CONNECTION show_acl_conn.visible_conn TO show_acl_conn_user;
 
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_conn_user -d "$SLT_DB" -Atc "SHOW CONNECTIONS FROM show_acl_conn;"); grep -q "visible_conn" <<<"$out"; grep -q "<redacted>" <<<"$out"; ! grep -q "hidden_registry" <<<"$out"'
+
+statement ok
+GRANT USAGE ON SECRET show_acl_conn.hidden_registry TO show_acl_conn_user;
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_conn_user -d "$SLT_DB" -Atc "SHOW CONNECTIONS FROM show_acl_conn;"); grep -q "hidden_registry" <<<"$out"'
+
+statement ok
+DROP CONNECTION show_acl_conn.visible_conn;
+
+statement ok
+DROP SECRET show_acl_conn.hidden_registry;
+
+statement ok
+DROP SCHEMA show_acl_conn;
+
+statement ok
+DROP USER show_acl_conn_user;

e2e_test/source_inline/kafka/secret_dep.slt

@@ -59,3 +59,51 @@ drop secret test_secret_dep_schema.sec_1;
 
 statement ok
 set streaming_use_shared_source to true;
+
+# Verify SHOW SOURCES visibility in the Kafka docker environment.
+system ok
+rpk topic create show_acl_source_topic -p 1 || true
+
+statement ok
+CREATE USER show_acl_source_user WITH PASSWORD 'password';
+
+statement ok
+CREATE SCHEMA show_acl_source;
+
+statement ok
+GRANT USAGE ON SCHEMA show_acl_source TO show_acl_source_user;
+
+statement ok
+CREATE SECRET show_acl_source.hidden_topic WITH (backend = 'meta') AS 'show_acl_source_topic';
+
+statement ok
+CREATE SOURCE show_acl_source.hidden_source(x varchar)
+WITH(
+    ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+    topic = secret show_acl_source.hidden_topic,
+    scan.startup.mode = 'earliest'
+) FORMAT PLAIN ENCODE JSON;
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_source_user -d "$SLT_DB" -Atc "SHOW SOURCES FROM show_acl_source;"); ! grep -q "hidden_source" <<<"$out"'
+
+statement ok
+GRANT SELECT ON SOURCE show_acl_source.hidden_source TO show_acl_source_user;
+
+system ok
+bash -c 'set -euo pipefail; out=$(PGPASSWORD=password psql -h "$SLT_HOST" -p "$SLT_PORT" -U show_acl_source_user -d "$SLT_DB" -Atc "SHOW SOURCES FROM show_acl_source;"); grep -q "hidden_source" <<<"$out"'
+
+statement ok
+DROP SOURCE show_acl_source.hidden_source;
+
+statement ok
+DROP SECRET show_acl_source.hidden_topic;
+
+statement ok
+DROP SCHEMA show_acl_source;
+
+statement ok
+DROP USER show_acl_source_user;
+
+system ok
+rpk topic delete show_acl_source_topic || true;

src/frontend/src/catalog/system_catalog/rw_catalog/rw_connections.rs

@@ -18,7 +18,7 @@ use risingwave_frontend_macro::system_catalog;
 
 use crate::catalog::system_catalog::{SysCatalogReaderImpl, get_acl_items};
 use crate::error::Result;
-use crate::handler::create_connection::print_connection_params;
+use crate::handler::create_connection::print_connection_params_with_secret_visibility;
 
 #[derive(Fields)]
 struct RwConnection {
@@ -64,11 +64,13 @@ fn read_rw_connections(reader: &SysCatalogReaderImpl) -> Result<Vec<RwConnection
                         rw_connection.provider = conn.provider().into();
                     }
                     risingwave_pb::catalog::connection::Info::ConnectionParams(params) => {
-                        rw_connection.connection_params = print_connection_params(
-                            &reader.auth_context.database,
-                            params,
-                            &catalog_reader,
-                        );
+                        rw_connection.connection_params =
+                            print_connection_params_with_secret_visibility(
+                                &reader.auth_context.database,
+                                params,
+                                &catalog_reader,
+                                current_user,
+                            );
                     }
                 };
 

src/frontend/src/handler/create_connection.rs

@@ -37,6 +37,8 @@ use crate::error::ErrorCode::ProtocolError;
 use crate::error::{ErrorCode, Result, RwError};
 use crate::handler::HandlerArgs;
 use crate::session::SessionImpl;
+use crate::user::has_access_to_object;
+use crate::user::user_catalog::UserCatalog;
 use crate::utils::{resolve_privatelink_in_with_option, resolve_secret_ref_in_with_options};
 
 pub(crate) const CONNECTION_TYPE_PROP: &str = "type";
@@ -153,17 +155,50 @@ pub fn print_connection_params(
     db_name: &str,
     params: &PbConnectionParams,
     catalog_reader: &CatalogReadGuard,
+) -> String {
+    print_connection_params_impl(db_name, params, catalog_reader, None)
+}
+
+pub fn print_connection_params_with_secret_visibility(
+    db_name: &str,
+    params: &PbConnectionParams,
+    catalog_reader: &CatalogReadGuard,
+    current_user: &UserCatalog,
+) -> String {
+    print_connection_params_impl(db_name, params, catalog_reader, Some(current_user))
+}
+
+fn print_connection_params_impl(
+    db_name: &str,
+    params: &PbConnectionParams,
+    catalog_reader: &CatalogReadGuard,
+    current_user: Option<&UserCatalog>,
 ) -> String {
     let print_secret_ref = |secret_ref: &SecretRef| -> String {
-        // the lookup across all schemas in the database but should guarantee the secret exists
-        let (schema_name, secret_name) = catalog_reader
-            .find_schema_secret_by_secret_id(db_name, SecretId::from(secret_ref.secret_id))
-            .unwrap();
         let maybe_print_as = match secret_ref.get_ref_as().unwrap() {
             RefAsType::Text => "",
             RefAsType::File => " AS FILE",
             RefAsType::Unspecified => "",
         };
+
+        let secret_id = SecretId::from(secret_ref.secret_id);
+        let (schema_name, secret) = catalog_reader
+            .iter_schemas(db_name)
+            .unwrap()
+            .find_map(|schema| {
+                schema
+                    .get_secret_by_id(secret_id)
+                    .map(|secret| (schema.name.clone(), secret.clone()))
+            })
+            .unwrap();
+
+        if let Some(current_user) = current_user
+            && !has_access_to_object(current_user, secret.id, secret.owner)
+        {
+            return format!("SECRET <redacted>{}", maybe_print_as);
+        }
+
+        let secret_name = secret.name.clone();
         format!("SECRET {}.{}{}", schema_name, secret_name, maybe_print_as,)
     };
     let deref_secrets = params