Only attempt to sync schemas that have some/any possible permissions (#15291)

Only attempt to sync schemas that have some/any possible permissions

Make syncable-schemas multimethod, whose base implementation is the same as the previous private fn

Override syncable-schemas for Redshift to add a filtering step to the transducer pipeline to check for the schema privilege, by calling the HAS_SCHEMA_PRIVILEGE Redshift function to check for USAGE permission

Making all-schemas function public since it's now invoked from the Redshift driver

Fixing redshift-types-test to remove order flakiness

Adding test that confirms that a real schemas with no permissions aren't synced

modules/drivers/redshift/src/metabase/driver/redshift.clj

@@ -12,6 +12,7 @@
             [metabase.driver.sql-jdbc.execute :as sql-jdbc.execute]
             [metabase.driver.sql-jdbc.execute.legacy-impl :as legacy]
             [metabase.driver.sql-jdbc.sync :as sql-jdbc.sync]
+            [metabase.driver.sql-jdbc.sync.describe-database :as sync.describe-database]
             [metabase.driver.sql.query-processor :as sql.qp]
             [metabase.mbql.util :as mbql.u]
             [metabase.public-settings :as pubset]
@@ -141,8 +142,9 @@
         (.close ^Connection conn)
         (throw e)))))
 
-(defn- prepare-statement [^Connection conn sql]
-  (.prepareStatement conn sql
+(defn- prepare-statement ^PreparedStatement [^Connection conn sql]
+  (.prepareStatement conn
+                     sql
                      ResultSet/TYPE_FORWARD_ONLY
                      ResultSet/CONCUR_READ_ONLY
                      ResultSet/CLOSE_CURSORS_AT_COMMIT))
@@ -151,11 +153,11 @@
   "Quotes a string literal so that it can be safely inserted into Redshift queries, by returning the result of invoking
   the Redshift QUOTE_LITERAL function on the given string (which is set in a PreparedStatement as a parameter)."
   [^Connection conn ^String s]
-  (with-open [stmt ^PreparedStatement (prepare-statement conn "SELECT QUOTE_LITERAL(?);")]
-    (.setString stmt 1 s)
-    (with-open [rs ^ResultSet (.executeQuery stmt)]
-      (when (.next rs)
-        (.getString rs 1)))))
+  (with-open [stmt (doto (prepare-statement conn "SELECT QUOTE_LITERAL(?);")
+                     (.setString 1 s))
+              rs   (.executeQuery stmt)]
+    (when (.next rs)
+      (.getString rs 1))))
 
 (defn- quote-literal-for-database
   "This function invokes quote-literal-for-connection with a connection for the given database. See its docstring for
@@ -251,3 +253,34 @@
                               :filter_values       (field->parameter-value query)})
        " */ "
        (qputil/default-query->remark query)))
+
+(defn- reducible-schemas-with-usage-permissions
+  "Takes something `reducible` that returns a collection of string schema names (e.g. an `Eduction`) and returns an
+  `IReduceInit` that filters out schemas for which the DB user has no schema privileges."
+  [^Connection conn reducible]
+  (reify clojure.lang.IReduceInit
+    (reduce [_ rf init]
+      (with-open [stmt (prepare-statement conn "SELECT HAS_SCHEMA_PRIVILEGE(?, 'USAGE');")]
+        (reduce
+         rf
+         init
+         (eduction
+          (filter (fn [^String table-schema]
+                    (try
+                      (with-open [rs (.executeQuery (doto stmt (.setString 1 table-schema)))]
+                        (let [has-perm? (and (.next rs)
+                                             (.getBoolean rs 1))]
+                          (or has-perm?
+                              (log/tracef "Ignoring schema %s because no USAGE privilege on it" table-schema))))
+                      (catch Throwable e
+                        (log/error e (trs "Error checking schema permissions"))
+                        false))))
+          reducible))))))
+
+(defmethod sql-jdbc.sync/syncable-schemas :redshift
+  [driver conn metadata]
+  (reducible-schemas-with-usage-permissions
+   conn
+   (eduction
+    (remove (set (sql-jdbc.sync/excluded-schemas driver)))
+    (sync.describe-database/all-schemas metadata))))

