diff --git a/src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java b/src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java
index 4ebae9bb00..7f8c0716f1 100644
--- a/src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java
+++ b/src/main/java/io/cdap/plugin/gcp/common/GCPUtils.java
@@ -16,6 +16,10 @@
 
 package io.cdap.plugin.gcp.common;
 
+import com.google.api.client.json.GenericJson;
+import com.google.api.client.json.JsonObjectParser;
+import com.google.api.client.json.gson.GsonFactory;
+import com.google.api.client.util.Data;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.ExternalAccountCredentials;
@@ -35,6 +39,7 @@
 import com.google.cloud.storage.StorageException;
 import com.google.cloud.storage.StorageOptions;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.io.ByteStreams;
 import com.google.gson.reflect.TypeToken;
 import io.cdap.plugin.gcp.gcs.GCSPath;
 import io.cdap.plugin.gcp.gcs.ServiceAccountAccessTokenProvider;
@@ -94,6 +99,16 @@ public class GCPUtils {
   public static final String GCE_METADATA_SERVER_ERROR_SUPPORTED_DOC_URL =
     "https://cloud.google.com/compute/docs/troubleshooting/troubleshoot-metadata-server";
 
+  // Credential configuration type for workload identity federation
+  private static final String EXTERNAL_ACCOUNT_TYPE = "external_account";
+  // Google endpoints an external account configuration may send tokens to, including the locational (.rep) and
+  // mTLS (.mtls) variants generated by 'gcloud iam workload-identity-pools create-cred-config'. See
+  // https://cloud.google.com/docs/authentication/client-libraries#validate_other_credential_configurations
+  private static final Pattern STS_TOKEN_URL_PATTERN = Pattern.compile(
+    "^https://sts(\\.[a-z0-9-]+\\.rep)?(\\.mtls)?\\.googleapis\\.com/v1/token$");
+  private static final Pattern IAM_IMPERSONATION_URL_PATTERN = Pattern.compile(
+    "^https://iamcredentials(\\.mtls)?\\.googleapis\\.com/v1/projects/-/serviceAccounts/[^/?#:]+:generateAccessToken$");
+
   /**
    * Load a service account from the local file system.
    *
@@ -117,9 +132,68 @@ public static GoogleCredentials loadServiceAccountFileCredentials(String path) t
   public static GoogleCredentials loadServiceAccountCredentials(String serviceAccount,
                                                                 boolean isServiceAccountFilePath)
     throws IOException {
+    byte[] serviceAccountBytes;
     try (InputStream inputStream = openServiceAccount(serviceAccount, isServiceAccountFilePath)) {
-      return GoogleCredentials.fromStream(inputStream);
+      serviceAccountBytes = ByteStreams.toByteArray(inputStream);
+    }
+    validateExternalAccountCredentials(serviceAccountBytes);
+    return GoogleCredentials.fromStream(new ByteArrayInputStream(serviceAccountBytes));
+  }
+
+  /**
+   * Validates a credential configuration of type 'external_account' before it is used.
+   *
+   * An external account (workload identity federation) configuration instructs the auth library where to obtain a
+   * subject token from ({@code credential_source}) and where to send it to: the subject token is sent to
+   * {@code token_url} and the resulting access token to {@code service_account_impersonation_url}. When the
+   * configuration is supplied by a user, these two endpoints must be the Google endpoints documented at
+   * https://cloud.google.com/docs/authentication/client-libraries#validate_other_credential_configurations,
+   * otherwise the configuration could be used to read the ambient credentials of the environment and send them to
+   * an arbitrary endpoint. With both endpoints pinned to Google, whatever the configuration reads can only be sent
+   * to Google.
+   *
+   * The content is parsed with the same parser as {@link GoogleCredentials#fromStream(InputStream)} so that both
+   * see identical content. Content that does not parse fails here with the library's own error. Configurations of
+   * any other type are not validated and are handled by the library as before.
+   *
+   * @param credentialBytes the raw content of the credential configuration
+   * @throws IOException if the content cannot be parsed, or is of type 'external_account' and is not allowed
+   */
+  private static void validateExternalAccountCredentials(byte[] credentialBytes) throws IOException {
+    GenericJson json = new JsonObjectParser(GsonFactory.getDefaultInstance())
+      .parseAndClose(new ByteArrayInputStream(credentialBytes), StandardCharsets.UTF_8, GenericJson.class);
+    if (json == null || !EXTERNAL_ACCOUNT_TYPE.equals(json.get("type"))) {
+      return;
+    }
+
+    String tokenUrl = getStringField(json, "token_url");
+    if (tokenUrl == null || !STS_TOKEN_URL_PATTERN.matcher(tokenUrl).matches()) {
+      throw new IOException("Invalid external account credentials: 'token_url' must be "
+                              + "https://sts.googleapis.com/v1/token or a Google locational or mTLS variant of it.");
+    }
+    String impersonationUrl = getStringField(json, "service_account_impersonation_url");
+    if (impersonationUrl != null && !IAM_IMPERSONATION_URL_PATTERN.matcher(impersonationUrl).matches()) {
+      throw new IOException("Invalid external account credentials: 'service_account_impersonation_url' must be "
+                              + "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/"
+                              + "<service account email>:generateAccessToken or the Google mTLS variant of it.");
+    }
+  }
+
+  /**
+   * Returns the value of the given field as a string, or null if the field is absent or null.
+   *
+   * @throws IOException if the field is present but not a string
+   */
+  @Nullable
+  private static String getStringField(Map<String, Object> json, String field) throws IOException {
+    Object value = json.get(field);
+    if (value == null || Data.isNull(value)) {
+      return null;
+    }
+    if (!(value instanceof String)) {
+      throw new IOException(String.format("Invalid external account credentials: '%s' must be a string.", field));
     }
+    return (String) value;
   }
 
   /**
diff --git a/src/test/java/io/cdap/plugin/gcp/common/GCPUtilsTest.java b/src/test/java/io/cdap/plugin/gcp/common/GCPUtilsTest.java
new file mode 100644
index 0000000000..358398d533
--- /dev/null
+++ b/src/test/java/io/cdap/plugin/gcp/common/GCPUtilsTest.java
@@ -0,0 +1,333 @@
+/*
+ * Copyright © 2026 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.gcp.common;
+
+import com.google.auth.oauth2.ExternalAccountCredentials;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auth.oauth2.ServiceAccountCredentials;
+import com.google.common.io.ByteStreams;
+import com.sun.net.httpserver.HttpServer;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.security.KeyPairGenerator;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Tests for {@link GCPUtils#loadServiceAccountCredentials(String, boolean)}.
+ */
+public class GCPUtilsTest {
+  private static final String STS_URL = "https://sts.googleapis.com/v1/token";
+  private static final String IMPERSONATION_URL = "https://iamcredentials.googleapis.com/v1/projects/-/"
+    + "serviceAccounts/sa@p.iam.gserviceaccount.com:generateAccessToken";
+  private static final String FAKE_SA_TOKEN = "fake-service-agent-token";
+
+  // local stand-ins for the GCE metadata server and an attacker controlled token endpoint
+  private HttpServer metadataServer;
+  private HttpServer attackerServer;
+  private final AtomicInteger metadataHits = new AtomicInteger();
+  private final AtomicReference<String> attackerBody = new AtomicReference<>();
+
+  @Before
+  public void setUp() throws IOException {
+    metadataServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+    metadataServer.createContext("/computeMetadata/v1/instance/service-accounts/default/token", exchange -> {
+      metadataHits.incrementAndGet();
+      byte[] body = ("{\"access_token\":\"" + FAKE_SA_TOKEN + "\",\"expires_in\":3599,\"token_type\":\"Bearer\"}")
+        .getBytes(StandardCharsets.UTF_8);
+      exchange.sendResponseHeaders(200, body.length);
+      try (OutputStream os = exchange.getResponseBody()) {
+        os.write(body);
+      }
+    });
+    metadataServer.start();
+
+    attackerServer = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+    attackerServer.createContext("/token-exchange", exchange -> {
+      try (InputStream is = exchange.getRequestBody()) {
+        attackerBody.set(new String(ByteStreams.toByteArray(is), StandardCharsets.UTF_8));
+      }
+      byte[] body = "{}".getBytes(StandardCharsets.UTF_8);
+      exchange.sendResponseHeaders(500, body.length);
+      try (OutputStream os = exchange.getResponseBody()) {
+        os.write(body);
+      }
+    });
+    attackerServer.start();
+  }
+
+  @After
+  public void tearDown() {
+    metadataServer.stop(0);
+    attackerServer.stop(0);
+  }
+
+  @Test
+  public void testServiceAccountJson() throws Exception {
+    KeyPairGenerator keyPairGenerator = KeyPairGenerator.getInstance("RSA");
+    keyPairGenerator.initialize(2048);
+    String pem = "-----BEGIN PRIVATE KEY-----\\n"
+      + Base64.getEncoder().encodeToString(keyPairGenerator.generateKeyPair().getPrivate().getEncoded())
+      + "\\n-----END PRIVATE KEY-----\\n";
+    String json = "{\"type\":\"service_account\",\"project_id\":\"p\",\"private_key_id\":\"k\","
+      + "\"private_key\":\"" + pem + "\",\"client_email\":\"sa@p.iam.gserviceaccount.com\",\"client_id\":\"1\"}";
+
+    GoogleCredentials credentials = GCPUtils.loadServiceAccountCredentials(json, false);
+
+    Assert.assertTrue(credentials instanceof ServiceAccountCredentials);
+    Assert.assertNotNull(credentials.createScoped(GCPUtils.BIGQUERY_SCOPES));
+  }
+
+  @Test
+  public void testExternalAccountAsGeneratedByCdapWorkloadIdentity() throws Exception {
+    // shape produced by cdap-kubernetes WorkloadIdentityUtil
+    String json = "{\"type\":\"external_account\",\"audience\":\"identitynamespace:pool:provider\","
+      + "\"subject_token_type\":\"urn:ietf:params:oauth:token-type:jwt\","
+      + "\"token_url\":\"" + STS_URL + "\","
+      + "\"service_account_impersonation_url\":\"" + IMPERSONATION_URL + "\","
+      + "\"credential_source\":{\"file\":\"/var/run/secrets/tokens/gcp-ksa/token\"}}";
+
+    GoogleCredentials credentials = GCPUtils.loadServiceAccountCredentials(json, false);
+
+    Assert.assertTrue(credentials instanceof ExternalAccountCredentials);
+  }
+
+  @Test
+  public void testExternalAccountWithoutImpersonationUrl() throws Exception {
+    String json = externalAccount(STS_URL, null, "{\"file\":\"/var/run/secrets/token\"}");
+
+    GoogleCredentials credentials = GCPUtils.loadServiceAccountCredentials(json, false);
+
+    Assert.assertTrue(credentials instanceof ExternalAccountCredentials);
+  }
+
+  @Test
+  public void testExternalAccountWithUrlAndAwsSources() throws Exception {
+    // credential_source is not restricted; only the endpoints tokens are sent to are
+    String url = externalAccount(STS_URL, null, urlSource("http://localhost:8080/token"));
+    String aws = externalAccount(STS_URL, null, "{\"environment_id\":\"aws1\","
+      + "\"region_url\":\"http://169.254.169.254/latest/meta-data/placement/availability-zone\","
+      + "\"url\":\"http://169.254.169.254/latest/meta-data/iam/security-credentials\","
+      + "\"regional_cred_verification_url\":\"https://sts.{region}.amazonaws.com?Action=GetCallerIdentity"
+      + "&Version=2011-06-15\"}");
+
+    Assert.assertTrue(GCPUtils.loadServiceAccountCredentials(url, false) instanceof ExternalAccountCredentials);
+    Assert.assertTrue(GCPUtils.loadServiceAccountCredentials(aws, false) instanceof ExternalAccountCredentials);
+  }
+
+  @Test
+  public void testExternalAccountWithGcloudGeneratedEndpoints() throws Exception {
+    // locational (--sts-location) and mTLS (--enable-mtls) endpoints generated by
+    // 'gcloud iam workload-identity-pools create-cred-config'
+    String[][] endpoints = {
+      { "https://sts.us-central1.rep.googleapis.com/v1/token", null },
+      { "https://sts.mtls.googleapis.com/v1/token", null },
+      { "https://sts.mtls.googleapis.com/v1/token",
+        "https://iamcredentials.mtls.googleapis.com/v1/projects/-/serviceAccounts/sa@p.iam.gserviceaccount.com"
+          + ":generateAccessToken" },
+    };
+    for (String[] endpoint : endpoints) {
+      String json = externalAccount(endpoint[0], endpoint[1], "{\"file\":\"/var/run/secrets/token\"}");
+      GoogleCredentials credentials = GCPUtils.loadServiceAccountCredentials(json, false);
+      Assert.assertTrue(json, credentials instanceof ExternalAccountCredentials);
+    }
+  }
+
+  @Test
+  public void testExternalAccountRequiresGoogleTokenUrl() {
+    String[] badTokenUrls = {
+      "https://attacker.example.com/token",
+      "https://sts.googleapis.com.attacker.example.com/v1/token",
+      "https://attacker.example.com/?sts.googleapis.com",
+      "https://user@sts.googleapis.com@attacker.example.com/v1/token",
+      "http://sts.googleapis.com/v1/token",
+      "https://sts.googleapis.com/v1/token/",
+      "https://sts.googleapis.com/v1/token?x=1",
+      "https://STS.googleapis.com/v1/token",
+      "https://sts-myendpoint.p.googleapis.com/v1/token",
+      "https://sts.attacker.example.com/v1/token",
+      "https://sts.us-central1.rep.attacker.example.com/v1/token",
+      "https://sts.mtls.googleapis.com.attacker.example.com/v1/token",
+      "https://sts..rep.googleapis.com/v1/token",
+      "https://sts.us-central1.googleapis.com/v1/token",
+      " https://sts.googleapis.com/v1/token",
+    };
+    for (String tokenUrl : badTokenUrls) {
+      assertRejected(externalAccount(tokenUrl, null, "{\"file\":\"/tmp/token\"}"), "token_url");
+    }
+    // absent, null and non-string values
+    assertRejected(externalAccount(null, null, "{\"file\":\"/tmp/token\"}"), "token_url");
+    assertRejected("{\"type\":\"external_account\",\"token_url\":null,\"credential_source\":{\"file\":\"/t\"}}",
+                   "token_url");
+    assertRejected("{\"type\":\"external_account\",\"token_url\":[\"" + STS_URL + "\"],"
+                     + "\"credential_source\":{\"file\":\"/t\"}}", "token_url");
+  }
+
+  @Test
+  public void testExternalAccountRequiresGoogleImpersonationUrl() {
+    String iam = "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/";
+    String[] badUrls = {
+      "https://attacker.example.com/generateAccessToken",
+      "https://iamcredentials.googleapis.com.attacker.example.com/v1/projects/-/serviceAccounts/x:generateAccessToken",
+      "https://iamcredentials.mtls.googleapis.com.attacker.example.com/v1/projects/-/serviceAccounts/x"
+        + ":generateAccessToken",
+      "http://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/x:generateAccessToken",
+      "https://iamcredentials.us-central1.rep.googleapis.com/v1/projects/-/serviceAccounts/x:generateAccessToken",
+      iam + "sa@p.iam.gserviceaccount.com:generateIdToken",
+      iam + "sa@p.iam.gserviceaccount.com:generateAccessToken/../../../x",
+      iam + "sa@p.iam.gserviceaccount.com:generateAccessToken?x=1",
+      "https://iamcredentials.googleapis.com/v1/projects/p/serviceAccounts/sa@p.iam.gserviceaccount.com"
+        + ":generateAccessToken",
+    };
+    for (String url : badUrls) {
+      assertRejected(externalAccount(STS_URL, url, "{\"file\":\"/tmp/token\"}"), "service_account_impersonation_url");
+    }
+    assertRejected("{\"type\":\"external_account\",\"token_url\":\"" + STS_URL + "\","
+                     + "\"service_account_impersonation_url\":123,\"credential_source\":{\"file\":\"/t\"}}",
+                   "service_account_impersonation_url");
+  }
+
+  /**
+   * Reproduces b/501543027: the credential configuration points the subject token source at the metadata server
+   * and the token exchange at an attacker controlled endpoint. Loading must fail before anything is contacted.
+   */
+  @Test
+  public void testTokenTheftPayloadIsRejectedBeforeUse() {
+    String json = externalAccount(attackerUrl(), null, urlSource(metadataUrl()));
+
+    try {
+      GoogleCredentials credentials = GCPUtils.loadServiceAccountCredentials(json, false);
+      try {
+        // any API call does this under the hood
+        credentials.createScoped(Collections.singleton("https://www.googleapis.com/auth/cloud-platform"))
+          .refreshAccessToken();
+      } catch (Exception ignored) {
+        // the exchange is expected to fail, but by then the token would already have been sent
+      }
+      Assert.fail("Credential configuration should have been rejected");
+    } catch (IOException e) {
+      Assert.assertTrue(e.getMessage(), e.getMessage().contains("token_url"));
+    }
+    Assert.assertEquals("subject token source must not be contacted", 0, metadataHits.get());
+    Assert.assertNull("nothing must be sent to the token_url", attackerBody.get());
+  }
+
+  /**
+   * Variant of the attack where the token exchange goes to Google but the resulting access token is sent to the
+   * attacker through the impersonation endpoint.
+   */
+  @Test
+  public void testImpersonationUrlExfiltrationIsRejected() {
+    String json = externalAccount(STS_URL, attackerUrl(), urlSource(metadataUrl()));
+
+    assertRejected(json, "service_account_impersonation_url");
+    Assert.assertEquals(0, metadataHits.get());
+    Assert.assertNull(attackerBody.get());
+  }
+
+  /**
+   * The auth library parser is lenient and ignores content after the top level object. Validation must see the
+   * same document as the library, so such content must not cause the configuration to escape validation.
+   */
+  @Test
+  public void testTrailingContentDoesNotBypassValidation() {
+    String payload = externalAccount(attackerUrl(), null, urlSource(metadataUrl()));
+    for (String suffix : new String[] { " garbage", "\n{}", "]", "\u0000" }) {
+      assertRejected(payload + suffix, "token_url");
+    }
+    Assert.assertEquals(0, metadataHits.get());
+    Assert.assertNull(attackerBody.get());
+  }
+
+  @Test
+  public void testOtherContentIsHandledByLibrary() {
+    // behaviour for content that is not an external account configuration is unchanged, including content that
+    // the library's lenient parser accepts
+    for (String json : new String[] { "not json", "null", "[]", "{\"project_id\":\"no-type\"}",
+      "{\"type\":\"authorized_user\"}", "{\"type\":\"authorized_user\"} trailing", "{\"type\":\"service_account\"}]",
+      "{\"type\":\"service_account\",\"type\":\"authorized_user\"}" }) {
+      Exception expected = loadWithLibrary(json);
+      Exception actual = null;
+      try {
+        GCPUtils.loadServiceAccountCredentials(json, false);
+      } catch (Exception e) {
+        actual = e;
+      }
+      Assert.assertNotNull(json, expected);
+      Assert.assertNotNull(json, actual);
+      Assert.assertEquals(json, expected.getClass(), actual.getClass());
+      Assert.assertEquals(json, expected.getMessage(), actual.getMessage());
+    }
+  }
+
+  private String metadataUrl() {
+    return "http://127.0.0.1:" + metadataServer.getAddress().getPort()
+      + "/computeMetadata/v1/instance/service-accounts/default/token";
+  }
+
+  private String attackerUrl() {
+    return "http://127.0.0.1:" + attackerServer.getAddress().getPort() + "/token-exchange";
+  }
+
+  private static Exception loadWithLibrary(String json) {
+    try {
+      GoogleCredentials.fromStream(new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8)));
+      return null;
+    } catch (Exception e) {
+      return e;
+    }
+  }
+
+  private static String externalAccount(String tokenUrl, String impersonationUrl, String credentialSource) {
+    StringBuilder json = new StringBuilder("{\"type\":\"external_account\",")
+      .append("\"audience\":\"//iam.googleapis.com/projects/1/locations/global/workloadIdentityPools/p/providers/x\",")
+      .append("\"subject_token_type\":\"urn:ietf:params:oauth:token-type:jwt\",");
+    if (tokenUrl != null) {
+      json.append("\"token_url\":\"").append(tokenUrl).append("\",");
+    }
+    if (impersonationUrl != null) {
+      json.append("\"service_account_impersonation_url\":\"").append(impersonationUrl).append("\",");
+    }
+    return json.append("\"credential_source\":").append(credentialSource).append("}").toString();
+  }
+
+  private static String urlSource(String url) {
+    return "{\"url\":\"" + url + "\",\"headers\":{\"Metadata-Flavor\":\"Google\"},"
+      + "\"format\":{\"type\":\"json\",\"subject_token_field_name\":\"access_token\"}}";
+  }
+
+  private static void assertRejected(String json, String expectedField) {
+    try {
+      GCPUtils.loadServiceAccountCredentials(json, false);
+      Assert.fail("Expected rejection of " + json);
+    } catch (IOException e) {
+      Assert.assertTrue(e.getMessage(), e.getMessage().startsWith("Invalid external account credentials"));
+      Assert.assertTrue(e.getMessage(), e.getMessage().contains("'" + expectedField + "'"));
+    }
+  }
+}