[SS-188] Check the token_uri in Google Cloud Service Account keys (#36997)

This PR adds a simple framework for checking the Secrets used by a
Connection when either the Secret or the Connection changes.

So far, it's just a hook to check GCP secrets for a bad `token_uri`
field.

This is a followup to #36694.

_Just for fun: [Opus vs
Fable](https://github.com/ublubu/materialize/compare/f1cc6f85bd0e70bfafc6aec843e54b64087fc762..ublubu:materialize:3eb3a022489401038eab4f84cbdff0a184ff2063?diff=split&w)
code comparison_

Author: ublubu

src/adapter/src/coord/sequencer/inner.rs

@@ -64,7 +64,6 @@ use mz_sql::plan::{
     StatementContext,
 };
 use mz_sql::pure::{PurifiedSourceExport, generate_subsource_statements};
-use mz_storage_types::connections::gcp::GcpServiceAccountKeyTokenUri;
 use mz_storage_types::sinks::StorageSinkDesc;
 // Import `plan` module, but only import select elements to avoid merge conflicts on use statements.
 use mz_sql::plan::{
@@ -625,6 +624,41 @@ impl Coordinator {
         }
     }
 
+    /// Applies `details.secret_content_guards()` by reading each guarded
+    /// secret's current contents. Must be called whenever a connection is
+    /// created or its options altered, before the change takes effect.
+    async fn check_connection_secret_content_guards(
+        &self,
+        details: &ConnectionDetails,
+    ) -> Result<(), AdapterError> {
+        for (secret_id, guard) in details.secret_content_guards() {
+            let contents = self.caching_secrets_reader.read_string(secret_id).await?;
+            guard(&contents)?;
+        }
+        Ok(())
+    }
+
+    /// Applies the secret-content guards of every connection that uses
+    /// `secret_id` against proposed new `contents` for that secret. Must be
+    /// called whenever a secret's contents change, before the new value is
+    /// persisted.
+    pub(super) fn check_secret_content_guards_of_dependents(
+        &self,
+        secret_id: CatalogItemId,
+        contents: &str,
+    ) -> Result<(), AdapterError> {
+        for dependent_id in self.catalog().get_entry(&secret_id).used_by() {
+            if let CatalogItem::Connection(conn) = self.catalog().get_entry(dependent_id).item() {
+                for (guarded_id, guard) in conn.details.secret_content_guards() {
+                    if guarded_id == secret_id {
+                        guard(contents)?;
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
     #[instrument]
     pub(super) async fn sequence_create_connection(
         &mut self,
@@ -664,22 +698,18 @@ impl Coordinator {
                 }
                 self.caching_secrets_reader.invalidate(connection_id);
             }
-            ConnectionDetails::Gcp(gcp) => {
-                // A service account key defines its own OAuth2 token URI.
-                // We only want to send requests to the actual Google OAuth2 token API,
-                // so we inspect the key as early as we can.
-                if let Err(err) = self
-                    .caching_secrets_reader
-                    .read_string(gcp.credentials_json)
-                    .await
-                    .and_then(|json| GcpServiceAccountKeyTokenUri::validate_json(&json))
-                {
-                    return ctx.retire(Err(err.into()));
-                }
-            }
             _ => (),
         };
 
+        // Inspect guarded secrets as early as we can, before the connection is
+        // installed in the catalog.
+        if let Err(err) = self
+            .check_connection_secret_content_guards(&plan.connection.details)
+            .await
+        {
+            return ctx.retire(Err(err));
+        }
+
         if plan.validate {
             let internal_cmd_tx = self.internal_cmd_tx.clone();
             let transient_revision = self.catalog().transient_revision();
@@ -3646,6 +3676,15 @@ impl Coordinator {
             }
         };
 
+        // Inspect guarded secrets whether or not validation was requested,
+        // before the altered connection is installed in the catalog.
+        if let Err(err) = self
+            .check_connection_secret_content_guards(&conn.details)
+            .await
+        {
+            return ctx.retire(Err(err));
+        }
+
         if validate {
             let connection = conn
                 .details

src/adapter/src/coord/sequencer/inner/secret.rs

@@ -282,6 +282,8 @@ impl Coordinator {
         let caching_secrets_reader = self.caching_secrets_reader.clone();
         let secrets_controller = Arc::clone(&self.secrets_controller);
         let payload = self.extract_secret(session, &mut secret_as)?;
+        let contents = std::str::from_utf8(&payload).expect("validated as UTF-8 by extract_secret");
+        self.check_secret_content_guards_of_dependents(id, contents)?;
         let span = Span::current();
         Ok(StageResult::HandleRetire(mz_ore::task::spawn(
             || "alter secret ensure",

src/sql/src/plan.rs

@@ -57,7 +57,7 @@ use mz_sql_parser::ast::{
 };
 use mz_ssh_util::keys::SshKeyPair;
 use mz_storage_types::connections::aws::AwsConnection;
-use mz_storage_types::connections::gcp::GcpConnection;
+use mz_storage_types::connections::gcp::{GcpConnection, GcpServiceAccountKeyTokenUri};
 use mz_storage_types::connections::inline::ReferencedConnection;
 use mz_storage_types::connections::{
     AwsPrivatelinkConnection, CsrConnection, GlueSchemaRegistryConnection,
@@ -1715,6 +1715,31 @@ impl ConnectionDetails {
             }
         }
     }
+
+    /// Secrets whose *contents* this connection places requirements on, paired
+    /// with the check to apply. Callers must re-apply these checks whenever the
+    /// connection is created or altered, and whenever the contents of one of
+    /// the returned secrets change (e.g. `ALTER SECRET`).
+    ///
+    /// We rely on the caller to actually execute these checks because we don't know:
+    /// - which secrets the caller cares about
+    /// - which secrets require an async operation to fetch
+    ///
+    /// For example, the ALTER SECRET caller should only perform checks on its own secret,
+    /// while the ALTER CONNECTION caller fetches and checks every secret from its connection.
+    pub fn secret_content_guards(
+        &self,
+    ) -> Vec<(CatalogItemId, fn(&str) -> Result<(), anyhow::Error>)> {
+        match self {
+            // A service-account key defines its own OAuth2 token URI. We only
+            // want to send requests to the actual Google OAuth2 token API.
+            ConnectionDetails::Gcp(gcp) => vec![(
+                gcp.credentials_json,
+                GcpServiceAccountKeyTokenUri::validate_json,
+            )],
+            _ => vec![],
+        }
+    }
 }
 
 #[derive(Debug, Clone, Serialize, PartialEq, Eq, Ord, PartialOrd, Hash)]

test/testdrive/connection-create-external.td

@@ -263,3 +263,37 @@ contains:token_uri must be https://oauth2.googleapis.com/token
 # In fact, the GCP allowlist should even apply when VALIDATE = false.
 ! CREATE CONNECTION gcp_ssrf_private_conn TO GCP (SERVICE ACCOUNT KEY = SECRET gcp_ssrf_private) WITH (VALIDATE = false);
 contains:token_uri must be https://oauth2.googleapis.com/token
+
+# The CREATE-time check is not enough: ALTER CONNECTION and ALTER SECRET can both
+# install a malicious token_uri into an already-created GCP connection, so the
+# allowlist must guard those paths too.
+
+# A benign service-account key whose token_uri is the real Google endpoint. It
+# passes the allowlist, so the connection and secret below are created cleanly
+# and give the ALTER paths a valid starting point to subvert.
+> CREATE SECRET gcp_good AS '{"type":"service_account","project_id":"x","client_email":"a@b.com","private_key":"${gcp-ssrf-key}","token_uri":"https://oauth2.googleapis.com/token"}'
+> CREATE CONNECTION gcp_alter_conn TO GCP (SERVICE ACCOUNT KEY = SECRET gcp_good) WITH (VALIDATE = false);
+
+# ALTER CONNECTION: swapping in a SERVICE ACCOUNT KEY that points at a private
+# token_uri is the same SSRF primitive as CREATE and must be rejected, whether or
+# not validation is requested.
+! ALTER CONNECTION gcp_alter_conn SET (SERVICE ACCOUNT KEY = SECRET gcp_ssrf_private) WITH (VALIDATE = true);
+contains:token_uri must be https://oauth2.googleapis.com/token
+
+! ALTER CONNECTION gcp_alter_conn SET (SERVICE ACCOUNT KEY = SECRET gcp_ssrf_private) WITH (VALIDATE = false);
+contains:token_uri must be https://oauth2.googleapis.com/token
+
+# ALTER SECRET: rewriting the secret behind a GCP connection slips a malicious
+# token_uri past the CREATE-time check entirely, since the connection was already
+# validated against the benign key above. This must be rejected as well.
+! ALTER SECRET gcp_good AS '{"type":"service_account","project_id":"x","client_email":"a@b.com","private_key":"${gcp-ssrf-key}","token_uri":"http://127.0.0.1:1/token"}'
+contains:token_uri must be https://oauth2.googleapis.com/token
+
+# Benign changes must still go through: a guard that rejected every ALTER would
+# also pass the tests above. ALTER SECRET first, while gcp_alter_conn still
+# depends on gcp_good -- afterwards the connection moves off it and the
+# dependent-based guard no longer applies.
+> ALTER SECRET gcp_good AS '{"type":"service_account","project_id":"y","client_email":"a@b.com","private_key":"${gcp-ssrf-key}","token_uri":"https://oauth2.googleapis.com/token"}'
+
+> CREATE SECRET gcp_good_2 AS '{"type":"service_account","project_id":"x","client_email":"a@b.com","private_key":"${gcp-ssrf-key}","token_uri":"https://oauth2.googleapis.com/token"}'
+> ALTER CONNECTION gcp_alter_conn SET (SERVICE ACCOUNT KEY = SECRET gcp_good_2) WITH (VALIDATE = false);