modules/drivers/redshift/test/metabase/driver/redshift_test.clj

 (ns metabase.driver.redshift-test
-  (:require [clojure.string :as str]
+  (:require [clojure.java.jdbc :as jdbc]
+            [clojure.string :as str]
             [clojure.test :refer :all]
-            [environ.core :as env]
+            [metabase.driver.redshift :as redshift]
+            [metabase.driver.sql-jdbc.connection :as sql-jdbc.conn]
             [metabase.driver.sql-jdbc.execute :as sql-jdbc.execute]
+            [metabase.driver.sql-jdbc.sync :as sql-jdbc.sync]
+            [metabase.driver.sql-jdbc.sync.describe-database :as sync.describe-database]
             [metabase.models.database :refer [Database]]
             [metabase.models.field :refer [Field]]
             [metabase.models.setting :as setting]
@@ -15,9 +19,10 @@
             [metabase.test.data.interface :as tx]
             [metabase.test.data.redshift :as rstest]
             [metabase.test.fixtures :as fixtures]
+            [metabase.test.util :as tu]
             [metabase.util :as u]
             [toucan.db :as db])
-  (:import (java.sql ResultSetMetaData ResultSet)
+  (:import [java.sql ResultSet ResultSetMetaData]
            metabase.plugins.jdbc_proxy.ProxyDriver))
 
 (use-fixtures :once (fixtures/initialize :plugins))
@@ -197,11 +202,11 @@
                view-nm))
           (let [table-id (db/select-one-id Table :db_id (u/the-id database), :name view-nm)]
             ;; and its columns' :base_type should have been identified correctly
-            (is (= [{:name "weird_varchar", :database_type "character varying(50)", :base_type :type/Text}
-                    {:name "numeric_col",   :database_type "numeric(10,2)",         :base_type :type/Decimal}]
+            (is (= [{:name "numeric_col",   :database_type "numeric(10,2)",         :base_type :type/Decimal}
+                    {:name "weird_varchar", :database_type "character varying(50)", :base_type :type/Text}]
                    (map
                     (partial into {})
-                    (db/select [Field :name :database_type :base_type] :table_id table-id))))))))))
+                    (db/select [Field :name :database_type :base_type] :table_id table-id {:order-by [:name]}))))))))))
 
 (defn- assert-jdbc-url-fetch-size [db fetch-size]
   (with-open [conn (.getConnection (sql-jdbc.execute/datasource db))]
@@ -231,3 +236,59 @@
                                    (mt/native-query)
                                    (qp/process-query)
                                    (mt/rows)))))))))))))
+
+(deftest syncable-schemas-test
+  (mt/test-driver :redshift
+    (testing "Should filter out schemas for which the user has no perms"
+      ;; create a random username and random schema name, and grant the user USAGE permission for it
+      (let [temp-username (str/lower-case (tu/random-name))
+            random-schema (str/lower-case (tu/random-name))
+            user-pw       "Password1234"
+            db-det        (:details (mt/db))]
+        (#'rstest/execute! (str "CREATE SCHEMA %s;"
+                                "CREATE USER %s PASSWORD '%s';%n"
+                                "GRANT USAGE ON SCHEMA %s TO %s;%n")
+                           random-schema
+                           temp-username
+                           user-pw
+                           random-schema
+                           temp-username)
+        (try
+          (mt/with-temp Database [db {:engine :redshift, :details (assoc db-det :user temp-username :password user-pw)}]
+            (with-open [conn (jdbc/get-connection (sql-jdbc.conn/db->pooled-connection-spec db))]
+              (let [schemas (reduce conj
+                                    #{}
+                                    (sql-jdbc.sync/syncable-schemas :redshift conn (.getMetaData conn)))]
+                ;; the syncable-schemas for the user should contain the newly created random schema
+                (is (contains? schemas random-schema))
+                ;; but it should not contain the current session-schema name (since that was never granted)
+                (is (not (contains? schemas rstest/session-schema-name))))))
+          (finally
+            (#'rstest/execute! (str "REVOKE USAGE ON SCHEMA %s FROM %s;%n"
+                                    "DROP USER IF EXISTS %s;%n"
+                                    "DROP SCHEMA IF EXISTS %s;%n")
+             random-schema
+             temp-username
+             temp-username
+             random-schema)))))
+    (testing "Should filter out non-existent schemas (for which nobody has permissions)"
+      (let [fake-schema-name (u/qualified-name ::fake-schema)]
+        ;; override `all-schemas` so it returns our fake schema in addition to the real ones.
+        (with-redefs [sync.describe-database/all-schemas (let [orig sync.describe-database/all-schemas]
+                                                           (fn [metadata]
+                                                             (eduction
+                                                              cat
+                                                              [(orig metadata) [fake-schema-name]])))]
+          (let [jdbc-spec (sql-jdbc.conn/db->pooled-connection-spec (mt/db))]
+            (with-open [conn (jdbc/get-connection jdbc-spec)]
+              (letfn [(schemas []
+                        (reduce
+                         conj
+                         #{}
+                         (sql-jdbc.sync/syncable-schemas :redshift conn (.getMetaData conn))))]
+                (testing "if schemas-with-usage-permissions is disabled, the ::fake-schema should come back"
+                  (with-redefs [redshift/reducible-schemas-with-usage-permissions (fn [_ reducible]
+                                                                                    reducible)]
+                    (is (contains? (schemas) fake-schema-name))))
+                (testing "normally, ::fake-schema should be filtered out (because it does not exist)"
+                  (is (not (contains? (schemas) fake-schema-name))))))))))))

