首页 > 解决方案 > Google Cloud - 使用 JWT 令牌访问存储 API:非法 URI 错误

问题描述

我正在尝试使用此处列出的示例对存储 API 进行 JWT 调用,其中一些更改如下 -

def generate_jwt():

"""Generates a signed JSON Web Token using a Google API Service Account."""

now = int(time.time())
sa_email = os.environ["FUNCTION_IDENTITY"]
expiry_length = 3600

# build payload
payload = {
    'iat': now,
    # expires after 'expiry_length' seconds.
    "exp": now + expiry_length,
    # iss must match 'issuer' in the security configuration in your
    # swagger spec (e.g. service account email). It can be any string.
    'iss': sa_email,
    # aud must be either your Endpoints service name, or match the value
    # specified as the 'x-google-audience' in the OpenAPI document.
    'aud': "https://storage.googleapis.com", 
    # sub and email should match the service account's email address
    'sub': sa_email,
    'email': sa_email
    
}
# sign with keyfile
sa_keyfile="cred.json"
signer = google.auth.crypt.RSASigner.from_service_account_file(sa_keyfile)
jwt = google.auth.jwt.encode(signer, payload)
return jwt

并在这里调用它

def make_jwt_request(signed_jwt, url="https://storage.googleapis.com/storage/v1/b/BUCKET_NAME"):
"""Makes an authorized request to the endpoint"""
headers = {
    'Authorization': 'Bearer {}'.format(signed_jwt.decode('utf-8')),
    'content-type': 'application/json',
    "Host": "www.googleapis.com",
}
response = requests.get(url, headers=headers)
print(response.status_code, response.content)
response.raise_for_status()

但由于无法解析指定的 URI 而出现错误。非法 URI。

我不明白为什么它是非法的 URI。我尝试使用https://googleapis.com/storage/b/BUCKETNMAE但仍然是同样的错误。在 SO 或谷歌文档上找不到任何关于此的内容。知道我在这里做错了什么吗?

标签: google-cloud-platformgoogle-apigoogle-cloud-functionsgoogle-cloud-storage

解决方案


Google Cloud Storage 不接受已签名的 JWT 进行授权。创建签名 JWT 后,您必须将 JWT 交换为访问令牌。

有关 Python 中的完整示例,请参阅我的答案或我的文章

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

推荐阅读