Google Cloud Function authentication. Obtain Identity token authorization bearer header curl

1
votes

Setting up a pub/sub based on a cron style deployment to call a google function that will check for new data and then push it through a pipeline. Part of this pipeline requires submitting a curl call with an authorization header that takes an identity token. I have not found a good way of generating this identity token.

I currently have tried changing the owner of the cloud function to a service account that has permissions across storage/data-labeling/cloud functions and I have also used a stored credential file (i.e. access.json) with a private key. I have an environment variable set (GOOGLE_APPLICATION_CREDENTIALS) that points to this private key and attempt to pull an identity token within the google cloud function via $(gcloud auth application-default print-access-token) - this returns an empty string with no error. 

# I have tried something very similar to this

command = "echo $(gcloud auth application-default print-access-token)"
p = subprocess.Popen(command, shell=True, 
                   stdout=subprocess.PIPE,
                   stderr=subprocess.PIPE)
p.wait()
out = p.communicate()
print("OUT_CODE: ", out)

I simply want to submit this curl command with a properly obtained token.

command = "GOOGLE_APPLICATION_CREDENTIALS=/user_code/dl_access.json bash -c 'gcloud auth activate-service-account --key-file=/user_code/dl_access.json; echo $(gcloud auth application-default print-access-token)'"
p = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE)
p.wait()
out, err = p.communicate()
auth = out.decode().rstrip()
print("OUT_CODE: ", out, err)
command = "curl -X POST "
command += '-H "Authorization: Bearer $(gcloud auth application-default print-access-token)" '
command += '-H "Content-Type: application/json" '
command += 'https://datalabeling.googleapis.com/v1beta1/projects/'
command += '{}/datasets/{}/image:label '.format(PROJECT_ID, dataset.name.split("/")[-1])
command += "-d '{"
command += '"basicConfig": {'
command += '"labelGroup": "{}", '.format("test_label_group")
command += '"instruction": "{}", '.format("projects/cv/instructions/5cd5da11_0sdfgsdfgsdfg2c0b8eb8")
command += '"annotatedDatasetDisplayName": "{}", '.format(dataset.display_name)
command += '"replica_count": 3 '
command += '}, '
command += '"feature": "BOUNDING_BOX", '
command += '"boundingPolyConfig": { '
command += '"annotationSpecSet": "{}", '.format(
    "projects/cv/annotationSpecSets/_b3b1_005csdfgc6_0000_297e1a11439bdc")
command += '}, '
command += "}' "

print(command)
p = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE)
p.wait()
out, err = p.communicate()
print("out:", out)
print("err:", err)

The above fails due to the Authorization: Bearer <ID_Token> being an empty string for ID_Token.

2
python-3.xgoogle-cloud-platformgoogle-cloud-functionsgoogle-oauth

2 Answers

1
votes

Your cloud function is sandboxed, you can't perform system calls. Remember, you are in serverless mode, you don't know what issue the underlayer server: what is its own? Are gcloud and curl installed? Which version?....

Thus, you have to write code and make Python call. Checkout the libraries for data labeling. You can also get inspiration from function to function code for inspiration if you prefer calling directly the API.

1
votes

Do not use shell scripts, external commands, etc in Cloud Functions.

Below is an example of how to obtain an OAuth 2.0 Identity Token when running in Cloud Functions. In your real code, you will need to change the "audience" value to whatever the service you are calling requires. If you call this function, it will display "SUCCESS" or "FAILURE" in your browser. This code also demonstrates how to use the Python requests library to make HTTP requests. Use this library instead of trying to execute the program CURL.

import requests
import json

def entry(request):
    id_token = requestIdentityToken('http://www.example.com')

    if id_token is not None:
        print('ID Token', id_token)
        return f'SUCCESS'
    else:
        return f'FAILURE'

def requestIdentityToken(audience=None):
        host = 'http://metadata.google.internal'
        header = {'Metadata-Flavor': 'Google'}

        if audience is None:
                audience = 'http://example.com'

        url = '{}/computeMetadata/v1/instance/service-accounts/default/identity?audience={}'.format(host, audience)

        try:
                r = requests.get(url=url, headers=header)

                if r.status_code < 200 or r.status_code >= 300:
                        print('Error:', r.reason)
                        return None

                return r.text
        except Exception as e:
                print(str(e))
                return None

Example command to deploy this function:

gcloud functions deploy requestIdentityToken --runtime python37 --trigger-http --entry-point entry

This command will "print" the Identity Token which you will find in the Stackdriver logs for this function.

Additional Information: