python - 如何在 Tornado 中使用 OpenID 进行身份验证？

Question

我正在将 Tornado Web 服务器用于一个简单的 Web 应用程序，并希望使用 OpenID 对用户进行身份验证。我是 Tornado 的新手，我设法使用 Node.js 的 Passport 包来工作(首先在 Node.js 上进行测试)，我能够在回调中获取 id_token。

我正在使用 tornado.auth 中的 OAuth2Mixin 使用用户凭据授予来授权访问，然后在重定向时，我从 code 获取 获取参数。我不知道如何从那里继续:D

from tornado.auth import OpenIdMixin, OAuth2Mixin
from .base import BaseHandler

class LoginHandler(BaseHandler, OAuth2Mixin, OpenIdMixin):
  def get(self):
    self._OAUTH_AUTHORIZE_URL = 'https://authserver.io/uas/oauth2/authorization'
    self._OAUTH_ACCESS_TOKEN_URL = 'https://authserver.io/uas/oauth2/token'
    self.authorize_redirect(
      redirect_uri='http://localhost:3001/success-login',
      client_id='abcd',
      client_secret='1234',
    )

然后在另一个处理程序上。

from tornado.auth import OpenIdMixin, OAuth2Mixin
import tornado.httpclient
from .base import BaseHandler

class SuccessLoginHandler(BaseHandler, OpenIdMixin, OAuth2Mixin):
  async def get(self):
    code = self.get_argument('code', None)
    if code is not None:
      return self.write(code)

    self.write('no code')

我希望返回id_token；这是一个 JWT。我可以对其进行解码并获取所需的数据。

更新:如果需要配置。

{"issuer":"https://authserver.io/uas","authorization_endpoint":"https://authserver.io/uas/oauth2/authorization","token_endpoint":"https://authserver.io/uas/oauth2/token","userinfo_endpoint":"https://authserver.io/uas/oauth2/userinfo","jwks_uri":"https://authserver.io/uas/oauth2/metadata.jwks","tokeninfo_endpoint":"https://authserver.io/uas/oauth2/introspection","introspection_endpoint":"https://authserver.io/uas/oauth2/introspection","revocation_endpoint":"https://authserver.io/uas/oauth2/revocation","response_types_supported":["code"],"grant_types_supported":["authorization_code","password","refresh_token","urn:ietf:params:oauth:grant-type:saml2-bearer","http://globalsign.com/iam/sso/oauth2/grant-type/sms-mt-otp","http://globalsign.com/iam/sso/oauth2/grant-type/smtp-otp"],"subject_types_supported":["public"],"request_object_signing_alg_values_supported":["RS256","HS256"],"request_object_encryption_alg_values_supported":["RSA-OAEP","RSA1_5","A128KW"],"request_object_encryption_enc_values_supported":["A128GCM","A128CBC-HS256"],"id_token_signing_alg_values_supported":["RS256","HS256"],"id_token_encryption_alg_values_supported":["RSA-OAEP","RSA1_5","A128KW"],"id_token_encryption_enc_values_supported":["A128GCM","A128CBC-HS256"],"userinfo_signing_alg_values_supported":["RS256","HS256"],"userinfo_encryption_alg_values_supported":["RSA-OAEP","RSA1_5","A128KW"],"userinfo_encryption_enc_values_supported":["A128GCM","A128CBC-HS256"],"token_endpoint_auth_methods_supported":["client_secret_post","client_secret_basic","client_secret_jwt","private_key_jwt"],"token_endpoint_auth_signing_alg_values_supported":["RS256","HS256"],"introspection_endpoint_auth_methods_supported":["client_secret_post","client_secret_basic","client_secret_jwt","private_key_jwt"],"introspection_endpoint_auth_signing_alg_values_supported":["RS256","HS256"],"revocation_endpoint_auth_methods_supported":["client_secret_post","client_secret_basic","client_secret_jwt","private_key_jwt"],"revocation_endpoint_auth_signing_alg_values_supported":["RS256","HS256"],"scopes_supported":["openid","userinfo"]}

Answer 1

您需要调用get_authenticated_user从 SuccessLoginHandler 获取访问 token 。

但是，我宁愿将所有内容编写在一个处理程序中，以保持代码更短且不重复。您可以像这样重写 LoginHandler:

class LoginHandler(BaseHandler, OAuth2Mixin, OpenIdMixin):

    _OAUTH_AUTHORIZE_URL = 'https://authserver.io/uas/oauth2/authorization'
    _OAUTH_ACCESS_TOKEN_URL = 'https://authserver.io/uas/oauth2/token'

    async def get(self):
        redirect_uri = 'http://localhost:3001/login'

        code = self.get_argument('code', None)

        if code:
            # if there's `code`, get access token
            user = await self.get_authenticated_user()

            # the `user` variable now contains the returned data
            # from the oauth server.
            # you'll probably want to `set_secure_cookie`
            # or do something else to save the user

            # then redirect the user to some page
            self.redirect("/") # redirects to home page
            return 

        else:
            # otherwise get authorization `code`       
            self.authorize_redirect(
                redirect_uri=redirec_uri,
                client_id='abcd',
                client_secret='1234',
            )