src/metabase/driver/sql_jdbc/sync.clj

@@ -15,7 +15,8 @@
   db-default-timezone
   excluded-schemas
   fallback-metadata-query
-  have-select-privilege?]
+  have-select-privilege?
+  syncable-schemas]
 
  [sync.describe-table
   add-table-pks

src/metabase/driver/sql_jdbc/sync/describe_database.clj

@@ -13,15 +13,17 @@
 
 (defmethod i/excluded-schemas :sql-jdbc [_] nil)
 
-(defn- all-schemas [^DatabaseMetaData metadata]
-  {:pre [(instance? DatabaseMetaData metadata)]}
+(defn all-schemas
+  "Get a *reducible* sequence of all string schema names for the current database from its JDBC database metadata."
+  [^DatabaseMetaData metadata]
+  {:added "0.39.0", :pre [(instance? DatabaseMetaData metadata)]}
   (common/reducible-results
    #(.getSchemas metadata)
    (fn [^ResultSet rs]
      #(.getString rs "TABLE_SCHEM"))))
 
-(defn- syncable-schemas
-  [driver metadata]
+(defmethod i/syncable-schemas :sql-jdbc
+  [driver _ metadata]
   (eduction (remove (set (i/excluded-schemas driver)))
             (all-schemas metadata)))
 
@@ -94,7 +96,7 @@
                      (db-tables driver metadata schema db-name-or-nil)))
            (filter (fn [{table-schema :schema, table-name :name}]
                      (i/have-select-privilege? driver conn table-schema table-name))))
-     (syncable-schemas driver metadata))))
+     (i/syncable-schemas driver conn metadata))))
 
 (defmethod i/active-tables :sql-jdbc
   [driver connection]

src/metabase/driver/sql_jdbc/sync/interface.clj

@@ -27,6 +27,14 @@
   driver/dispatch-on-initialized-driver
   :hierarchy #'driver/hierarchy)
 
+(defmulti syncable-schemas
+  "Return a reducible sequence of string names of schemas that should be synced for the given database. Schemas for
+  which the current DB user has no `SELECT` permissions should be filtered out. The default implementation will fetch
+  a sequence of all schema names from the JDBC database metadata and filter out any schemas in `excluded-schemas`."
+  {:added "0.39.0", :arglists '([driver ^java.sql.Connection connection ^java.sql.DatabaseMetaData metadata])}
+  driver/dispatch-on-initialized-driver
+  :hierarchy #'driver/hierarchy)
+
 (defmulti database-type->base-type
   "Given a native DB column type (as a keyword), return the corresponding `Field` `base-type`, which should derive from
   `:type/*`. You can use `pattern-based-database-type->base-type` in this namespace to implement this using regex