Merge branch 'main' into owl-bot-update-lock-04c35dc5f49f0f503a306397d6d043685f8d2bb822ab515818c4208d7fb2db3a

Author: parthea

CHANGELOG.md

@@ -4,6 +4,18 @@
 
 [1]: https://pypi.org/project/google-auth/#history
 
+## [2.38.0](https://github.com/googleapis/google-auth-library-python/compare/v2.37.0...v2.38.0) (2025-01-23)
+
+
+### Features
+
+* Adding domain-wide delegation flow in impersonated credential ([#1624](https://github.com/googleapis/google-auth-library-python/issues/1624)) ([34ee3fe](https://github.com/googleapis/google-auth-library-python/commit/34ee3fef8cba6a1bbaa46fa16b43af0d89b60b0f))
+
+
+### Documentation
+
+* Add warnings regarding consuming externally sourced credentials ([d049370](https://github.com/googleapis/google-auth-library-python/commit/d049370d266b50db0e09d7b292dbf33052b27853))
+
 ## [2.37.0](https://github.com/googleapis/google-auth-library-python/compare/v2.36.1...v2.37.0) (2024-12-11)
 
 

docs/user-guide.rst

@@ -29,6 +29,17 @@ that supports OpenID Connect (OIDC).
 Obtaining credentials
 ---------------------
 
+.. warning::
+    Important: If you accept a credential configuration (credential JSON/File/Stream)
+    from an external source for authentication to Google Cloud Platform, you must
+    validate it before providing it to any Google API or client library. Providing an
+    unvalidated credential configuration to Google APIs or libraries can compromise
+    the security of your systems and data. For more information, refer to
+    `Validate credential configurations from external sources`_.
+
+.. _Validate credential configurations from external sources:
+    https://cloud.google.com/docs/authentication/external/externally-sourced-credentials
+
 .. _application-default:
 
 Application default credentials

google/auth/_default.py

@@ -85,6 +85,17 @@ def load_credentials_from_file(
     user credentials, external account credentials, or impersonated service
     account credentials.
 
+    .. warning::
+        Important: If you accept a credential configuration (credential JSON/File/Stream)
+        from an external source for authentication to Google Cloud Platform, you must
+        validate it before providing it to any Google API or client library. Providing an
+        unvalidated credential configuration to Google APIs or libraries can compromise
+        the security of your systems and data. For more information, refer to
+        `Validate credential configurations from external sources`_.
+
+        .. _Validate credential configurations from external sources:
+            https://cloud.google.com/docs/authentication/external/externally-sourced-credentials
+
     Args:
         filename (str): The full path to the credentials file.
         scopes (Optional[Sequence[str]]): The list of scopes for the credentials. If
@@ -137,6 +148,17 @@ def load_credentials_from_dict(
     user credentials, external account credentials, or impersonated service
     account credentials.
 
+    .. warning::
+        Important: If you accept a credential configuration (credential JSON/File/Stream)
+        from an external source for authentication to Google Cloud Platform, you must
+        validate it before providing it to any Google API or client library. Providing an
+        unvalidated credential configuration to Google APIs or libraries can compromise
+        the security of your systems and data. For more information, refer to
+        `Validate credential configurations from external sources`_.
+
+    .. _Validate credential configurations from external sources:
+        https://cloud.google.com/docs/authentication/external/externally-sourced-credentials
+
     Args:
         info (Dict[str, Any]): A dict object containing the credentials
         scopes (Optional[Sequence[str]]): The list of scopes for the credentials. If

google/auth/compute_engine/_metadata.py

@@ -201,7 +201,7 @@ def get(
     url = _helpers.update_query(base_url, query_params)
 
     backoff = ExponentialBackoff(total_attempts=retry_count)
-
+    failure_reason = None
     for attempt in backoff:
         try:
             response = request(url=url, method="GET", headers=headers_to_use)
@@ -213,6 +213,11 @@ def get(
                     retry_count,
                     response.status,
                 )
+                failure_reason = (
+                    response.data.decode("utf-8")
+                    if hasattr(response.data, "decode")
+                    else response.data
+                )
                 continue
             else:
                 break
@@ -225,10 +230,13 @@ def get(
                 retry_count,
                 e,
             )
+            failure_reason = e
     else:
         raise exceptions.TransportError(
             "Failed to retrieve {} from the Google Compute Engine "
-            "metadata service. Compute Engine Metadata server unavailable".format(url)
+            "metadata service. Compute Engine Metadata server unavailable due to {}".format(
+                url, failure_reason
+            )
         )
 
     content = _helpers.from_bytes(response.data)

google/auth/iam.py

@@ -48,6 +48,11 @@
     + "/serviceAccounts/{}:signBlob"
 )
 
+_IAM_SIGNJWT_ENDPOINT = (
+    "https://iamcredentials.googleapis.com/v1/projects/-"
+    + "/serviceAccounts/{}:signJwt"
+)
+
 _IAM_IDTOKEN_ENDPOINT = (
     "https://iamcredentials.googleapis.com/v1/"
     + "projects/-/serviceAccounts/{}:generateIdToken"

google/auth/impersonated_credentials.py

@@ -38,12 +38,15 @@
 from google.auth import iam
 from google.auth import jwt
 from google.auth import metrics
+from google.oauth2 import _client
 
 
 _REFRESH_ERROR = "Unable to acquire impersonated credentials"
 
 _DEFAULT_TOKEN_LIFETIME_SECS = 3600  # 1 hour in seconds
 
+_GOOGLE_OAUTH2_TOKEN_ENDPOINT = "https://oauth2.googleapis.com/token"
+
 
 def _make_iam_token_request(
     request,
@@ -177,6 +180,7 @@ def __init__(
         target_principal,
         target_scopes,
         delegates=None,
+        subject=None,
         lifetime=_DEFAULT_TOKEN_LIFETIME_SECS,
         quota_project_id=None,
         iam_endpoint_override=None,
@@ -204,9 +208,12 @@ def __init__(
             quota_project_id (Optional[str]): The project ID used for quota and billing.
                 This project may be different from the project used to
                 create the credentials.
-            iam_endpoint_override (Optiona[str]): The full IAM endpoint override
+            iam_endpoint_override (Optional[str]): The full IAM endpoint override
                 with the target_principal embedded. This is useful when supporting
                 impersonation with regional endpoints.
+            subject (Optional[str]): sub field of a JWT. This field should only be set
+                if you wish to impersonate as a user. This feature is useful when
+                using domain wide delegation.
         """
 
         super(Credentials, self).__init__()
@@ -231,6 +238,7 @@ def __init__(
         self._target_principal = target_principal
         self._target_scopes = target_scopes
         self._delegates = delegates
+        self._subject = subject
         self._lifetime = lifetime or _DEFAULT_TOKEN_LIFETIME_SECS
         self.token = None
         self.expiry = _helpers.utcnow()
@@ -275,6 +283,39 @@ def _update_token(self, request):
         # Apply the source credentials authentication info.
         self._source_credentials.apply(headers)
 
+        #  If a subject is specified a domain-wide delegation auth-flow is initiated
+        #  to impersonate as the provided subject (user).
+        if self._subject:
+            if self.universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN:
+                raise exceptions.GoogleAuthError(
+                    "Domain-wide delegation is not supported in universes other "
+                    + "than googleapis.com"
+                )
+
+            now = _helpers.utcnow()
+            payload = {
+                "iss": self._target_principal,
+                "scope": _helpers.scopes_to_string(self._target_scopes or ()),
+                "sub": self._subject,
+                "aud": _GOOGLE_OAUTH2_TOKEN_ENDPOINT,
+                "iat": _helpers.datetime_to_secs(now),
+                "exp": _helpers.datetime_to_secs(now) + _DEFAULT_TOKEN_LIFETIME_SECS,
+            }
+
+            assertion = _sign_jwt_request(
+                request=request,
+                principal=self._target_principal,
+                headers=headers,
+                payload=payload,
+                delegates=self._delegates,
+            )
+
+            self.token, self.expiry, _ = _client.jwt_grant(
+                request, _GOOGLE_OAUTH2_TOKEN_ENDPOINT, assertion
+            )
+
+            return
+
         self.token, self.expiry = _make_iam_token_request(
             request=request,
             principal=self._target_principal,
@@ -478,3 +519,61 @@ def refresh(self, request):
         self.expiry = datetime.utcfromtimestamp(
             jwt.decode(id_token, verify=False)["exp"]
         )
+
+
+def _sign_jwt_request(request, principal, headers, payload, delegates=[]):
+    """Makes a request to the Google Cloud IAM service to sign a JWT using a
+    service account's system-managed private key.
+    Args:
+        request (Request): The Request object to use.
+        principal (str): The principal to request an access token for.
+        headers (Mapping[str, str]): Map of headers to transmit.
+        payload (Mapping[str, str]): The JWT payload to sign. Must be a
+            serialized JSON object that contains a JWT Claims Set.
+        delegates (Sequence[str]): The chained list of delegates required
+            to grant the final access_token.  If set, the sequence of
+            identities must have "Service Account Token Creator" capability
+            granted to the prceeding identity.  For example, if set to
+            [serviceAccountB, serviceAccountC], the source_credential
+            must have the Token Creator role on serviceAccountB.
+            serviceAccountB must have the Token Creator on
+            serviceAccountC.
+            Finally, C must have Token Creator on target_principal.
+            If left unset, source_credential must have that role on
+            target_principal.
+
+    Raises:
+        google.auth.exceptions.TransportError: Raised if there is an underlying
+            HTTP connection error
+        google.auth.exceptions.RefreshError: Raised if the impersonated
+            credentials are not available.  Common reasons are
+            `iamcredentials.googleapis.com` is not enabled or the
+            `Service Account Token Creator` is not assigned
+    """
+    iam_endpoint = iam._IAM_SIGNJWT_ENDPOINT.format(principal)
+
+    body = {"delegates": delegates, "payload": json.dumps(payload)}
+    body = json.dumps(body).encode("utf-8")
+
+    response = request(url=iam_endpoint, method="POST", headers=headers, body=body)
+
+    # support both string and bytes type response.data
+    response_body = (
+        response.data.decode("utf-8")
+        if hasattr(response.data, "decode")
+        else response.data
+    )
+
+    if response.status != http_client.OK:
+        raise exceptions.RefreshError(_REFRESH_ERROR, response_body)
+
+    try:
+        jwt_response = json.loads(response_body)
+        signed_jwt = jwt_response["signedJwt"]
+        return signed_jwt
+
+    except (KeyError, ValueError) as caught_exc:
+        new_exc = exceptions.RefreshError(
+            "{}: No signed JWT in response.".format(_REFRESH_ERROR), response_body
+        )
+        raise new_exc from caught_exc

google/auth/version.py

@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version__ = "2.37.0"
+__version__ = "2.38.0"

system_tests/secrets.tar.enc

@@ -344,12 +344,32 @@ def test_get_return_none_for_not_found_error():
 @mock.patch("time.sleep", return_value=None)
 def test_get_failure_connection_failed(mock_sleep):
     request = make_request("")
-    request.side_effect = exceptions.TransportError()
+    request.side_effect = exceptions.TransportError("failure message")
 
     with pytest.raises(exceptions.TransportError) as excinfo:
         _metadata.get(request, PATH)
 
-    assert excinfo.match(r"Compute Engine Metadata server unavailable")
+    assert excinfo.match(
+        r"Compute Engine Metadata server unavailable due to failure message"
+    )
+
+    request.assert_called_with(
+        method="GET",
+        url=_metadata._METADATA_ROOT + PATH,
+        headers=_metadata._METADATA_HEADERS,
+    )
+    assert request.call_count == 5
+
+
+def test_get_too_many_requests_retryable_error_failure():
+    request = make_request("too many requests", status=http_client.TOO_MANY_REQUESTS)
+
+    with pytest.raises(exceptions.TransportError) as excinfo:
+        _metadata.get(request, PATH)
+
+    assert excinfo.match(
+        r"Compute Engine Metadata server unavailable due to too many requests"
+    )
 
     request.assert_called_with(
         method="GET",