无法使用服务帐户授权连接到自己的 GAE 端点 API

Posted 2023-02-24

【中文标题】无法使用服务帐户授权连接到自己的 GAE 端点 API

【英文标题】：Failing to authorize connection to own GAE endpoints API with service account

【发布时间】：2014-11-08 11:35:11

【问题描述】：

我一直在努力尝试成功授权对我使用 OAuth2 和服务帐户从 python 脚本运行的 Google App Engine (GAE) 项目上的 API 命中。

我创建了服务帐户，将服务帐户 ID 添加到 api 文件中允许的客户端 ID，将私钥从 .p12 转换为 .pem，并授权 httplib2 调用。我尝试使用 .authorize() 方法传递凭据，并将凭据加载为 JSON 并将 access_token 参数手动添加到标头 - "Authorization": "Bearer " + token_string.

API 的每次调用也会产生“无效令牌”。

我注意到一件奇怪的事情，当我第一次调用 SignedJwtAssertionCredentials 时，凭据中没有访问令牌——“access_token”为无。但是，当我从存储中的 .dat 文件中检索凭据时，访问令牌确实会显示出来。

以下是 GAE endpoints_api.py 文件、python_test.py 文件和 401 响应。

任何想法都将不胜感激。

首先是应用引擎端点服务器文件：

# endpoints_api.py running on GAE

import endpoints
import time
from protorpc import messages
from protorpc import message_types
from protorpc import remote

SERVICE_ACCOUNT_ID = 'random_account_id_string.apps.googleusercontent.com'
WEB_CLIENT_ID = 'random_web_client_id_string.apps.googleusercontent.com'
android_AUDIENCE = WEB_CLIENT_ID

package = "MyPackage"

class Status(messages.Message):
    message = messages.StringField(1)
    when = messages.IntegerField(2)

class StatusCollection(messages.Message):
    items = messages.MessageField(Status, 1, repeated=True)

STORED_STATUSES = StatusCollection(items=[
    Status(message='Go.', when=int(time.time())),
    Status(message='Balls.', when=int(time.time())),
    Status(message='Deep!', when=int(time.time())),
])

@endpoints.api(name='myserver', version='v1')
class MyServerApi(remote.Service):
    """MyServer API v1."""

    @endpoints.method(message_types.VoidMessage, StatusCollection,
                  allowed_client_ids=[SERVICE_ACCOUNT_ID,
                                      endpoints.API_EXPLORER_CLIENT_ID],
                  audiences=[ANDROID_AUDIENCE],
                  scopes=[endpoints.EMAIL_SCOPE],
                  path='status', http_method='GET',
                  name='statuses.listStatus')
    def statuses_list(self, unused_request):
        current_user = endpoints.get_current_user()
        if current_user is None:
            raise endpoints.UnauthorizedException('Invalid token.')
        else:
            return current_user.email(), STORED_STATUSES

APPLICATION = endpoints.api_server([MyServerApi])

接下来是本地python脚本：

# python_test.py file running from local server

from apiclient.discovery import build
from oauth2client.file import Storage
from oauth2client.client import SignedJwtAssertionCredentials
import httplib2
import os.path
import json

SERVICE_ACCOUNT_EMAIL = "[e-mail]"
ENDPOINT_URL = "http://my-project.appspot.com/_ah/api/myserver/v1/status"
SCOPE = 'https://www.googleapis.com/auth/userinfo.email'
SITE_ROOT = os.path.dirname(os.path.realpath(__file__))

f = file('%s/%s' % (SITE_ROOT, 'pk2.pem'), 'rb')
key = f.read()
f.close()

http = httplib2.Http()
storage = Storage('credentials.dat')
credentials = storage.get()

if credentials is None or credentials.invalid:
    credentials = SignedJwtAssertionCredentials(
        SERVICE_EMAIL, key, scope=SCOPE)
    storage.put(credentials)
else:
    credentials.refresh(http)

http = credentials.authorize(http)
headers = 'Content-Type': 'application/json'

(resp, content) = http.request(ENDPOINT_URL,
                           "GET",
                           headers=headers)

print(resp)
print(content)

最后是控制台输出：

'status': '401', 'alternate-protocol': '443:quic,p=0.002', 'content-length': '238', 'x- xss-protection': '1; mode=block', 'x-content-type-options': 'nosniff', 'transfer-encoding': 'chunked', 'expires': 'Sun, 14 Sep 2014 23:51:36 GMT', 'server': 'GSE', '-content-encoding': 'gzip', 'cache-control': 'private, max-age=0', 'date': 'Sun, 14 Sep 2014 23:51:36 GMT', 'x-frame-options': 'SAMEORIGIN', 'content-type': 'application/json; charset=UTF-8', 'www-authenticate': 'Bearer realm="https://accounts.google.com/AuthSubRequest"'

 "error": 
  "errors": [
   
    "domain": "global",
    "reason": "required",
    "message": "Invalid token.",
    "locationType": "header",
    "location": "Authorization"
   
  ],
  "code": 401,
  "message": "Invalid token."
 

【问题讨论】：

一定要用上面的方法吗？有一个 OAuth 装饰器将为您处理所有工作。重新编写代码不会花费太多时间。

developers.google.com/api-client-library/python/guide/… 你给它你的 OAtuh 详细信息，它会处理令牌和实际工作。

我对我在说什么一无所知，但看起来这仅在您处于应用引擎环境中时才有效。如果您在外面，则需要提供私钥。我得到了一个解决方案，使用 SignedJwtAssertionCredentials 并构建非常相似。我会发布它作为答案。

【参考方案1】：

感谢您的帮助。我在上面做错了几件事。

首先，我需要在允许的 ID 列表中引用应用引擎项目 ID。

API_ID = 'project-id-number.apps.googleusercontent.com'

@endpoints.method(message_types.VoidMessage, StatusCollection,
              allowed_client_ids=[API_ID, SERVICE_ACCOUNT_ID,
                                  endpoints.API_EXPLORER_CLIENT_ID],
              audiences=[ANDROID_AUDIENCE],
              scopes=[endpoints.EMAIL_SCOPE],
              path='status', http_method='GET',
              name='statuses.listStatus')

我错误地假设 build() 方法只能与官方的 google APIS 一起使用，但结果证明我错误地引用了我的项目。有了项目 ID，我可以使用 build() 而不是在服务器端文件中编写自己的 httplib2 调用（这不起作用）。

删除 'http = credentials.authorize(http)' 下面的代码，我将其替换为以下内容：

myservice = build('my-service', 'v1', http=http, discoveryServiceUrl=discoveryServiceUrl)
data = myservice.my-path().endpoint-name()
results = data.execute()

这成功授权我的帐户并调用端点。哇！如果其他人对此有疑问，或者我的解决方案不清楚，请随时发表评论。这是一个痛苦的过程，我不希望任何人发生。