224 lines
7.5 KiB
Python
224 lines
7.5 KiB
Python
# Tweepy
|
|
# Copyright 2009-2023 Joshua Roesslein
|
|
# See LICENSE for details.
|
|
|
|
import logging
|
|
import warnings
|
|
|
|
import requests
|
|
from requests.auth import AuthBase, HTTPBasicAuth
|
|
from requests_oauthlib import OAuth1, OAuth1Session, OAuth2Session
|
|
|
|
from tweepy.errors import TweepyException
|
|
|
|
WARNING_MESSAGE = """Warning! Due to a Twitter API bug, signin_with_twitter
|
|
and access_type don't always play nice together. Details
|
|
https://dev.twitter.com/discussions/21281"""
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class OAuth1UserHandler:
|
|
"""OAuth 1.0a User Context authentication handler
|
|
|
|
.. versionchanged:: 4.5
|
|
Renamed from :class:`OAuthHandler`
|
|
"""
|
|
|
|
def __init__(self, consumer_key, consumer_secret, access_token=None,
|
|
access_token_secret=None, callback=None):
|
|
if not isinstance(consumer_key, (str, bytes)):
|
|
raise TypeError("Consumer key must be string or bytes, not "
|
|
+ type(consumer_key).__name__)
|
|
if not isinstance(consumer_secret, (str, bytes)):
|
|
raise TypeError("Consumer secret must be string or bytes, not "
|
|
+ type(consumer_secret).__name__)
|
|
|
|
self.consumer_key = consumer_key
|
|
self.consumer_secret = consumer_secret
|
|
self.access_token = access_token
|
|
self.access_token_secret = access_token_secret
|
|
self.callback = callback
|
|
self.username = None
|
|
self.request_token = {}
|
|
self.oauth = OAuth1Session(consumer_key, client_secret=consumer_secret,
|
|
callback_uri=self.callback)
|
|
|
|
def apply_auth(self):
|
|
return OAuth1(
|
|
self.consumer_key, client_secret=self.consumer_secret,
|
|
resource_owner_key=self.access_token,
|
|
resource_owner_secret=self.access_token_secret, decoding=None
|
|
)
|
|
|
|
def _get_oauth_url(self, endpoint):
|
|
return 'https://api.twitter.com/oauth/' + endpoint
|
|
|
|
def _get_request_token(self, access_type=None):
|
|
try:
|
|
url = self._get_oauth_url('request_token')
|
|
if access_type:
|
|
url += f'?x_auth_access_type={access_type}'
|
|
return self.oauth.fetch_request_token(url)
|
|
except Exception as e:
|
|
raise TweepyException(e)
|
|
|
|
def get_authorization_url(self, signin_with_twitter=False,
|
|
access_type=None):
|
|
"""Get the authorization URL to redirect the user to"""
|
|
try:
|
|
if signin_with_twitter:
|
|
url = self._get_oauth_url('authenticate')
|
|
if access_type:
|
|
log.warning(WARNING_MESSAGE)
|
|
else:
|
|
url = self._get_oauth_url('authorize')
|
|
self.request_token = self._get_request_token(
|
|
access_type=access_type
|
|
)
|
|
return self.oauth.authorization_url(url)
|
|
except Exception as e:
|
|
raise TweepyException(e)
|
|
|
|
def get_access_token(self, verifier=None):
|
|
"""After user has authorized the app, get access token and secret with
|
|
verifier
|
|
"""
|
|
try:
|
|
url = self._get_oauth_url('access_token')
|
|
self.oauth = OAuth1Session(
|
|
self.consumer_key, client_secret=self.consumer_secret,
|
|
resource_owner_key=self.request_token['oauth_token'],
|
|
resource_owner_secret=self.request_token['oauth_token_secret'],
|
|
verifier=verifier, callback_uri=self.callback
|
|
)
|
|
resp = self.oauth.fetch_access_token(url)
|
|
self.access_token = resp['oauth_token']
|
|
self.access_token_secret = resp['oauth_token_secret']
|
|
return self.access_token, self.access_token_secret
|
|
except Exception as e:
|
|
raise TweepyException(e)
|
|
|
|
def set_access_token(self, key, secret):
|
|
"""
|
|
.. deprecated:: 4.5
|
|
Set through initialization instead.
|
|
"""
|
|
self.access_token = key
|
|
self.access_token_secret = secret
|
|
|
|
|
|
class OAuthHandler(OAuth1UserHandler):
|
|
"""Alias for :class:`OAuth1UserHandler`
|
|
|
|
.. deprecated:: 4.5
|
|
Use :class:`OAuth1UserHandler` instead.
|
|
"""
|
|
|
|
def __init__(self, consumer_key, consumer_secret, access_token=None,
|
|
access_token_secret=None, callback=None):
|
|
warnings.warn(
|
|
"OAuthHandler is deprecated; use OAuth1UserHandler instead.",
|
|
DeprecationWarning
|
|
)
|
|
super().__init__(consumer_key, consumer_secret, access_token,
|
|
access_token_secret, callback)
|
|
|
|
|
|
class OAuth2AppHandler:
|
|
"""OAuth 2.0 Bearer Token (App-Only) using API / Consumer key and secret
|
|
authentication handler
|
|
|
|
.. versionchanged:: 4.5
|
|
Renamed from :class:`AppAuthHandler`
|
|
"""
|
|
|
|
def __init__(self, consumer_key, consumer_secret):
|
|
self.consumer_key = consumer_key
|
|
self.consumer_secret = consumer_secret
|
|
self._bearer_token = ''
|
|
|
|
resp = requests.post(
|
|
'https://api.twitter.com/oauth2/token',
|
|
auth=(self.consumer_key, self.consumer_secret),
|
|
data={'grant_type': 'client_credentials'}
|
|
)
|
|
data = resp.json()
|
|
if data.get('token_type') != 'bearer':
|
|
raise TweepyException('Expected token_type to equal "bearer", '
|
|
f'but got {data.get("token_type")} instead')
|
|
|
|
self._bearer_token = data['access_token']
|
|
|
|
def apply_auth(self):
|
|
return OAuth2BearerHandler(self._bearer_token)
|
|
|
|
|
|
class AppAuthHandler(OAuth2AppHandler):
|
|
"""Alias for :class:`OAuth2AppHandler`
|
|
|
|
.. deprecated:: 4.5
|
|
Use :class:`OAuth2AppHandler` instead.
|
|
"""
|
|
|
|
def __init__(self, consumer_key, consumer_secret):
|
|
warnings.warn(
|
|
"AppAuthHandler is deprecated; use OAuth2AppHandler instead.",
|
|
DeprecationWarning
|
|
)
|
|
super().__init__(consumer_key, consumer_secret)
|
|
|
|
|
|
class OAuth2BearerHandler(AuthBase):
|
|
"""OAuth 2.0 Bearer Token (App-Only) authentication handler
|
|
|
|
.. versionadded:: 4.5
|
|
"""
|
|
|
|
def __init__(self, bearer_token):
|
|
self.bearer_token = bearer_token
|
|
|
|
def __call__(self, request):
|
|
request.headers['Authorization'] = 'Bearer ' + self.bearer_token
|
|
return request
|
|
|
|
def apply_auth(self):
|
|
return self
|
|
|
|
|
|
class OAuth2UserHandler(OAuth2Session):
|
|
"""OAuth 2.0 Authorization Code Flow with PKCE (User Context)
|
|
authentication handler
|
|
|
|
.. versionadded:: 4.5
|
|
"""
|
|
|
|
def __init__(self, *, client_id, redirect_uri, scope, client_secret=None):
|
|
super().__init__(client_id, redirect_uri=redirect_uri, scope=scope)
|
|
if client_secret is not None:
|
|
self.auth = HTTPBasicAuth(client_id, client_secret)
|
|
else:
|
|
self.auth = None
|
|
|
|
def get_authorization_url(self):
|
|
"""Get the authorization URL to redirect the user to"""
|
|
authorization_url, state = self.authorization_url(
|
|
"https://twitter.com/i/oauth2/authorize",
|
|
code_challenge=self._client.create_code_challenge(
|
|
self._client.create_code_verifier(128), "S256"
|
|
), code_challenge_method="S256"
|
|
)
|
|
return authorization_url
|
|
|
|
def fetch_token(self, authorization_response):
|
|
"""After user has authorized the app, fetch access token with
|
|
authorization response URL
|
|
"""
|
|
return super().fetch_token(
|
|
"https://api.twitter.com/2/oauth2/token",
|
|
authorization_response=authorization_response,
|
|
auth=self.auth,
|
|
include_client_id=True,
|
|
code_verifier=self._client.code_verifier
|
|
)
|