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

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

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.

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

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.

and use the token in the headers

Full Dataflow Code

  • ** 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!

Data Engineer @ League Inc.