授权类型应设置为urn:ietf:params:oauth:grant-type:jwt-bearer 记录在 REST API Making the access token request 部分下的here。
如果您使用 google-auth 库，它会非常容易和简单，它会自动处理解析私钥 json 文件、获取访问令牌、刷新它们并将它们实际包含在请求中。
您只需要提供请求 URL 和正文，其余的由库处理。这是一个简化的例子：
#!/usr/bin/env python
from google.auth.transport.requests import AuthorizedSession
from google.oauth2.service_account import Credentials
# BEGIN CONFIGURATION - change as needed.
# Path to the JSON file containing the service account private key and email.
PRIVATE_KEY_JSON = '/path/to/json/file'
# The API scope this token will be valid for.
API_SCOPES = ['https://www.googleapis.com/auth/pubsub']
# END CONFIGURATION
if __name__ == '__main__':
credentials = Credentials.from_service_account_file(
PRIVATE_KEY_JSON, scopes=API_SCOPES)
authed_session = AuthorizedSession(credentials)
url = 'https://pubsub.googleapis.com/v1/<SOMETHING>'
response = authed_session.get(url)
print str(response.content)
没有额外库的工作示例
如果您不想使用任何其他库但可以使用标准 python 库，这里有一个 Python 中的工作示例（使用我自己的服务帐户亲自测试）（支持 2.x 和 3.x 版本) 它负责所有步骤：
#!/usr/bin/env python
import Crypto.PublicKey.RSA as RSA
import Crypto.Hash.SHA256 as SHA
import Crypto.Signature.PKCS1_v1_5 as PKCS1_v1_5
import base64
import json
import time
try:
from urllib.request import urlopen
except ImportError:
from urllib2 import urlopen
try:
from urllib.parse import urlencode
except ImportError:
from urllib import urlencode
# BEGIN CONFIGURATION - change as needed.
# Path to the JSON file containing the service account private key and email.
PRIVATE_KEY_JSON = '/path/to/json/file'
# The API scope this token will be valid for.
API_SCOPE = 'https://www.googleapis.com/auth/pubsub'
# The validity of the token in seconds. Max allowed is 3600s.
ACCESS_TOKEN_VALIDITY_SECS = 3600
# END CONFIGURATION
class OauthAccessTokenGetter:
"""Fetches a new Google OAuth 2.0 access token.
The code is based on the steps described here: https://developers.go
ogle.com/identity/protocols/OAuth2ServiceAccount#authorizingrequests
"""
ACCESS_TOKEN_AUD = 'https://www.googleapis.com/oauth2/v4/token'
REQUEST_URL = 'https://www.googleapis.com/oauth2/v4/token'
GRANT_TYPE = 'urn:ietf:params:oauth:grant-type:jwt-bearer'
def __init__(self, private_key_json_file, scope, token_valid_secs=3600):
self.private_key_json = self.LoadPrivateKeyJsonFromFile(
private_key_json_file)
self.scope = scope
self.token_valid_secs = token_valid_secs
@classmethod
def Base64UrlEncode(cls, data):
"""Returns the base64url encoded string for the specified data."""
return base64.urlsafe_b64encode(data)
@classmethod
def LoadPrivateKeyJsonFromFile(cls, private_key_json_file):
"""Returns JSON object by parsing the specified private key JSON
file."""
with open(private_key_json_file) as private_key_json_file:
return json.load(private_key_json_file)
def GetPrivateKey(self):
"""Returns the imported RSA private key from the JSON data."""
return RSA.importKey(self.private_key_json['private_key'])
def GetSigner(self):
"""Returns a PKCS1-V1_5 object for signing."""
return PKCS1_v1_5.new(self.GetPrivateKey())
@classmethod
def GetEncodedJwtHeader(cls):
"""Returns the base64url encoded JWT header."""
return cls.Base64UrlEncode(json.dumps({'alg': 'RS256', 'typ': 'JWT'}).encode('utf-8'))
def GetEncodedJwtClaimSet(self):
"""Returns the base64url encoded JWT claim set."""
current_time_secs = int(time.time())
jwt_claims = {
'iss': self.private_key_json['client_email'],
'scope': self.scope,
'aud': self.ACCESS_TOKEN_AUD,
'exp': current_time_secs + self.token_valid_secs,
'iat': current_time_secs
}
return self.Base64UrlEncode(json.dumps(jwt_claims).encode('utf-8'))
def GetJwtSignature(self, message):
"""Returns signature of JWT as per JSON Web Signature (JWS) spec."""
signed_message = self.GetSigner().sign(SHA.new(message))
return self.Base64UrlEncode(signed_message)
def GetSignedJwt(self):
"""Returns signed JWT."""
header = self.GetEncodedJwtHeader()
jwt_claim_set = self.GetEncodedJwtClaimSet()
signature = self.GetJwtSignature(header + b'.' + jwt_claim_set)
return header + b'.' + jwt_claim_set + b'.' + signature
def SendRequest(self, body):
"""Returns the response by sending the specified request."""
return urlopen(self.REQUEST_URL, urlencode(body).encode('utf-8')).read()
def GetAccessToken(self):
"""Returns the access token."""
body = {
'grant_type': self.GRANT_TYPE,
'assertion': self.GetSignedJwt()
}
response = json.loads(self.SendRequest(body))
return response['access_token']
if __name__ == '__main__':
print (OauthAccessTokenGetter(PRIVATE_KEY_JSON, API_SCOPE,
ACCESS_TOKEN_VALIDITY_SECS).GetAccessToken())
获得访问令牌后，您需要将其作为Bearer 标头包含在您以described here 发送的请求中。
GET /drive/v2/files HTTP/1.1
Authorization: Bearer <access_token>
Host: www.googleapis.com/
在 curl 中等效为：
curl -H "Authorization: Bearer <access_token>" https://www.googleapis.com/drive/v2/files
虽然您可以使用access_token= 参数指定令牌是described here，但我无法让它至少适用于Google Compute Engine API，可能它适用于PubSub，但Bearer 标头方法有根据我的经验一直工作。
更新：根据discovery doc for PubSub API，access_token= 似乎有一个查询参数，所以它也可能很好用。
"access_token": {
"description": "OAuth access token.",
"type": "string",
"location": "query"
},
而discovery doc for Compute Engine APIs 表示使用oauth_token 查询参数，我确实验证了它是否有效。
"oauth_token": {
"type": "string",
"description": "OAuth 2.0 token for the current user.",
"location": "query"
},