Authenticated calls to cloud functions with Python

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 urllibimport 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!

--

--

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store