我如何使用REST API从服务帐户凭据创建访问令牌？

Question

我已经在Google Cloud Platform中创建了一个服务帐户，并以JSON格式下载了私钥。我正在尝试通过REST API创建Compute资源。为了进行身份验证，我需要一个AccessToken，它需要设置为创建计算资源REST API的标头。是否有一个REST API可以从私钥获取访问令牌（不使用SDK或Google客户端）？

Answer 1

下面的示例向您展示了一些重要的步骤，这些步骤无需使用Python中的SDK即可调用Google Cloud API。类似的代码几乎可以在任何语言（c＃，java，php，nodejs）中工作。

使用服务帐户Json文件的文件名，Google区域和项目ID更改源代码。

此示例将在指定项目的一个区域中列出实例。从此示例中，您将了解调用API来创建GCE实例的框架。

此代码将向您展示如何：

1. 如何从Json文件加载服务帐户凭据。
2. 如何提取用于签署请求的私钥。
3. 如何为Google Oauth 2.0创建JWT（Json Web令牌）。
4. 如何设置Google范围（权限）。
5. 如何对JWT签名以创建签名JWT（JWS）。
6. 如何将Signed-JWT交换为Google OAuth 2.0访问令牌。
7. 如何设置到期时间。该程序默认为3600秒（1小时）。
8. 如何调用Google API并设置授权标头。
9. 如何处理返回的Json结果并显示每个实例的名称。

Python 3.x中的示例程序：

'''
This program lists lists the Google Compute Engine Instances in one zone
'''
# Author: John Hanley
# https://www.jhanley.com

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)

有关更多信息，请访问我的博客。我写这样的文章并发布源代码，以帮助其他人了解如何为云编写软件。

www.jhanley.com

Answer 2

有一种更简单的方法，可以使用Google库从服务帐户生成令牌

from google.auth.transport import requests
from google.oauth2 import service_account

CREDENTIAL_SCOPES = ["https://www.googleapis.com/auth/cloud-platform"]
CREDENTIALS_KEY_PATH = '/PATH/TO/SERVICE_ACCOUNT.json'

def get_service_account_token():
  credentials = service_account.Credentials.from_service_account_file(
          CREDENTIALS_KEY_PATH, scopes=CREDENTIAL_SCOPES)
  credentials.refresh(requests.Request())
  return credentials.token

或者如果您想使用默认身份验证

import google
from google.auth.transport import requests

CREDENTIAL_SCOPES = ["https://www.googleapis.com/auth/cloud-platform"] 

def get_default_token():
  credentials, project_id = google.auth.default(scopes=CREDENTIAL_SCOPES)
  credentials.refresh(requests.Request())
  return credentials.token

创建credentials对象时，令牌为空，但刷新credentials之后，它包含可在API请求中用作标头的访问令牌