feat: add Google Cloud OIDC detector

cloudsmith-iduffy · claude · cloudsmith-iduffy · commit c80922f1bb10 · 2026-06-09T00:00:39.000+01:00
Adds a Google Cloud OIDC environment detector to the credential
auto-discovery chain, mirroring the existing AWS detector. When
CLOUDSMITH_ORG and CLOUDSMITH_SERVICE_SLUG are set, the CLI obtains a
Google-signed ID token from the ambient GCP identity via the google-auth
SDK and exchanges it for a short-lived Cloudsmith token.

get_token() tries three strategies in order:
1. Metadata server (GCE/Cloud Run/GKE/Cloud Functions/App Engine/Cloud
   Build) via a direct request with format=full, required for the
   email/email_verified claims Cloudsmith pins (the SDK metadata path
   cannot request format=full).
2. Service-account key file (GOOGLE_APPLICATION_CREDENTIALS) via
   google-auth.
3. Application Default Credentials (e.g. local
   `gcloud auth application-default login`) via google-auth.

detect() activates only when google-auth is importable and either ADC
resolves or the metadata server is reachable; it never raises into the
chain, and environment probing is side-effect-free (no throwaway token).
google-auth is an optional dependency under a new [gcp] extra.

Co-Authored-By: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/cloudsmith_cli/core/credentials/oidc/detectors/__init__.py b/cloudsmith_cli/core/credentials/oidc/detectors/__init__.py
@@ -7,6 +7,7 @@
 
 from .aws import AWSDetector
 from .base import EnvironmentDetector
+from .gcp import GCPDetector
 
 if TYPE_CHECKING:
     from ... import CredentialContext
@@ -15,6 +16,7 @@
 
 _DETECTORS: list[type[EnvironmentDetector]] = [
     AWSDetector,
+    GCPDetector,
 ]
 
 
