1

我成功地使用第 3 方 OAuth2 提供程序 (Xero) 授权我的应用程序，但无法自动或手动刷新令牌。

文档建议 authlib 可以自动执行此操作。我从 Authlib 文档中尝试了两种不同的方法，在烧瓶客户端文档中，他们给出了“通过信号自动更新令牌”的示例，在Web 客户端文档中，他们注册了“update_token”函数。

使用任何一种方法，都不会尝试刷新令牌，请求会使用过期令牌传递给 Xero，我收到错误消息，唯一继续的方法是使用 Xero 手动重新授权应用程序。

这是来自网络客户端文档的“update_token”方法的相关代码：

#this never ends up getting called.
def save_xero_token(name,token,refresh_token=None,access_token=None,tenant_id=None):
    logging.info('Called save xero token.')
    #removed irrelevant code that stores token in NDB here.

cache = Cache()
oauth = OAuth(app,cache=cache)
oauth.register(name='xero',
               client_id = Meta.xero_consumer_client_id,
               client_secret = Meta.xero_consumer_secret,
               access_token_url = 'https://identity.xero.com/connect/token',
               authorize_url = 'https://login.xero.com/identity/connect/authorize',
               fetch_token = fetch_xero_token,
               update_token = save_xero_token,
               client_kwargs={'scope':' '.join(Meta.xero_oauth_scopes)},
              )

xero_tenant_id = 'abcd-123-placeholder-for-stackoverflow'
url = 'https://api.xero.com/api.xro/2.0/Invoices/ABCD-123-PLACEHOLDER-FOR-STACKOVERFLOW'
headers = {'Xero-tenant-id':xero_tenant_id,'Accept':'application/json'}

response = oauth.xero.get(url,headers=headers)    #works fine until token is expired.

我将我的令牌存储在以下 NDB 模型中：

class OAuth2Token(ndb.Model):
    name = ndb.StringProperty()
    token_type = ndb.StringProperty()
    access_token = ndb.StringProperty()
    refresh_token = ndb.StringProperty()
    expires_at = ndb.IntegerProperty()
    xero_tenant_id = ndb.StringProperty()

    def to_token(self):
        return dict(
            access_token=self.access_token,
            token_type=self.token_type,
            refresh_token=self.refresh_token,
            expires_at=self.expires_at
        )

为了完整起见，这是我存储来自 Xero 的初始响应的方式（效果很好）：

@app.route('/XeroOAuthRedirect')
def xeroOAuthLanding():
    token = oauth.xero.authorize_access_token()
    connections_response = oauth.xero.get('https://api.xero.com/connections')
    connections = connections_response.json()
    for tenant in connections:
        print('saving first org, this app currently supports one xero org only.')
        save_xero_token('xero',token,tenant_id=tenant['tenantId'])

    return 'Authorized application with Xero'

如何让自动刷新工作，以及如何在使用烧瓶客户端时手动触发刷新请求，以防自动刷新失败？

4

1 回答 1

2

我相信我在这里找到了问题，其根源是在初始化 OAuth 时传递了一个缓存（用于临时凭证存储）：

cache = Cache()
oauth = OAuth(app,cache=cache)

当缓存被传递时，它似乎抢占了 update_token（可能还有 fetch_token）参数。

应该很简单：

oauth = OAuth(app)

oauth.register(name='xero',
               client_id = Meta.xero_consumer_client_id,
               client_secret = Meta.xero_consumer_secret,
               access_token_url = 'https://identity.xero.com/connect/token',
               authorize_url = 'https://login.xero.com/identity/connect/authorize',
               fetch_token = fetch_xero_token,
               update_token = save_xero_token,
               client_kwargs={'scope':' '.join(Meta.xero_oauth_scopes)},
              )

此外，我的“save_xero_token”函数的参数需要调整以匹配文档，但这与问题所解决的原始问题无关。

于 2020-04-17T19:29:00.560 回答