iam.py 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114
  1. # Copyright 2017 Google LLC
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. """Tools for using the Google `Cloud Identity and Access Management (IAM)
  15. API`_'s auth-related functionality.
  16. .. _Cloud Identity and Access Management (IAM) API:
  17. https://cloud.google.com/iam/docs/
  18. """
  19. import base64
  20. import http.client as http_client
  21. import json
  22. from google.auth import _helpers
  23. from google.auth import crypt
  24. from google.auth import exceptions
  25. _IAM_SCOPE = ["https://www.googleapis.com/auth/iam"]
  26. _IAM_ENDPOINT = (
  27. "https://iamcredentials.googleapis.com/v1/projects/-"
  28. + "/serviceAccounts/{}:generateAccessToken"
  29. )
  30. _IAM_SIGN_ENDPOINT = (
  31. "https://iamcredentials.googleapis.com/v1/projects/-"
  32. + "/serviceAccounts/{}:signBlob"
  33. )
  34. _IAM_IDTOKEN_ENDPOINT = (
  35. "https://iamcredentials.googleapis.com/v1/"
  36. + "projects/-/serviceAccounts/{}:generateIdToken"
  37. )
  38. class Signer(crypt.Signer):
  39. """Signs messages using the IAM `signBlob API`_.
  40. This is useful when you need to sign bytes but do not have access to the
  41. credential's private key file.
  42. .. _signBlob API:
  43. https://cloud.google.com/iam/reference/rest/v1/projects.serviceAccounts
  44. /signBlob
  45. """
  46. def __init__(self, request, credentials, service_account_email):
  47. """
  48. Args:
  49. request (google.auth.transport.Request): The object used to make
  50. HTTP requests.
  51. credentials (google.auth.credentials.Credentials): The credentials
  52. that will be used to authenticate the request to the IAM API.
  53. The credentials must have of one the following scopes:
  54. - https://www.googleapis.com/auth/iam
  55. - https://www.googleapis.com/auth/cloud-platform
  56. service_account_email (str): The service account email identifying
  57. which service account to use to sign bytes. Often, this can
  58. be the same as the service account email in the given
  59. credentials.
  60. """
  61. self._request = request
  62. self._credentials = credentials
  63. self._service_account_email = service_account_email
  64. def _make_signing_request(self, message):
  65. """Makes a request to the API signBlob API."""
  66. message = _helpers.to_bytes(message)
  67. method = "POST"
  68. url = _IAM_SIGN_ENDPOINT.format(self._service_account_email)
  69. headers = {"Content-Type": "application/json"}
  70. body = json.dumps(
  71. {"payload": base64.b64encode(message).decode("utf-8")}
  72. ).encode("utf-8")
  73. self._credentials.before_request(self._request, method, url, headers)
  74. response = self._request(url=url, method=method, body=body, headers=headers)
  75. if response.status != http_client.OK:
  76. raise exceptions.TransportError(
  77. "Error calling the IAM signBlob API: {}".format(response.data)
  78. )
  79. return json.loads(response.data.decode("utf-8"))
  80. @property
  81. def key_id(self):
  82. """Optional[str]: The key ID used to identify this private key.
  83. .. warning::
  84. This is always ``None``. The key ID used by IAM can not
  85. be reliably determined ahead of time.
  86. """
  87. return None
  88. @_helpers.copy_docstring(crypt.Signer)
  89. def sign(self, message):
  90. response = self._make_signing_request(message)
  91. return base64.b64decode(response["signedBlob"])