谷歌云-使用JWT令牌访问存储API：非法URI错误

Question

我尝试使用下面列出的示例对存储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()

但是获取错误as无法解析指定的URI。非法URI。

我不明白为什么它是一个非法的URI。我尝试了https://googleapis.com/storage/b/BUCKETNMAE，但还是一样的错误。在SO或谷歌文档上找不到任何关于这个的东西。任何想法我在这里做什么错了吗？

Answer 1

谷歌云存储不接受签名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