Authenticated calls to cloud functions with Python

The past few weeks I developed and deployed a cloud function that is supposed to get called only by authorized users/service accounts and the truth is that the documentation I found wasn’t really helpful.

First I created a service account, gave it roles/cloudfunctions.invoker permission. If you’re dealing with an extracted service account the code is pretty simple

from google.oauth2 import service_account 
from google.auth.transport.requests import AuthorizedSession
keypath = '.../my_sa_key.json
'base_url='https://us-central1-my_project.cloudfunctions.net/my_function'
creds=service_account.IDTokenCredentials.from_service_account_file(      keypath, target_audience=base_url)
authed_session = AuthorizedSession(creds)
# make authenticated request and print the response, status_code
resp = authed_session.get(base_url)
print(resp.status_code)
print(resp.text)

Attempt 1

But in my case, the call to the cloud function is going to happen from within a google service (Dataflow) and I don’t have access to the service account file. So I tried to find how can I use target_audience with the default credentials. According to the documentation we need to generate a token and use it in the headers

import urllib
import google.auth.transport.requests
import google.oauth2.id_token

def make_authorized_get_request(service_url):
    """
    make_authorized_get_request makes a GET request to the specified HTTP endpoint
    in service_url (must be a complete URL) by authenticating with the
    ID token obtained from the google-auth client library.
    """
    req = urllib.request.Request(service_url)
    auth_req = google.auth.transport.requests.Request()
    id_token = google.oauth2.id_token.fetch_id_token(auth_req, service_url)
    req.add_header("Authorization", f"Bearer {id_token}")
    response = urllib.request.urlopen(req)
    return response.read()

Even though that seemed like a good idea since fetch_id_token will try to find the token from various sources. In my case, it didn’t work. Calling that from Dataflow gives megoogle.oauth2' has no attribute ‘id_token’

Attempt 2

So I searched more and another solution came from a post on StackOverflow.

from google.oauth2 import service_account
from google.auth.transport.requests import AuthorizedSession
base_url='https://us-central1-my_project.cloudfunctions.net/my_function
'IAM_SCOPE = 'https://www.googleapis.com/auth/iam'
OAUTH_TOKEN_URI = 'https://www.googleapis.com/oauth2/v4/token'
my_credentials, _ = google.auth.default(scopes=[IAM_SCOPE])
signer_email = my_credentials.service_account_email
signer = my_credentials.signercreds = google.oauth2.service_account.IDTokenCredentials(signer, signer_email, token_uri=OAUTH_TOKEN_URI, target_audience=base_url)
authed_session = AuthorizedSession(creds)
resp = authed_session.get(base_url)
print(resp.status_code)
print(resp.text)

That, too, worked when I ran it on my local dev, but deploying it to Dataflow I was getting more errors. Even though the my_credentials had the correct signer, email etc.. I was getting

"Error calling the IAM signBytes API: {}".format(response.data) google.auth.exceptions.TransportError: Error calling the IAM signBytes API: 
'{ 
  "error": { 
    "code": 403, 
    "message": "The caller does not have permission",
    "status": "PERMISSION_DENIED" 
  }
}'

Back to Google then…

Attempt 3 (Solution)

I accidentally found this article explaining how to authenticate from service to service on GCP. And even though I don’t like the idea, it worked!

Created a function to get the JWT token, I also used cache-tools to cache the token.

@cached(cache=TTLCache(maxsize=1024, ttl=3600 - 30))
def get_token(base_url):
    logging.info("Getting token id")
    # Set up metadata server request
    metadata_server_token_url = 'http://metadata/computeMetadata/v1/instance/service-accounts/default/identity?audience='

    token_request_url = metadata_server_token_url + base_url
    token_request_headers = {'Metadata-Flavor': 'Google'}

    # Fetch the token
    token_response = requests.get(token_request_url, headers=token_request_headers)
    jwt = token_response.content.decode("utf-8")
    return jwt

and use the token in the headers

headers = {
    'Authorization': f'Bearer {get_token(self.base_url)}',
    'Content-Type': 'application/fhir+json;charset=utf-8'
}

service_response = requests.put(base_url, headers=headers, json=data)
print(f"service_response_code = {service_response.status_code}")

Full Dataflow Code

@cached(cache=TTLCache(maxsize=1024, ttl=3600 - 30))
def get_token(base_url):
    metadata_server_token_url = 'http://metadata/computeMetadata/v1/instance/service-accounts/default/identity?audience='

    token_request_url = metadata_server_token_url + base_url
    token_request_headers = {'Metadata-Flavor': 'Google'}

    # Fetch the token
    token_response = requests.get(token_request_url, headers=token_request_headers)
    jwt = token_response.content.decode("utf-8")
    return jwt


class WriteToCloudFunction(beam.DoFn):

    def __init__(self):
        self.base_url = "https://us-central1-my_project.cloudfunctions.net/my_function"

    def process(self, message):

        data = json.loads(message.data.decode("utf-8"))

        headers = {
            'Authorization': f'Bearer {get_token(self.base_url)}',
            'Content-Type': 'application/fhir+json;charset=utf-8'
        }

        service_response = requests.put(self.base_url, headers=headers, json=data)
        logging.info(f"service_response_code = {service_response.status_code}")
with beam.Pipeline(options=pipeline_options) as pipeline:
  (
    pipeline
    | "Read PubSub Messages" >> beam.io.ReadFromPubSub(subscription=options.input_subscription,
with_attributes=True)
    | .....
    | "Write to CF" >> beam.ParDo(WriteToCloudFunction())

• ** For the caching to work the get_token function needs to be outside the beam.DoFn class

In times like this, I’m super thankful for Stack Overflow and all the people who contribute there!

I hope this article will help people like me who are frustrated with similar issues.

Let me know if you have another/a better way to make authenticated calls on cloud functions and I can add it here!