使用Python进行身份验证以访问GCP计算API端点

Question

我的目标是在不依赖gcloud二进制文件的情况下复制/复制gcloud compute addresses create的功能。

我正在尝试使用Python根据https://cloud.google.com/compute/docs/ip-addresses/reserve-static-external-ip-address中有关保留静态外部IP地址的文档对googleapis计算终端进行POST身份验证。

但是我的POST每次都返回401。

我已经从google.auth.jwt Python模块创建了一个JWT，当我解码它时，JWT中嵌入了我期望出现的所有字符串。

我还尝试了包括以下OAuth范围的组合： - "https://www.googleapis.com/auth/userinfo.email" - "https://www.googleapis.com/auth/compute" - "https://www.googleapis.com/auth/cloud-platform"

这是我获取JWT的函数，使用我的服务帐户JSON密钥文件中的信息。

def _generate_jwt( tokenPath, expiry_length=3600 ):
    now = int(time.time())
    tokenData = load_json_data( tokenPath )
    sa_email = tokenData['client_email']
    payload = {
        'iat': now, 
        # expires after 'expiry_length' seconds.
        "exp": now + expiry_length,
        'iss': sa_email,
        "scope": " ".join( [
            "https://www.googleapis.com/auth/cloud-platform",
            "https://www.googleapis.com/auth/compute",
            "https://www.googleapis.com/auth/userinfo.email"
        ] ),
        'aud': "https://www.googleapis.com/oauth2/v4/token",
        'email': sa_email
    }
    # sign with keyfile
    signer = google.auth.crypt.RSASigner.from_service_account_file( tokenPath )
    jwt = google.auth.jwt.encode(signer, payload)

    return jwt

一旦我获得JWT，然后进行以下POST请求失败，返回401错误。

    gapiURL = 'https://www.googleapis.com/compute/v1/projects/' + projectID + '/regions/' + region + '/addresses'
    jwtToken = _generate_jwt( servicetoken )
    headers = {  
        'Authorization': 'Bearer {}'.format( jwtToken ),
        'content-type' : 'application/json',
    }    
    post = requests.post( url=gapiURL, headers=headers, data=data ) 
    post.raise_for_status()
    return post.text

无论我在JWT中使用了多少个范围组合或为我的服务帐户提供了多少权限，我都收到了401错误。我做错了什么吗？

编辑：非常感谢@JohnHanley指出我在GCP的身份验证序列中缺少下一个/第二个POST到https://www.googleapis.com/oauth2/v4/token URL。因此，您会获得一个JWT来获取“访问令牌”。

我已更改我的调用，使用python jwt模块而不是google.auth.jwt模块和google.auth.crypt.RSASigner结合使用。所以代码有点简单，我将其放在一个方法中。

## serviceAccount auth sequence for google :: JWT -> accessToken
def gke_get_token( serviceKeyDict, expiry_seconds=3600 ):

    epoch_time = int(time.time())
    # Generate a claim from the service account file.
    claim = {
        "iss": serviceKeyDict["client_email"],
        "scope": " ".join([
            "https://www.googleapis.com/auth/cloud-platform",
            "https://www.googleapis.com/auth/userinfo.email"
        ]),
        "aud": "https://www.googleapis.com/oauth2/v4/token",
        "exp": epoch_time + expiry_seconds,
        "iat": epoch_time
    }    
    # Sign claim with JWT.
    assertion = jwt.encode( claim, serviceKeyDict["private_key"], algorithm='RS256' ).decode() 
    data = urllib.urlencode( {
        "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
        "assertion": assertion
    } )  
    # Request the access token.
    result = requests.post(
        url="https://www.googleapis.com/oauth2/v4/token",
        headers={
            "Content-Type": "application/x-www-form-urlencoded"
        },
        data=data
    )    
    result.raise_for_status()
    return loadJsonData(result.text)["access_token"]

- nollimahere

---

您尚未创建访问令牌，而是创建了已签名的JWT。您需要交换已签名的JWT以获取访问令牌。我在Python中撰写了一篇关于如何执行此操作的文章。https://www.jhanley.com/google-cloud-creating-oauth-access-tokens-for-rest-api-calls/ - John Hanley

@JohnHanley - 感谢您的澄清。我阅读了您的文章，并成功地为我的服务帐户获取了身份验证令牌。请将您的评论更改为答案，以便我选择它，因为这是正确的解决方案，我希望您能得到相应的认可。我稍后会根据阅读您的文章所做的更改编辑我的问题。 - nollimahere

回答已发布。谢谢。 - John Hanley

Answer 1

在Google Cloud中，有三种授予访问权限的“令牌”类型：

• Signed JWT
• Access Token
• Identity Token

在您的情况下，您创建了一个Signed JWT。一些Google服务接受此令牌，但大多数不接受。

创建Signed JWT后，下一步是调用Google OAuth端点并交换Access Token。我写了一篇详细描述这个过程的文章： Google Cloud – Creating OAuth Access Tokens for REST API Calls 现在，一些Google服务接受身份标识令牌。这称为基于身份的访问控制（IBAC）。这与您的问题无关，但是它是Google Cloud授权未来的趋势。例如，我的Cloud Run + Cloud Storage + KMS文章： Google Cloud – Go – Identity Based Access Control 以下示例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

- John Hanley