Да, вы можете создать свой собственный подписанный JWT из файла Json (или P12) служебной учетной записи и обменять JWT на токен доступа, который вы затем используете как Authorization: Bearer TOKEN
Я написал рядстатьи о том, как использовать учетные данные Json и P12.
Google Cloud - Создание токенов доступа OAuth для вызовов REST API
По вашим вопросам:
Я не уверен, каким должен быть «aud» при создании JWT.Я видел примеры, где это URL, а также, где это идентификатор проекта.Ни одна из них не работает для меня.
Установите aud на "https://www.googleapis.com/oauth2/v4/token"
В одном из примеров JSON API сказано, что токен авторизации должен быть токеном oauth.Могу ли я использовать JWT вместо этого или мне нужно сделать вызов с использованием JWT для получения токена доступа?
Некоторые API принимают подписанные JWT, другие ожидают OAuth Access Token.Просто всегда проще получить токен доступа OAuth.В приведенном ниже примере кода я покажу вам, как.
Правильно ли указан путь моего сегмента?Является ли базовая папка для пути сегмента вашим идентификатором проекта?Должен ли мой путь быть / {projectId} / test.Я пробовал оба варианта, но ни один из них не работает.
Ваш URL выглядит так (пример построения строки Python)
url = "https://www.googleapis.com/storage/v1/b?project=" + project
Ниже я покажу вам, как вызывать две службы (GCEи GCS).Большинство API Google используют аналогичные стили для создания URL-адресов REST API.
Из кода в вашем вопросе вы пропускаете последний шаг в процессе OAuth.Вам необходимо обменять подписанный JWT на токен доступа.
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
Вот полный пример Python 3.x, в котором будут перечислены экземпляры GCE.Ниже этого кода приведены изменения для отображения GCS Buckets.
'''
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)
Чтобы вместо этого отображать GCS Buckets, измените код:
# Create the HTTP url for the Google Storage REST API
url = "https://www.googleapis.com/storage/v1/b?project=" + project
resp, content = h.request(uri=url, method="GET", headers=headers)
s = content.decode('utf-8').replace('\n', '')
j = json.loads(s)
print('')
print('Buckets')
print('----------------------------------------')
for item in j['items']:
print(item['name'])