使用Google App Engine上的flask_client自动或手动刷新访问令牌

卡梅伦·罗伯茨(Cameron Roberts)

我可以使用第三方OAuth2提供程序(Xero)成功授权我的应用程序,但是无法自动或手动刷新令牌。

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

使用这两种方法,都不会尝试刷新令牌,将请求与过期的令牌一起传递到Xero,我收到一个错误,唯一的继续方法是使用Xero手动重新授权应用程序。

这是Web客户端文档中“ 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'

在自动刷新失败的情况下,如何使用自动刷新工作,并在使用flask客户端时如何手动触发刷新请求?

卡梅伦·罗伯茨(Cameron Roberts)

我相信我已经在这里找到问题了,其根源是在初始化OAuth时传递了Cache(用于临时凭证存储):

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”函数上的参数需要进行调整以匹配文档,但这与该问题正在解决的原始问题无关。

本文收集自互联网,转载请注明来源。

如有侵权,请联系 [email protected] 删除。

编辑于
0

我来说两句

0 条评论
登录 后参与评论

相关文章