diff --git a/cloudsmith_cli/core/credentials/oidc/detectors/gcp.py b/cloudsmith_cli/core/credentials/oidc/detectors/gcp.py
@@ -0,0 +1,144 @@
+"""Google Cloud OIDC detector.
+
+Uses the google-auth SDK to auto-discover Google Cloud credentials and obtain a
+signed ID token (JWT) for Cloudsmith. get_token() tries three strategies in
+order: the metadata server (GCE/Cloud Run/GKE/Cloud Functions/App Engine/Cloud
+Build), a service-account key file, and local Application Default Credentials.
+
+Requires google-auth (optional dependency): pip install cloudsmith-cli[gcp]
+
+References:
+    https://cloudsmith.com/blog/authenticate-to-cloudsmith-with-your-google-cloud-identity
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+
+import requests
+
+from .base import EnvironmentDetector
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_AUDIENCE = "cloudsmith"
+
+METADATA_IDENTITY_URL = (
+    "http://metadata.google.internal/computeMetadata/v1/"
+    "instance/service-accounts/default/identity"
+)
+METADATA_PROBE_URL = (
+    "http://metadata.google.internal/computeMetadata/v1/"
+    "instance/service-accounts/default/"
+)
+METADATA_FLAVOR_HEADER = {"Metadata-Flavor": "Google"}
+METADATA_TIMEOUT = 5
+
+
+class GCPDetector(EnvironmentDetector):
+    """Detects Google Cloud environments and obtains an OIDC ID token."""
+
+    name = "Google Cloud"
+
+    def detect(self) -> bool:
+        try:
+            import google.auth
+            from google.auth.exceptions import DefaultCredentialsError
+        except ImportError:
+            logger.debug("GCPDetector: google-auth not installed, skipping")
+            return False
+
+        try:
+            try:
+                google.auth.default()
+                return True
+            except DefaultCredentialsError:
+                return self._metadata_available()
+        except Exception:  # pylint: disable=broad-exception-caught
+            logger.debug(
+                "GCPDetector: unexpected error during detection", exc_info=True
+            )
+            return False
+
+    def _metadata_available(self) -> bool:
+        try:
+            response = requests.get(
+                METADATA_PROBE_URL,
+                headers=METADATA_FLAVOR_HEADER,
+                timeout=METADATA_TIMEOUT,
+            )
+            return response.status_code == 200
+        except requests.exceptions.RequestException:
+            return False
+
+    def get_token(self) -> str:
+        audience = self.context.oidc_audience or DEFAULT_AUDIENCE
+        for strategy in (
+            self._token_from_metadata,
+            self._token_from_service_account_file,
+            self._token_from_adc,
+        ):
+            token = strategy(audience)
+            if token:
+                return token
+        raise ValueError(
+            "Google Cloud detector could not obtain an OIDC token via the "
+            "metadata server, a service-account key file, or Application "
+            "Default Credentials. Set CLOUDSMITH_OIDC_TOKEN to provide a token "
+            "directly."
+        )
+
+    def _token_from_metadata(self, audience: str) -> str | None:
+        try:
+            response = requests.get(
+                METADATA_IDENTITY_URL,
+                headers=METADATA_FLAVOR_HEADER,
+                params={"audience": audience, "format": "full"},
+                timeout=METADATA_TIMEOUT,
+            )
+        except requests.exceptions.RequestException:
+            return None
+        if response.status_code != 200:
+            return None
+        return response.text.strip() or None
+
+    def _token_from_service_account_file(self, audience: str) -> str | None:
+        key_path = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
+        if not key_path:
+            return None
+        try:
+            from google.auth.transport.requests import (  # pylint: disable=import-error
+                Request,
+            )
+            from google.oauth2 import service_account  # pylint: disable=import-error
+
+            credentials = service_account.IDTokenCredentials.from_service_account_file(
+                key_path,
+                target_audience=audience,
+            )
+            credentials.refresh(Request())
+            return credentials.token or None
+        except Exception:  # pylint: disable=broad-exception-caught
+            logger.debug(
+                "GCPDetector: service-account key file token fetch failed",
+                exc_info=True,
+            )
+            return None
+
+    def _token_from_adc(self, audience: str) -> str | None:
+        # audience is unused here: ADC user credentials mint an id_token whose
+        # aud is gcloud's OAuth client id, not a custom audience. The signature
+        # stays uniform so get_token can call every strategy as strategy(audience).
+        try:
+            import google.auth  # pylint: disable=import-error
+            from google.auth.transport.requests import (  # pylint: disable=import-error
+                Request,
+            )
+
+            credentials, _ = google.auth.default()
+            credentials.refresh(Request())
+            return getattr(credentials, "id_token", None) or None
+        except Exception:  # pylint: disable=broad-exception-caught
+            logger.debug("GCPDetector: ADC id_token fetch failed", exc_info=True)
+            return None
diff --git a/cloudsmith_cli/core/tests/test_gcp_detector.py b/cloudsmith_cli/core/tests/test_gcp_detector.py
@@ -0,0 +1,171 @@
+"""Tests for the Google Cloud OIDC detector."""
+
+import sys
+from unittest import mock
+
+import pytest
+import requests
+
+from cloudsmith_cli.core.credentials.models import CredentialContext
+from cloudsmith_cli.core.credentials.oidc.detectors.gcp import GCPDetector
+
+GCP_MODULE = "cloudsmith_cli.core.credentials.oidc.detectors.gcp"
+
+
+def make_detector(**ctx):
+    return GCPDetector(context=CredentialContext(**ctx))
+
+
+class TestDetect:
+    def test_not_detected_when_google_auth_missing(self):
+        with mock.patch.dict(sys.modules, {"google.auth": None}):
+            assert make_detector().detect() is False
+
+    def test_detected_when_adc_resolves(self):
+        with mock.patch("google.auth.default", return_value=(mock.MagicMock(), "proj")):
+            assert make_detector().detect() is True
+
+    def test_detected_via_metadata_when_adc_unavailable(self):
+        from google.auth.exceptions import (  # pylint: disable=import-error
+            DefaultCredentialsError,
+        )
+
+        with mock.patch(
+            "google.auth.default", side_effect=DefaultCredentialsError("no adc")
+        ), mock.patch(f"{GCP_MODULE}.requests.get") as get:
+            get.return_value = mock.Mock(status_code=200)
+            assert make_detector().detect() is True
+
+    def test_not_detected_when_nothing_available(self):
+        from google.auth.exceptions import (  # pylint: disable=import-error
+            DefaultCredentialsError,
+        )
+
+        with mock.patch(
+            "google.auth.default", side_effect=DefaultCredentialsError("no adc")
+        ), mock.patch(f"{GCP_MODULE}.requests.get") as get:
+            get.side_effect = requests.exceptions.ConnectionError()
+            assert make_detector().detect() is False
+
+    def test_detect_never_raises(self):
+        with mock.patch("google.auth.default", side_effect=RuntimeError("boom")):
+            assert make_detector().detect() is False
+
+
+class TestGetTokenMetadata:
+    def test_returns_metadata_token_with_format_full(self):
+        with mock.patch(f"{GCP_MODULE}.requests.get") as get:
+            get.return_value = mock.Mock(status_code=200, text="meta-jwt\n")
+            token = make_detector().get_token()
+        assert token == "meta-jwt"
+        _, kwargs = get.call_args
+        assert kwargs["params"] == {"audience": "cloudsmith", "format": "full"}
+        assert kwargs["headers"] == {"Metadata-Flavor": "Google"}
+
+    def test_uses_configured_audience(self):
+        with mock.patch(f"{GCP_MODULE}.requests.get") as get:
+            get.return_value = mock.Mock(status_code=200, text="meta-jwt")
+            make_detector(oidc_audience="custom-aud").get_token()
+        _, kwargs = get.call_args
+        assert kwargs["params"]["audience"] == "custom-aud"
+
+    def test_metadata_skipped_when_non_200(self):
+        from google.auth.exceptions import (  # pylint: disable=import-error
+            DefaultCredentialsError,
+        )
+
+        with mock.patch(f"{GCP_MODULE}.requests.get") as get, mock.patch(
+            "google.auth.default", side_effect=DefaultCredentialsError("no adc")
+        ), mock.patch.dict("os.environ", {}, clear=True):
+            get.return_value = mock.Mock(status_code=404, text="")
+            with pytest.raises(ValueError):
+                make_detector().get_token()
+
+    def test_metadata_returns_none_on_request_exception(self):
+        with mock.patch(
+            f"{GCP_MODULE}.requests.get",
+            side_effect=requests.exceptions.ConnectionError(),
+        ):
+            assert (
+                make_detector()._token_from_metadata(  # pylint: disable=protected-access
+                    "cloudsmith"
+                )
+                is None
+            )
+
+    def test_metadata_returns_none_on_blank_body(self):
+        with mock.patch(f"{GCP_MODULE}.requests.get") as get:
+            get.return_value = mock.Mock(status_code=200, text="   ")
+            assert (
+                make_detector()._token_from_metadata(  # pylint: disable=protected-access
+                    "cloudsmith"
+                )
+                is None
+            )
+
+
+class TestGetTokenServiceAccountFile:
+    def test_returns_token_from_key_file(self):
+        from google.oauth2 import service_account  # pylint: disable=import-error
+
+        fake_creds = mock.Mock(token="sa-jwt")
+        with mock.patch(f"{GCP_MODULE}.requests.get") as get, mock.patch.dict(
+            "os.environ",
+            {"GOOGLE_APPLICATION_CREDENTIALS": "/tmp/key.json"},
+            clear=True,
+        ), mock.patch.object(
+            service_account.IDTokenCredentials,
+            "from_service_account_file",
+            return_value=fake_creds,
+        ) as from_file:
+            # Metadata unavailable so we fall through to the key-file strategy.
+            get.side_effect = requests.exceptions.ConnectionError()
+            token = make_detector().get_token()
+        assert token == "sa-jwt"
+        args, kwargs = from_file.call_args
+        assert args[0] == "/tmp/key.json"
+        assert kwargs["target_audience"] == "cloudsmith"
+        fake_creds.refresh.assert_called_once()
+
+    def test_skipped_when_env_unset(self):
+        from google.auth.exceptions import (  # pylint: disable=import-error
+            DefaultCredentialsError,
+        )
+
+        with mock.patch(f"{GCP_MODULE}.requests.get") as get, mock.patch(
+            "google.auth.default", side_effect=DefaultCredentialsError("no adc")
+        ), mock.patch.dict("os.environ", {}, clear=True):
+            get.side_effect = requests.exceptions.ConnectionError()
+            with pytest.raises(ValueError):
+                make_detector().get_token()
+
+
+class TestGetTokenADC:
+    def test_returns_id_token_from_adc(self):
+        fake_creds = mock.Mock(id_token="adc-jwt")
+        with mock.patch(f"{GCP_MODULE}.requests.get") as get, mock.patch.dict(
+            "os.environ", {}, clear=True
+        ), mock.patch("google.auth.default", return_value=(fake_creds, "proj")):
+            get.side_effect = requests.exceptions.ConnectionError()
+            token = make_detector().get_token()
+        assert token == "adc-jwt"
+        fake_creds.refresh.assert_called_once()
+
+    def test_raises_when_adc_has_no_id_token(self):
+        fake_creds = mock.Mock(spec=["refresh"])  # no id_token attribute
+        with mock.patch(f"{GCP_MODULE}.requests.get") as get, mock.patch.dict(
+            "os.environ", {}, clear=True
+        ), mock.patch("google.auth.default", return_value=(fake_creds, "proj")):
+            get.side_effect = requests.exceptions.ConnectionError()
+            with pytest.raises(ValueError):
+                make_detector().get_token()
+
+
+class TestRegistration:
+    def test_registered_after_aws(self):
+        from cloudsmith_cli.core.credentials.oidc.detectors import (
+            _DETECTORS,
+            AWSDetector,
+        )
+
+        assert _DETECTORS == [AWSDetector, GCPDetector]
diff --git a/setup.py b/setup.py
@@ -69,8 +69,12 @@ def get_long_description():
         "aws": [
             "boto3[crt]>=1.26.0",
         ],
+        "gcp": [
+            "google-auth>=2.0.0",
+        ],
         "all": [
             "boto3[crt]>=1.26.0",
+            "google-auth>=2.0.0",
         ],
     },
     entry_points={
