google-cloud-storage: Cannot create signed url with ImpersonatedCredentials

[salrashid123]

impersonated_credentials cannot create signedURLs for google-cloud-storage since it does not require or have the impersonated accounts private key/json file and does not implement credentials.Signing

that is

import google.auth

from google.cloud import storage
from google.oauth2 import service_account
from google.auth import impersonated_credentials
import datetime

from pytz import UTC

svc_account_file = '/path/to/svc.json'
project = 'fabled-ray-104117'
target_scopes = ['https://www.googleapis.com/auth/devstorage.read_only', 'https://www.googleapis.com/auth/cloud-platform']
source_credentials = service_account.Credentials.from_service_account_file(
    svc_account_file,
    scopes=target_scopes)

target_credentials = impersonated_credentials.Credentials(
    source_credentials = source_credentials,
    target_principal='impersonated-account@fabled-ray-104117.iam.gserviceaccount.com',
    target_scopes = target_scopes,
    delegates=[],
    lifetime=300)
client = storage.Client(credentials=target_credentials,project=project)
bucket = client.get_bucket('fabled-ray-104117')
blob = bucket.get_blob('signed_url_file.txt')

delta = datetime.timedelta(seconds=60)
expiration = datetime.datetime.utcnow().replace(tzinfo=UTC) + delta

s = blob.generate_signed_url(expiration=expiration, method="GET", version="v4")
print s

yields

Traceback (most recent call last):
  File "main.py", line 43, in <module>
    s = blob.generate_signed_url(expiration=300, method="GET", version="v4")
  File "env/local/lib/python2.7/site-packages/google/cloud/storage/blob.py", line 444, in generate_signed_url
    query_parameters=query_parameters,
  File "/env/local/lib/python2.7/site-packages/google/cloud/storage/_signing.py", line 503, in generate_signed_url_v4
    ensure_signed_credentials(credentials)
  File "env/local/lib/python2.7/site-packages/google/cloud/storage/_signing.py", line 54, in ensure_signed_credentials
    "details." % (type(credentials), auth_uri)
AttributeError: you need a private key to sign credentials.the credentials you are currently using <class 'google.auth.impersonated_credentials.Credentials'> just contains a token. see https://google-cloud-python.readthedocs.io/en/latest/core/auth.html?highlight=authentication#setting-up-a-service-account for more details.

Potential solution is to use iamcredentials api once again to 'remotely sign' as in here:
see:
googleapis/google-cloud-java#5043

--

Which means iamcredentials would now look like

class Credentials(credentials.Credentials, credentials.Signing):

i made a working implementation here:
https://gist.github.com/salrashid123/9e3fb4ac87cfa7bbd8b4f6a902aecd00

    def sign_bytes(self, message):
        
        iam_sign_endpoint = _IAM_SIGN_ENDPOINT.format(self._target_principal)

        body = {
            "payload": base64.b64encode(message),
            "delegates": self._delegates
        }

        headers = {
            'Content-Type': 'application/json',
        }

        authed_session = AuthorizedSession(self._source_credentials)
        
        response = authed_session.post(
            url=iam_sign_endpoint,
            headers=headers,
            data=json.dumps(body))

        return base64.b64decode(response.json()['signedBlob'])

    @property
    def signer_email(self):
        return self._target_principal   

    @property
    def signer(self):
        raise NotImplementedError('Signer must be implemented.')

[salrashid123]

I just realized there's already an iam signer which could also be used to sign

here is the gist for impersonated_credentials.py using the iamsigner (see lines 236+)

• https://gist.github.com/salrashid123/f10549819cff103edc358f3eae657029

