时间戳

首页 > 解决方案 > 带有 JWT 令牌的 Google Cloud Storage JSON API

google-cloud-platform - 带有 JWT 令牌的 Google Cloud Storage JSON API

问题描述

我正在尝试使用 JSON API for Google Cloud Storage 从 Google Cloud Storage 检索文件。我不允许使用 SDK。是否可以从 ServiceAccount.json 文件创建 JWT 并使用 JWT 从 Google Cloud Storage 访问文件？我在 node.js 中有一个从服务帐户生成 JWT 的脚本，但我不确定受众是否正确

const jwt = require('jsonwebtoken');
const serviceAccount = require('./serviceAccount.json');
const issuedAt = Math.floor(Date.now() / 1000);

const TOKEN_DURATION_IN_SECONDS = 3600;

let params = {
    'iss': serviceAccount.client_email,
    'sub': serviceAccount.client_email,
    'aud': serviceAccount.project_id,
    'iat': issuedAt,
    'exp': issuedAt + TOKEN_DURATION_IN_SECONDS,
};

let options = {
    algorithm: 'RS256',
    header: {
        'kid': serviceAccount.private_key_id,
        'typ': 'JWT',
        'alg': 'RS256',
    },
};

let token = jwt.sign(params, serviceAccount.private_key, options);
console.log(token);

然后我使用该 JWT 调用 Google Cloud Storage JSON API：

https://www.googleapis.com/storage/v1/b/test

使用标题：Authorization Bearer {token}

这只是导致了Invalid Credentials回应。

几个问题：

回顾

这是一个物联网项目，我需要嵌入式设备从谷歌云存储下载文件。我需要创建一个门户网站以将文件上传到（使用 Firebase 函数）并将存储桶路径或私有/签名 URL 传递给设备。底线是我需要使用服务帐户密钥访问 Google Cloud Storage 存储桶。如果有嵌入式 SDK - 很好，但我找不到 C 的。我唯一的想法是使用 JSON API。如果有办法我可以签署一个只能使用服务帐户访问的 URL - 这也可以。

谢谢！

标签： google-cloud-platformgoogle-cloud-storage

解决方案

是的，您可以从服务帐户 Json（或 P12）文件创建自己的签名 JWT，并将 JWT 交换为访问令牌，然后将其用作Authorization: Bearer TOKEN

我写了很多关于如何使用 Json 和 P12 凭证的文章。

Google Cloud – 为 REST API 调用创建 OAuth 访问令牌

对于您的问题：

我不确定创建 JWT 时的“aud”应该是什么。我已经看到了它是一个 url 以及它是 projectId 的例子。都不适合我。

设置aud为"https://www.googleapis.com/oauth2/v4/token"

JSON API 示例之一说授权令牌应该是 oauth 令牌。我可以改用 JWT 还是需要使用 JWT 进行调用以获取访问令牌？

一些 API 接受签名的 JWT，而另一些则需要 OAuth 访问令牌。总是更容易获得 OAuth 访问令牌。在下面的示例代码中，我将向您展示如何操作。

我的存储桶路径是否正确？存储桶路径的基本文件夹是您的 projectId 吗？我的路径应该是/{projectId}/test。我都试过了，都没有。

您的 url 看起来像这样（Python 字符串构建示例）

url = "https://www.googleapis.com/storage/v1/b?project=" + project

下面我将向您展示如何调用两个服务（GCE 和 GCS）。大多数 Google API 将遵循类似的样式来构建 REST API url。

从您问题中的代码中，您错过了 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 存储桶的更改。

'''
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 存储桶，请修改代码：

# 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'])

推荐阅读