users may only send non-ddl native sql to h2 (#25220)

* users may only send non-ddl native sql to h2

* tests for diasllowing h2 ddl

* improve error message on throw

* fix linter + cleanup the-exploit example.

* refix linter

* handle garbage input by failing to classify it.

If it h2 can't parse it, then that input cannot trigger a vulnerability.

Our parser simply chews through erroneous sql, and classifies statements
that it is able to parse.

* When h2 is running in client-side mode, do not parse sql

- SessionRemote indicates that we are currently client side

* refactor so that building parser is easier

* remove unused import SessionRemote

* Revert "refactor so that building parser is easier"

This reverts commit a41800131696de00d98e4eb7124d4d4b1b1cb33c.

* check client-side conns => import SessionRemote

* replace truncate function via inlining

* fix drop arg order

src/metabase/driver/h2.clj

@@ -20,7 +20,10 @@
             [metabase.util.i18n :refer [deferred-tru tru]]
             [metabase.util.ssh :as ssh])
   (:import [java.sql Clob ResultSet ResultSetMetaData]
-           java.time.OffsetTime))
+           java.time.OffsetTime
+           org.h2.command.Parser
+           org.h2.engine.Session
+           org.h2.engine.SessionRemote))
 
 ;; method impls live in this namespace
 (comment h2.actions/keep-me)
@@ -91,9 +94,66 @@
            (ex-info (tru "Running SQL queries against H2 databases using the default (admin) database user is forbidden.")
              {:type qp.error-type/db})))))))
 
+(defn- make-h2-parser [h2-db-id]
+  (with-open [conn (.getConnection (sql-jdbc.execute/datasource-with-diagnostic-info! :h2 h2-db-id))]
+    (let [inner-field (doto
+                          (.getDeclaredField (class conn) "inner")
+                        (.setAccessible true))
+          h2-jdbc-conn (.get inner-field conn)
+          session-field (doto
+                            (.getDeclaredField (class h2-jdbc-conn) "session")
+                          (.setAccessible true))
+          session (.get session-field h2-jdbc-conn)]
+      (cond
+        (instance? Session session)
+        (Parser. session)
+
+        ;; a SessionRemote cannot be used to make a parser
+        (instance? SessionRemote session)
+        ::client-side-session
+
+        :else
+        (throw (ex-info "Unknown session type" {:session session}))))))
+
+(defn- parse
+  ([h2-db-id s]
+   (let [h2-parser (make-h2-parser h2-db-id)]
+     (when-not (= ::client-side-session h2-parser)
+       (let [parse-method (doto (.getDeclaredMethod (class h2-parser)
+                                                    "parse"
+                                                    (into-array Class [java.lang.String]))
+                            (.setAccessible true))
+             parse-index-field (doto (.getDeclaredField (class h2-parser) "parseIndex")
+                                 (.setAccessible true))]
+         ;; parser moves parseIndex, so get-offset will be the index in the string that was parsed "up to"
+         (parse s
+                (fn parser [s]
+                  (try (.invoke parse-method h2-parser (object-array [s]))
+                       ;; need to chew through error scenarios because of a query like:
+                       ;;
+                       ;; vulnerability; abc;
+                       ;;
+                       ;; which would cause this parser to break w/o the error handling here, but this way we
+                       ;; still return the org.h2.command.ddl.* classes.
+                       (catch Throwable _ ::parse-fail)))
+                (fn get-offset [] (.get parse-index-field h2-parser)))))))
+  ([s parser get-offset] (vec (concat
+                               [(parser s)];; this call to parser parses up to the end of the first sql statement
+                               (let [more (apply str (drop (get-offset) s))] ;; more is the unparsed part of s
+                                 (when-not (str/blank? more)
+                                   (parse more parser get-offset)))))))
+
+(defn- check-disallow-ddl-commands [{:keys [database] :as query}]
+  (when query
+    (let [operations (parse database (-> query :native :query))
+          op-classes (map class operations)]
+      (when (some #(re-find #"org.h2.command.ddl." (str %)) op-classes)
+        (throw (ex-info "DDL commands are not allowed to be used with h2." {:classes op-classes}))))))
+
 (defmethod driver/execute-reducible-query :h2
   [driver query chans respond]
   (check-native-query-not-using-default-user query)
+  (check-disallow-ddl-commands query)
   ((get-method driver/execute-reducible-query :sql-jdbc) driver query chans respond))
 
 (defmethod sql.qp/add-interval-honeysql-form :h2