however, the current impersonated_credentials.py uses the different api than the iam signer (i.,e

• iamcredentials
  https://cloud.google.com/iam/credentials/reference/rest/v1/projects.serviceAccounts/signBlob
• iam https://cloud.google.com/iam/reference/rest/v1/projects.serviceAccounts/signBlob

The former is preferred way to sign. I would suggest either

1. update google.auth.iam to use iamcredentials and then implement the google.auth.crypt.base.Signer into impersonated_credentials which inturn would use the new iamsigner
2. Directly use iamcredentials in the impersonated_credentials.py module for now as showed in the previous comment

(1) is better long term but i i'm not sure of the ramifications of 'just replacing' the underlying api call..

[salrashid123]

heres' an example to do impersonated_credentials sign_bytes()

• https://gist.github.com/salrashid123/e047031e78b14dd733e46d358f42c229#file-impersonated_credentials-py-L243

and also to get a GoogleIDToken:

• https://gist.github.com/salrashid123/e047031e78b14dd733e46d358f42c229#file-impersonated_credentials-py-L285

the usage would be something like

source_credentials = service_account.Credentials.from_service_account_file(
    'cert.json')
target_scopes = ['https://www.googleapis.com/auth/cloud-platform']
        
target_credentials = impersonated_credentials.Credentials(
    source_credentials = source_credentials,
    target_principal='impersonated-account@project.iam.gserviceaccount.com',
    target_scopes = target_scopes,
    delegates=[],
    lifetime=300)


# signer anything you want as the impersonated credentials
b = target_credentials.sign_bytes('badff')
print base64.b64encode(b)


storage_client = storage.Client('fabled-ray-104117', target_credentials)
data_bucket = storage_client.lookup_bucket('fabled-ray-104117')
signed_blob_path = data_bucket.blob("FILENAME")
expires_at_ms = datetime.now() + timedelta(minutes=30)
signed_url = signed_blob_path.generate_signed_url(expires_at_ms, credentials=target_credentials, version="v4")

print signed_url

# =====================   IDToken

target_audience = 'https://myapp-6w42z6vi3q-uc.a.run.app'  

id_creds = impersonated_credentials.IDTokenCredentials(
    target_credentials, target_audience=target_audience)

i've got the code ready but finding some difficulty getting the tests done...i'm also using AuthorzedSession() internally in code (see gist above)...and i'm not familar with how to construct a test case to mock a requests() object and the AuthorizedSession

[maroux]

I believe this works now except for this bug on Python3:

app_1  |   File "/usr/local/lib/python3.7/site-packages/google/cloud/storage/_signing.py", line 619, in generate_signed_url_v4
app_1  |     signature_bytes = credentials.sign_bytes(string_to_sign.encode("ascii"))
app_1  |   File "/usr/local/lib/python3.7/site-packages/google/auth/impersonated_credentials.py", line 269, in sign_bytes
app_1  |     url=iam_sign_endpoint, headers=headers, json=body
app_1  |   File "/usr/local/lib/python3.7/site-packages/requests/sessions.py", line 578, in post
app_1  |     return self.request('POST', url, data=data, json=json, **kwargs)
app_1  |   File "/usr/local/lib/python3.7/site-packages/google/auth/transport/requests.py", line 452, in request
app_1  |     **kwargs
app_1  |   File "/usr/local/lib/python3.7/site-packages/requests/sessions.py", line 516, in request
app_1  |     prep = self.prepare_request(req)
app_1  |   File "/usr/local/lib/python3.7/site-packages/requests/sessions.py", line 459, in prepare_request
app_1  |     hooks=merge_hooks(request.hooks, self.hooks),
app_1  |   File "/usr/local/lib/python3.7/site-packages/requests/models.py", line 317, in prepare
app_1  |     self.prepare_body(data, files, json)
app_1  |   File "/usr/local/lib/python3.7/site-packages/requests/models.py", line 467, in prepare_body
app_1  |     body = complexjson.dumps(json)
app_1  |   File "/usr/local/lib/python3.7/json/__init__.py", line 231, in dumps
app_1  |     return _default_encoder.encode(obj)
app_1  |   File "/usr/local/lib/python3.7/json/encoder.py", line 199, in encode
app_1  |     chunks = self.iterencode(o, _one_shot=True)
app_1  |   File "/usr/local/lib/python3.7/json/encoder.py", line 257, in iterencode
app_1  |     return _iterencode(o, 0)
app_1  |   File "/usr/local/lib/python3.7/json/encoder.py", line 179, in default
app_1  |     raise TypeError(f'Object of type {o.__class__.__name__} '
app_1  | TypeError: Object of type bytes is not JSON serializable

The fix, of course, is to decode the bytes returned by base64.b64encode:

- body = {"payload": base64.b64encode(message), "delegates": self._delegates}
+ body = {"payload": base64.b64encode(message).decode(), "delegates": self._delegates}