Google Cloud - Creating OAuth Access Tokens for REST API Calls

The following example shows several important steps to call Google Cloud APIs without using an SDK in Python. Similar code works in just about any language (c#, java, php, nodejs).

Change the source code with the filename of your service account Json file, your Google Zone and your Project ID.

This example will list the instances in one zone for the specified project. From this example you will know the framework to call almost any Google Cloud API.

This code will show you how to:

How to load service account credentials from a Json file.
How to extract the Private Key used to sign requests.
How to create a JWT (Json Web Token) for Google Oauth 2.0.
How to set the Google Scopes (permissions).
How to sign a JWT to create a Signed-JWT (JWS).
How to exchange the Signed-JWT for a Google OAuth 2.0 Access Token.
How to set the expiration time. This program defaults to 3600 seconds (1 Hour).
How to call a Google API and set the Authorization Header.
How to process the returned Json results and display the name of each instance.

Example program in Python 3.x:

############################################################
# Version 1.00
# Date Created: 2018-12-23
# Last Update:  2018-12-23
# https://www.jhanley.com
# Copyright (c) 2018, John J. Hanley
# Author: John Hanley
############################################################

'''
This program lists lists the Google Compute Engine Instances in one zone
'''

import time
import json
import jwt
import requests
import httplib2

# Project ID for this request.
project = 'development-123456'

# The name of the zone for this request.
zone = 'us-west1-a'

# Service Account Credentials, Json format
json_filename = 'service-account.json'

# Permissions to request for Access Token
scopes = "https://www.googleapis.com/auth/cloud-platform"

# Set how long this token will be valid in seconds
expires_in = 3600	# Expires in 1 hour

def load_json_credentials(filename):
	''' Load the Google Service Account Credentials from Json file '''

	with open(filename, 'r') as f:
		data = f.read()

	return json.loads(data)

def load_private_key(json_cred):
	''' Return the private key from the json credentials '''

	return json_cred['private_key']

def create_signed_jwt(pkey, pkey_id, email, scope):
	'''
	Create a Signed JWT from a service account Json credentials file
	This Signed JWT will later be exchanged for an Access Token
	'''

	# Google Endpoint for creating OAuth 2.0 Access Tokens from Signed-JWT
	auth_url = "https://www.googleapis.com/oauth2/v4/token"

	issued = int(time.time())
	expires = issued + expires_in	# expires_in is in seconds

	# Note: this token expires and cannot be refreshed. The token must be recreated

	# JWT Headers
	additional_headers = {
			'kid': pkey_id,
			"alg": "RS256",
			"typ": "JWT"	# Google uses SHA256withRSA
	}

	# JWT Payload
	payload = {
		"iss": email,		# Issuer claim
		"sub": email,		# Issuer claim
		"aud": auth_url,	# Audience claim
		"iat": issued,		# Issued At claim
		"exp": expires,		# Expire time
		"scope": scope		# Permissions
	}

	# Encode the headers and payload and sign creating a Signed JWT (JWS)
	sig = jwt.encode(payload, pkey, algorithm="RS256", headers=additional_headers)

	return sig

def exchangeJwtForAccessToken(signed_jwt):
	'''
	This function takes a Signed JWT and exchanges it for a Google OAuth Access Token
	'''

	auth_url = "https://www.googleapis.com/oauth2/v4/token"

	params = {
		"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
		"assertion": signed_jwt
	}

	r = requests.post(auth_url, data=params)

	if r.ok:
		return(r.json()['access_token'], '')

	return None, r.text

def gce_list_instances(accessToken):
	'''
	This functions lists the Google Compute Engine Instances in one zone
	'''

	# Endpoint that we will call
	url = "https://www.googleapis.com/compute/v1/projects/" + project + "/zones/" + zone + "/instances"

	# One of the headers is "Authorization: Bearer $TOKEN"
	headers = {
		"Host": "www.googleapis.com",
		"Authorization": "Bearer " + accessToken,
		"Content-Type": "application/json"
	}

	h = httplib2.Http()

	resp, content = h.request(uri=url, method="GET", headers=headers)

	status = int(resp.status)

	if status < 200 or status >= 300:
		print('Error: HTTP Request failed')
		return

	j = json.loads(content.decode('utf-8').replace('\n', ''))

	print('Compute instances in zone', zone)
	print('------------------------------------------------------------')
	for item in j['items']:
		print(item['name'])

if __name__ == '__main__':
	cred = load_json_credentials(json_filename)

	private_key = load_private_key(cred)

	s_jwt = create_signed_jwt(
			private_key,
			cred['private_key_id'],
			cred['client_email'],
			scopes)

	token, err = exchangeJwtForAccessToken(s_jwt)

	if token is None:
		print('Error:', err)
		exit(1)

	gce_list_instances(token)

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

############################################################

# Version 1.00

# Date Created: 2018-12-23

# Last Update: 2018-12-23

# https://www.jhanley.com

# Author: John Hanley

############################################################

'''

This program lists lists the Google Compute Engine Instances in one zone

'''

import time

import json

import jwt

import requests

import httplib2

# Project ID for this request.

project = 'development-123456'

# The name of the zone for this request.

zone = 'us-west1-a'

# Service Account Credentials, Json format

json_filename = 'service-account.json'

# Permissions to request for Access Token

scopes = "https://www.googleapis.com/auth/cloud-platform"

# Set how long this token will be valid in seconds

expires_in = 3600 # Expires in 1 hour

def load_json_credentials(filename):

''' Load the Google Service Account Credentials from Json file '''

with open(filename, 'r') as f:

data = f.read()

return json.loads(data)

def load_private_key(json_cred):

''' Return the private key from the json credentials '''

return json_cred['private_key']

def create_signed_jwt(pkey, pkey_id, email, scope):

'''

Create a Signed JWT from a service account Json credentials file

This Signed JWT will later be exchanged for an Access Token

'''

# Google Endpoint for creating OAuth 2.0 Access Tokens from Signed-JWT

auth_url = "https://www.googleapis.com/oauth2/v4/token"

issued = int(time.time())

expires = issued + expires_in # expires_in is in seconds

# Note: this token expires and cannot be refreshed. The token must be recreated

# JWT Headers

additional_headers = {

'kid': pkey_id,

"alg": "RS256",

"typ": "JWT" # Google uses SHA256withRSA

}

# JWT Payload

payload = {

"iss": email, # Issuer claim

"sub": email, # Issuer claim

"aud": auth_url, # Audience claim

"iat": issued, # Issued At claim

"exp": expires, # Expire time

"scope": scope # Permissions

}

# Encode the headers and payload and sign creating a Signed JWT (JWS)

sig = jwt.encode(payload, pkey, algorithm="RS256", headers=additional_headers)

return sig

def exchangeJwtForAccessToken(signed_jwt):

'''

This function takes a Signed JWT and exchanges it for a Google OAuth Access Token

'''

auth_url = "https://www.googleapis.com/oauth2/v4/token"

params = {

"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",

"assertion": signed_jwt

}

r = requests.post(auth_url, data=params)

if r.ok:

return(r.json()['access_token'], '')

return None, r.text

def gce_list_instances(accessToken):

'''

This functions lists the Google Compute Engine Instances in one zone

'''

# Endpoint that we will call

url = "https://www.googleapis.com/compute/v1/projects/" + project + "/zones/" + zone + "/instances"

# One of the headers is "Authorization: Bearer $TOKEN"

headers = {

"Host": "www.googleapis.com",

"Authorization": "Bearer " + accessToken,

"Content-Type": "application/json"

}

h = httplib2.Http()

resp, content = h.request(uri=url, method="GET", headers=headers)

status = int(resp.status)

if status < 200 or status >= 300:

print('Error: HTTP Request failed')

return

j = json.loads(content.decode('utf-8').replace('\n', ''))

print('Compute instances in zone', zone)

print('------------------------------------------------------------')

for item in j['items']:

print(item['name'])

if __name__ == '__main__':

cred = load_json_credentials(json_filename)

private_key = load_private_key(cred)

s_jwt = create_signed_jwt(

private_key,

cred['private_key_id'],

cred['client_email'],

scopes)

token, err = exchangeJwtForAccessToken(s_jwt)

if token is None:

print('Error:', err)

exit(1)

gce_list_instances(token)

John Hanley

Security, software development and devops in a cloud world - AWS, Azure, Google, IBM & Alibaba

Google Cloud – Creating OAuth Access Tokens for REST API Calls

John Hanley

Leave a Reply Cancel reply

Subscribe to Blog via Email

Search

Tags

Recent Posts

Find Us