Source code for google.auth.iam

# Copyright 2017 Google LLC
# 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

"""Tools for using the Google `Cloud Identity and Access Management (IAM)
API`_'s auth-related functionality.

.. _Cloud Identity and Access Management (IAM) API:

import base64
import http.client as http_client
import json

from google.auth import _helpers
from google.auth import crypt
from google.auth import exceptions

_SIGN_BLOB_URI = _IAM_API_ROOT_URI + "/projects/-/serviceAccounts/{}:signBlob?alt=json"

[docs] class Signer(crypt.Signer): """Signs messages using the IAM `signBlob API`_. This is useful when you need to sign bytes but do not have access to the credential's private key file. .. _signBlob API: /signBlob """ def __init__(self, request, credentials, service_account_email): """ Args: request (google.auth.transport.Request): The object used to make HTTP requests. credentials (google.auth.credentials.Credentials): The credentials that will be used to authenticate the request to the IAM API. The credentials must have of one the following scopes: - - service_account_email (str): The service account email identifying which service account to use to sign bytes. Often, this can be the same as the service account email in the given credentials. """ self._request = request self._credentials = credentials self._service_account_email = service_account_email def _make_signing_request(self, message): """Makes a request to the API signBlob API.""" message = _helpers.to_bytes(message) method = "POST" url = _SIGN_BLOB_URI.format(self._service_account_email) headers = {"Content-Type": "application/json"} body = json.dumps( {"payload": base64.b64encode(message).decode("utf-8")} ).encode("utf-8") self._credentials.before_request(self._request, method, url, headers) response = self._request(url=url, method=method, body=body, headers=headers) if response.status != http_client.OK: raise exceptions.TransportError( "Error calling the IAM signBlob API: {}".format( ) return json.loads("utf-8")) @property def key_id(self): """Optional[str]: The key ID used to identify this private key. .. warning:: This is always ``None``. The key ID used by IAM can not be reliably determined ahead of time. """ return None
[docs] @_helpers.copy_docstring(crypt.Signer) def sign(self, message): response = self._make_signing_request(message) return base64.b64decode(response["signedBlob"])