python - Flask 社交认证类问题

Question

我正在编写有关社交认证的 Miguel Grinberg 教程。

在主页模板上，我有此代码，并且我从教程中删除了 Twitter 部分:

    <h2>I don't know you!</h2>
    <p><a href="{{ url_for('oauth_authorize', provider='facebook') }}">Login with Facebook</a></p>
{% endif %}

因此，当您单击该链接时，您将通过此 View 函数将 Facebook 作为提供者传递:

@app.route('/authorize/<provider>')
def oauth_authorize(provider):
    if not current_user.is_anonymous():
        return redirect(url_for('index'))
    oauth = OAuthSignIn.get_provider(provider)
    return oauth.authorize()

现在，在另一个文件 oauth.py 中，我有以下内容，我的问题是这样的。当我单击 Facebook 链接时，除非删除 TwitterSignIn 类，否则我不断收到错误消息。我想我很好奇为什么需要删除 TwitterSignIn 类才能使其工作，因为没有数据传递给它，对吗？即使 Facebook 不是唯一的选择，为什么单击 Facebook 登录链接会将任何数据传递到 TwitterSignIn 类？

from rauth import OAuth1Service, OAuth2Service
from flask import current_app, url_for, request, redirect, session


class OAuthSignIn(object):
    providers = None

    def __init__(self, provider_name):
        self.provider_name = provider_name
        credentials = current_app.config['OAUTH_CREDENTIALS'][provider_name]
        self.consumer_id = credentials['id']
        self.consumer_secret = credentials['secret']

    def authorize(self):
        pass

    def callback(self):
        pass

    def get_callback_url(self):
        return url_for('oauth_callback', provider=self.provider_name,
                       _external=True)

    @classmethod
    def get_provider(self, provider_name):
        if self.providers is None:
            self.providers = {}
            for provider_class in self.__subclasses__():
                provider = provider_class()
                self.providers[provider.provider_name] = provider
        return self.providers[provider_name]


class FacebookSignIn(OAuthSignIn):
    def __init__(self):
        super(FacebookSignIn, self).__init__('facebook')
        self.service = OAuth2Service(
            name='facebook',
            client_id=self.consumer_id,
            client_secret=self.consumer_secret,
            authorize_url='https://graph.facebook.com/oauth/authorize',
            access_token_url='https://graph.facebook.com/oauth/access_token',
            base_url='https://graph.facebook.com/'
        )

    def authorize(self):
        return redirect(self.service.get_authorize_url(
            scope='email',
            response_type='code',
            redirect_uri=self.get_callback_url())
        )

    def callback(self):
        if 'code' not in request.args:
            return None, None, None
        oauth_session = self.service.get_auth_session(
            data={'code': request.args['code'],
                  'grant_type': 'authorization_code',
                  'redirect_uri': self.get_callback_url()}
        )
        me = oauth_session.get('me').json()
        return (
            'facebook$' + me['id'],
            me.get('email').split('@')[0],  # Facebook does not provide
                                            # username, so the email's user
                                            # is used instead
            me.get('email')
        )


class TwitterSignIn(OAuthSignIn):
    def __init__(self):
        super(TwitterSignIn, self).__init__('twitter')
        self.service = OAuth1Service(
            name='twitter',
            consumer_key=self.consumer_id,
            consumer_secret=self.consumer_secret,
            request_token_url='https://api.twitter.com/oauth/request_token',
            authorize_url='https://api.twitter.com/oauth/authorize',
            access_token_url='https://api.twitter.com/oauth/access_token',
            base_url='https://api.twitter.com/1.1/'
        )

    def authorize(self):
        request_token = self.service.get_request_token(
            params={'oauth_callback': self.get_callback_url()}
        )
        session['request_token'] = request_token
        return redirect(self.service.get_authorize_url(request_token[0]))

    def callback(self):
        request_token = session.pop('request_token')
        if 'oauth_verifier' not in request.args:
            return None, None, None
        oauth_session = self.service.get_auth_session(
            request_token[0],
            request_token[1],
            data={'oauth_verifier': request.args['oauth_verifier']}
        )
        me = oauth_session.get('account/verify_credentials.json').json()
        social_id = 'twitter$' + str(me.get('id'))
        username = me.get('screen_name')
        return social_id, username, None   # Twitter does not provide email

一些附加信息 -

具体错误是这样的:

File "/Users/metersky/code/mylastapt/app/oauth.py", line 29, in get_provider
provider = provider_class()
File "/Users/metersky/code/mylastapt/app/oauth.py", line 73, in __init__
super(TwitterSignIn, self).__init__('twitter')
File "/Users/metersky/code/mylastapt/app/oauth.py", line 10, in __init__
credentials = current_app.config['OAUTH_CREDENTIALS'][provider_name]
KeyError: 'twitter'

这就是我认为问题可能发生的地方:

app.config['OAUTH_CREDENTIALS'] = {
    'facebook': {
        'id': 'XXX',
        'secret': 'XXXX'
    }
}

Answer 1

问题出在 OAuthSignIn.get_provider 中。

@classmethod
def get_provider(self, provider_name):
    if self.providers is None:
        self.providers = {}
        for provider_class in self.__subclasses__():
            provider = provider_class()
            self.providers[provider.provider_name] = provider
    return self.providers[provider_name]

第一次从 View 中调用它时

oauth = OAuthSignIn.get_provider(provider)

该方法会缓存您定义的提供程序。它通过检查 OAuthSignIn 的所有子类来实现此目的。

for provider_class in self.__subclasses__():

当您包含 TwitterSignIn 时，它将作为子类包含在内。然后您将实例化该类的实例

provider = provider_class()

在 OAuthSignIn.__init__ 内，您可以使用 current_app.config['OAUTH_CREDENTIALS'][provider_name] 加载提供程序的设置。由于不包含 Twitter，因此您会收到 KeyError。

如果您不想支持 Twitter，只需删除该类(class)即可。如果您想进一步保护您的应用程序，以便可以在不更新代码的情况下从您的设置中删除提供程序，则您需要检查异常。您可以在 OAuthSignIn.__init__ 内部进行检查，但在 OAuthSignIn.providers 中包含不受支持的提供程序可能没有太大值(value)。您最好将检查放入 OAuthSignIn.get_provider 中。

@classmethod
def get_provider(cls, provider_name):
    if cls.providers is None:
        cls.providers = {}
        for provider_class in cls.__subclassess__():
            try:
                provider = provider_class()
            except KeyError:
                pass  # unsupported provider
            else:
                cls.providers[provider.provider_name] = provider
    return cls.providers[provider_name]