626 lines
24 KiB
Python
626 lines
24 KiB
Python
import hashlib
|
|
|
|
__all__ = [
|
|
'BadRequestException',
|
|
'BadStateException',
|
|
'CsrfException',
|
|
'DropboxOAuth2Flow',
|
|
'DropboxOAuth2FlowNoRedirect',
|
|
'NotApprovedException',
|
|
'OAuth2FlowNoRedirectResult',
|
|
'OAuth2FlowResult',
|
|
'ProviderException',
|
|
]
|
|
|
|
import base64
|
|
import os
|
|
import six
|
|
import urllib
|
|
import re
|
|
from datetime import datetime, timedelta
|
|
|
|
from .session import (
|
|
API_HOST,
|
|
WEB_HOST,
|
|
pinned_session,
|
|
DEFAULT_TIMEOUT,
|
|
)
|
|
|
|
if six.PY3:
|
|
url_path_quote = urllib.parse.quote # pylint: disable=no-member,useless-suppression
|
|
url_encode = urllib.parse.urlencode # pylint: disable=no-member,useless-suppression
|
|
else:
|
|
url_path_quote = urllib.quote # pylint: disable=no-member,useless-suppression
|
|
url_encode = urllib.urlencode # pylint: disable=no-member,useless-suppression
|
|
|
|
TOKEN_ACCESS_TYPES = ['offline', 'online', 'legacy']
|
|
INCLUDE_GRANTED_SCOPES_TYPES = ['user', 'team']
|
|
PKCE_VERIFIER_LENGTH = 128
|
|
|
|
class OAuth2FlowNoRedirectResult(object):
|
|
"""
|
|
Authorization information for an OAuth2Flow performed with no redirect.
|
|
"""
|
|
|
|
def __init__(self, access_token, account_id, user_id, refresh_token, expiration, scope):
|
|
"""
|
|
:param str access_token: Token to be used to authenticate later requests.
|
|
:param str account_id: The Dropbox user's account ID.
|
|
:param str user_id: Deprecated (use :attr:`account_id` instead).
|
|
:param str refresh_token: Token to be used to acquire new access token when existing one
|
|
expires.
|
|
:param expiration: Either the number of seconds from now that the token expires in or the
|
|
datetime at which the token expires.
|
|
:type expiration: int or datetime
|
|
:param list scope: List of scopes to request in base oauth flow.
|
|
"""
|
|
self.access_token = access_token
|
|
if not expiration:
|
|
self.expires_at = None
|
|
elif isinstance(expiration, datetime):
|
|
self.expires_at = expiration
|
|
else:
|
|
self.expires_at = datetime.utcnow() + timedelta(seconds=int(expiration))
|
|
self.refresh_token = refresh_token
|
|
self.account_id = account_id
|
|
self.user_id = user_id
|
|
self.scope = scope
|
|
|
|
def __repr__(self):
|
|
return 'OAuth2FlowNoRedirectResult(%s, %s, %s, %s, %s, %s)' % (
|
|
self.access_token,
|
|
self.account_id,
|
|
self.user_id,
|
|
self.refresh_token,
|
|
self.expires_at,
|
|
self.scope,
|
|
)
|
|
|
|
|
|
class OAuth2FlowResult(OAuth2FlowNoRedirectResult):
|
|
"""
|
|
Authorization information for an :class:`OAuth2Flow` with redirect.
|
|
"""
|
|
|
|
def __init__(self, access_token, account_id, user_id, url_state, refresh_token,
|
|
expires_in, scope):
|
|
"""
|
|
Same as :class:`OAuth2FlowNoRedirectResult` but with url_state.
|
|
|
|
:param str url_state: The url state that was set by :meth:`DropboxOAuth2Flow.start`.
|
|
"""
|
|
super(OAuth2FlowResult, self).__init__(
|
|
access_token=access_token,
|
|
account_id=account_id,
|
|
user_id=user_id,
|
|
refresh_token=refresh_token,
|
|
expiration=expires_in,
|
|
scope=scope)
|
|
self.url_state = url_state
|
|
|
|
@classmethod
|
|
def from_no_redirect_result(cls, result, url_state):
|
|
assert isinstance(result, OAuth2FlowNoRedirectResult)
|
|
return cls(result.access_token, result.account_id, result.user_id,
|
|
url_state, result.refresh_token, result.expires_at, result.scope)
|
|
|
|
def __repr__(self):
|
|
return 'OAuth2FlowResult(%s, %s, %s, %s, %s, %s, %s)' % (
|
|
self.access_token,
|
|
self.account_id,
|
|
self.user_id,
|
|
self.url_state,
|
|
self.refresh_token,
|
|
self.expires_at,
|
|
self.scope,
|
|
)
|
|
|
|
|
|
class DropboxOAuth2FlowBase(object):
|
|
|
|
def __init__(self, consumer_key, consumer_secret=None, locale=None, token_access_type=None,
|
|
scope=None, include_granted_scopes=None, use_pkce=False, timeout=DEFAULT_TIMEOUT):
|
|
if scope is not None and (len(scope) == 0 or not isinstance(scope, list)):
|
|
raise BadInputException("Scope list must be of type list")
|
|
if token_access_type is not None and token_access_type not in TOKEN_ACCESS_TYPES:
|
|
raise BadInputException("Token access type must be from the following enum: {}".format(
|
|
TOKEN_ACCESS_TYPES))
|
|
if not (use_pkce or consumer_secret):
|
|
raise BadInputException("Must pass in either consumer secret or use PKCE")
|
|
if include_granted_scopes and not scope:
|
|
raise BadInputException("Must pass in scope to pass include_granted_scopes")
|
|
|
|
self.consumer_key = consumer_key
|
|
self.consumer_secret = consumer_secret
|
|
self.locale = locale
|
|
self.token_access_type = token_access_type
|
|
self.requests_session = pinned_session()
|
|
self.scope = scope
|
|
self.include_granted_scopes = include_granted_scopes
|
|
self._timeout = timeout
|
|
|
|
if use_pkce:
|
|
self.code_verifier = _generate_pkce_code_verifier()
|
|
self.code_challenge = _generate_pkce_code_challenge(self.code_verifier)
|
|
else:
|
|
self.code_verifier = None
|
|
self.code_challenge = None
|
|
|
|
def _get_authorize_url(self, redirect_uri, state, token_access_type=None, scope=None,
|
|
include_granted_scopes=None, code_challenge=None):
|
|
params = dict(response_type='code',
|
|
client_id=self.consumer_key)
|
|
if redirect_uri is not None:
|
|
params['redirect_uri'] = redirect_uri
|
|
if state is not None:
|
|
params['state'] = state
|
|
if token_access_type is not None:
|
|
assert token_access_type in TOKEN_ACCESS_TYPES
|
|
params['token_access_type'] = token_access_type
|
|
if code_challenge:
|
|
params['code_challenge'] = code_challenge
|
|
params['code_challenge_method'] = 'S256'
|
|
|
|
if scope is not None:
|
|
params['scope'] = " ".join(scope)
|
|
if include_granted_scopes is not None:
|
|
assert include_granted_scopes in INCLUDE_GRANTED_SCOPES_TYPES
|
|
params['include_granted_scopes'] = include_granted_scopes
|
|
|
|
return self.build_url('/oauth2/authorize', params, WEB_HOST)
|
|
|
|
def _finish(self, code, redirect_uri, code_verifier):
|
|
url = self.build_url('/oauth2/token')
|
|
params = {'grant_type': 'authorization_code',
|
|
'code': code,
|
|
'client_id': self.consumer_key,
|
|
}
|
|
if code_verifier:
|
|
params['code_verifier'] = code_verifier
|
|
else:
|
|
params['client_secret'] = self.consumer_secret
|
|
if self.locale is not None:
|
|
params['locale'] = self.locale
|
|
if redirect_uri is not None:
|
|
params['redirect_uri'] = redirect_uri
|
|
|
|
resp = self.requests_session.post(url, data=params, timeout=self._timeout)
|
|
resp.raise_for_status()
|
|
|
|
d = resp.json()
|
|
|
|
if 'team_id' in d:
|
|
account_id = d['team_id']
|
|
else:
|
|
account_id = d['account_id']
|
|
|
|
access_token = d['access_token']
|
|
|
|
if 'refresh_token' in d:
|
|
refresh_token = d['refresh_token']
|
|
else:
|
|
refresh_token = ""
|
|
|
|
if 'expires_in' in d:
|
|
expires_in = d['expires_in']
|
|
else:
|
|
expires_in = None
|
|
|
|
if 'scope' in d:
|
|
scope = d['scope']
|
|
else:
|
|
scope = None
|
|
|
|
uid = d['uid']
|
|
|
|
return OAuth2FlowNoRedirectResult(
|
|
access_token,
|
|
account_id,
|
|
uid,
|
|
refresh_token,
|
|
expires_in,
|
|
scope)
|
|
|
|
def build_path(self, target, params=None):
|
|
"""Build the path component for an API URL.
|
|
|
|
This method urlencodes the parameters, adds them to the end of the target url, and puts a
|
|
marker for the API version in front.
|
|
|
|
:param str target: A target url (e.g. '/files') to build upon.
|
|
:param dict params: Optional dictionary of parameters (name to value).
|
|
:return: The path and parameters components of an API URL.
|
|
:rtype: str
|
|
"""
|
|
if six.PY2 and isinstance(target, six.text_type):
|
|
target = target.encode('utf8')
|
|
|
|
target_path = url_path_quote(target)
|
|
|
|
params = params or {}
|
|
params = params.copy()
|
|
|
|
if self.locale:
|
|
params['locale'] = self.locale
|
|
|
|
if params:
|
|
query_string = _params_to_urlencoded(params)
|
|
return "%s?%s" % (target_path, query_string)
|
|
else:
|
|
return target_path
|
|
|
|
def build_url(self, target, params=None, host=API_HOST):
|
|
"""Build an API URL.
|
|
|
|
This method adds scheme and hostname to the path returned from build_path.
|
|
|
|
:param str target: A target url (e.g. '/files') to build upon.
|
|
:param dict params: Optional dictionary of parameters (name to value).
|
|
:return: The full API URL.
|
|
:rtype: str
|
|
"""
|
|
return "https://%s%s" % (host, self.build_path(target, params))
|
|
|
|
|
|
class DropboxOAuth2FlowNoRedirect(DropboxOAuth2FlowBase):
|
|
"""
|
|
OAuth 2 authorization helper for apps that can't provide a redirect URI
|
|
(such as the command-line example apps).
|
|
|
|
See examples under `example/oauth <https://github.com/dropbox/dropbox-sdk-python/tree/main/
|
|
example/oauth>`_
|
|
|
|
"""
|
|
|
|
def __init__(self, consumer_key, consumer_secret=None, locale=None, token_access_type=None,
|
|
scope=None, include_granted_scopes=None, use_pkce=False, timeout=DEFAULT_TIMEOUT): # noqa: E501;
|
|
"""
|
|
Construct an instance.
|
|
|
|
:param str consumer_key: Your API app's "app key".
|
|
:param str consumer_secret: Your API app's "app secret".
|
|
:param str locale: The locale of the user of your application. For example "en" or "en_US".
|
|
Some API calls return localized data and error messages; this setting tells the server
|
|
which locale to use. By default, the server uses "en_US".
|
|
:param str token_access_type: the type of token to be requested.
|
|
From the following enum:
|
|
|
|
* None - creates a token with the app default (either legacy or online)
|
|
* legacy - creates one long-lived token with no expiration
|
|
* online - create one short-lived token with an expiration
|
|
* offline - create one short-lived token with an expiration with a refresh token
|
|
|
|
:param list scope: list of scopes to request in base oauth flow.
|
|
If left blank, will default to all scopes for app.
|
|
:param str include_granted_scopes: which scopes to include from previous grants.
|
|
From the following enum:
|
|
|
|
* user - include user scopes in the grant
|
|
* team - include team scopes in the grant
|
|
* *Note*: if this user has never linked the app, include_granted_scopes must be None
|
|
|
|
:param bool use_pkce: Whether or not to use Sha256 based PKCE. PKCE should be only use on
|
|
client apps which doesn't call your server. It is less secure than non-PKCE flow but
|
|
can be used if you are unable to safely retrieve your app secret.
|
|
:param Optional[float] timeout: Maximum duration in seconds that
|
|
client will wait for any single packet from the server. After the timeout the client
|
|
will give up on connection. If `None`, client will wait forever. Defaults
|
|
to 100 seconds.
|
|
"""
|
|
super(DropboxOAuth2FlowNoRedirect, self).__init__(
|
|
consumer_key=consumer_key,
|
|
consumer_secret=consumer_secret,
|
|
locale=locale,
|
|
token_access_type=token_access_type,
|
|
scope=scope,
|
|
include_granted_scopes=include_granted_scopes,
|
|
use_pkce=use_pkce,
|
|
timeout=timeout
|
|
)
|
|
|
|
def start(self):
|
|
"""
|
|
Starts the OAuth 2 authorization process.
|
|
|
|
:return: The URL for a page on Dropbox's website. This page will let the user "approve"
|
|
your app, which gives your app permission to access the user's Dropbox account.
|
|
Tell the user to visit this URL and approve your app.
|
|
"""
|
|
return self._get_authorize_url(None, None, self.token_access_type,
|
|
scope=self.scope,
|
|
include_granted_scopes=self.include_granted_scopes,
|
|
code_challenge=self.code_challenge)
|
|
|
|
def finish(self, code):
|
|
"""
|
|
If the user approves your app, they will be presented with an "authorization code".
|
|
Have the user copy/paste that authorization code into your app and then call this method to
|
|
get an access token.
|
|
|
|
:param str code: The authorization code shown to the user when they
|
|
approved your app.
|
|
:rtype: :class:`OAuth2FlowNoRedirectResult`
|
|
:raises: The same exceptions as :meth:`DropboxOAuth2Flow.finish()`.
|
|
"""
|
|
return self._finish(code, None, self.code_verifier)
|
|
|
|
|
|
class DropboxOAuth2Flow(DropboxOAuth2FlowBase):
|
|
"""
|
|
OAuth 2 authorization helper. Use this for web apps.
|
|
|
|
OAuth 2 has a two-step authorization process. The first step is having the user authorize your
|
|
app. The second involves getting an OAuth 2 access token from Dropbox.
|
|
|
|
See examples under `example/oauth <https://github.com/dropbox/dropbox-sdk-python/tree/main/
|
|
example/oauth>`_
|
|
|
|
"""
|
|
|
|
def __init__(self, consumer_key, redirect_uri, session,
|
|
csrf_token_session_key, consumer_secret=None, locale=None,
|
|
token_access_type=None, scope=None,
|
|
include_granted_scopes=None, use_pkce=False, timeout=DEFAULT_TIMEOUT):
|
|
"""
|
|
Construct an instance.
|
|
|
|
:param str consumer_key: Your API app's "app key".
|
|
:param str consumer_secret: Your API app's "app secret".
|
|
:param str redirect_uri: The URI that the Dropbox server will redirect the user to after the
|
|
user finishes authorizing your app. This URI must be HTTPS-based and pre-registered
|
|
with the Dropbox servers, though localhost URIs are allowed without pre-registration and
|
|
can be either HTTP or HTTPS.
|
|
:param dict session: A dict-like object that represents the current user's web session
|
|
(Will be used to save the CSRF token).
|
|
:param str csrf_token_session_key: The key to use when storing the CSRF token in the session
|
|
(For example: "dropbox-auth-csrf-token").
|
|
:param str locale: The locale of the user of your application. For example "en" or "en_US".
|
|
Some API calls return localized data and error messages; this setting tells the server
|
|
which locale to use. By default, the server uses "en_US".
|
|
:param str token_access_type: The type of token to be requested.
|
|
From the following enum:
|
|
|
|
* None - creates a token with the app default (either legacy or online)
|
|
* legacy - creates one long-lived token with no expiration
|
|
* online - create one short-lived token with an expiration
|
|
* offline - create one short-lived token with an expiration with a refresh token
|
|
|
|
:param list scope: List of scopes to request in base oauth flow. If left blank,
|
|
will default to all scopes for app.
|
|
:param str include_granted_scopes: Which scopes to include from previous grants.
|
|
From the following enum:
|
|
|
|
* user - include user scopes in the grant
|
|
* team - include team scopes in the grant
|
|
* *Note*: If this user has never linked the app, :attr:`include_granted_scopes` must \
|
|
be `None`
|
|
|
|
:param bool use_pkce: Whether or not to use Sha256 based PKCE
|
|
:param Optional[float] timeout: Maximum duration in seconds that client will wait for any
|
|
single packet from the server. After the timeout the client will give up on connection.
|
|
If `None`, client will wait forever. Defaults to 100 seconds.
|
|
"""
|
|
|
|
super(DropboxOAuth2Flow, self).__init__(
|
|
consumer_key=consumer_key,
|
|
consumer_secret=consumer_secret,
|
|
locale=locale,
|
|
token_access_type=token_access_type,
|
|
scope=scope,
|
|
include_granted_scopes=include_granted_scopes,
|
|
use_pkce=use_pkce,
|
|
timeout=timeout
|
|
)
|
|
self.redirect_uri = redirect_uri
|
|
self.session = session
|
|
self.csrf_token_session_key = csrf_token_session_key
|
|
|
|
def start(self, url_state=None):
|
|
"""
|
|
Starts the OAuth 2 authorization process.
|
|
|
|
This function builds an "authorization URL". You should redirect your user's browser to this
|
|
URL, which will give them an opportunity to grant your app access to their Dropbox
|
|
account. When the user completes this process, they will be automatically redirected to
|
|
the :attr:`redirect_uri` you passed in to the constructor. This function will also save
|
|
a CSRF token to :attr:`session[csrf_token_session_key]`
|
|
(as provided to the constructor). This CSRF token will be checked on :meth:`finish()`
|
|
to prevent request forgery.
|
|
|
|
:param str url_state: Any data that you would like to keep in the URL through the
|
|
authorization process. This exact value will be returned to you by :meth:`finish()`.
|
|
:return: The URL for a page on Dropbox's website. This page will let the user "approve" your
|
|
app, which gives your app permission to access the user's Dropbox account. Tell the user
|
|
to visit this URL and approve your app.
|
|
"""
|
|
csrf_token = base64.urlsafe_b64encode(os.urandom(16)).decode('ascii')
|
|
state = csrf_token
|
|
if url_state is not None:
|
|
state += "|" + url_state
|
|
self.session[self.csrf_token_session_key] = csrf_token
|
|
|
|
return self._get_authorize_url(self.redirect_uri, state, self.token_access_type,
|
|
scope=self.scope,
|
|
include_granted_scopes=self.include_granted_scopes,
|
|
code_challenge=self.code_challenge)
|
|
|
|
def finish(self, query_params):
|
|
"""
|
|
Call this after the user has visited the authorize URL (see :meth:`start()`), approved your
|
|
app and was redirected to your redirect URI.
|
|
|
|
:param dict query_params: The query parameters on the GET request to your redirect URI.
|
|
:rtype: class:`OAuth2FlowResult`
|
|
:raises: :class:`BadRequestException` If the redirect URL was missing parameters or if the
|
|
given parameters were not valid.
|
|
:raises: :class:`BadStateException` If there's no CSRF token in the session.
|
|
:raises: :class:`CsrfException` If the :attr:`state` query parameter doesn't contain the
|
|
CSRF token from the user's session.
|
|
:raises: :class:`NotApprovedException` If the user chose not to approve your app.
|
|
:raises: :class:`ProviderException` If Dropbox redirected to your redirect URI with some
|
|
unexpected error identifier and error message.
|
|
"""
|
|
# Check well-formedness of request.
|
|
|
|
state = query_params.get('state')
|
|
if state is None:
|
|
raise BadRequestException("Missing query parameter 'state'.")
|
|
|
|
error = query_params.get('error')
|
|
error_description = query_params.get('error_description')
|
|
code = query_params.get('code')
|
|
|
|
if error is not None and code is not None:
|
|
raise BadRequestException(
|
|
"Query parameters 'code' and 'error' are both set; "
|
|
"only one must be set.")
|
|
if error is None and code is None:
|
|
raise BadRequestException(
|
|
"Neither query parameter 'code' or 'error' is set.")
|
|
|
|
# Check CSRF token
|
|
|
|
if self.csrf_token_session_key not in self.session:
|
|
raise BadStateException('Missing CSRF token in session.')
|
|
csrf_token_from_session = self.session[self.csrf_token_session_key]
|
|
if len(csrf_token_from_session) <= 20:
|
|
raise AssertionError('CSRF token unexpectedly short: %r' %
|
|
csrf_token_from_session)
|
|
|
|
split_pos = state.find('|')
|
|
if split_pos < 0:
|
|
given_csrf_token = state
|
|
url_state = None
|
|
else:
|
|
given_csrf_token = state[0:split_pos]
|
|
url_state = state[split_pos + 1:]
|
|
|
|
if not _safe_equals(csrf_token_from_session, given_csrf_token):
|
|
raise CsrfException('expected %r, got %r' %
|
|
(csrf_token_from_session, given_csrf_token))
|
|
|
|
del self.session[self.csrf_token_session_key]
|
|
|
|
# Check for error identifier
|
|
|
|
if error is not None:
|
|
if error == 'access_denied':
|
|
# The user clicked "Deny"
|
|
if error_description is None:
|
|
raise NotApprovedException(
|
|
'No additional description from Dropbox')
|
|
else:
|
|
raise NotApprovedException(
|
|
'Additional description from Dropbox: %s' %
|
|
error_description)
|
|
else:
|
|
# All other errors
|
|
full_message = error
|
|
if error_description is not None:
|
|
full_message += ": " + error_description
|
|
raise ProviderException(full_message)
|
|
|
|
# If everything went ok, make the network call to get an access token.
|
|
|
|
no_redirect_result = self._finish(code, self.redirect_uri, self.code_verifier)
|
|
return OAuth2FlowResult.from_no_redirect_result(
|
|
no_redirect_result, url_state)
|
|
|
|
|
|
class BadRequestException(Exception):
|
|
"""
|
|
Thrown if the redirect URL was missing parameters or if the given parameters were not valid.
|
|
|
|
The recommended action is to show an HTTP 400 error page.
|
|
"""
|
|
pass
|
|
|
|
|
|
class BadStateException(Exception):
|
|
"""
|
|
Thrown if all the parameters are correct, but there's no CSRF token in the session. This
|
|
probably means that the session expired.
|
|
|
|
The recommended action is to redirect the user's browser to try the approval process again.
|
|
"""
|
|
pass
|
|
|
|
|
|
class CsrfException(Exception):
|
|
"""
|
|
Thrown if the given 'state' parameter doesn't contain the CSRF token from the user's session.
|
|
This is blocked to prevent CSRF attacks.
|
|
|
|
The recommended action is to respond with an HTTP 403 error page.
|
|
"""
|
|
pass
|
|
|
|
|
|
class NotApprovedException(Exception):
|
|
"""
|
|
The user chose not to approve your app.
|
|
"""
|
|
pass
|
|
|
|
|
|
class ProviderException(Exception):
|
|
"""
|
|
Dropbox redirected to your redirect URI with some unexpected error identifier and error message.
|
|
|
|
The recommended action is to log the error, tell the user something went wrong, and let them try
|
|
again.
|
|
"""
|
|
pass
|
|
|
|
|
|
class BadInputException(Exception):
|
|
"""
|
|
Thrown if incorrect types/values are used
|
|
|
|
This should only ever be thrown during testing, app should have validation of input prior to
|
|
reaching this point
|
|
"""
|
|
pass
|
|
|
|
|
|
def _safe_equals(a, b):
|
|
if len(a) != len(b):
|
|
return False
|
|
res = 0
|
|
for ca, cb in zip(a, b):
|
|
res |= ord(ca) ^ ord(cb)
|
|
return res == 0
|
|
|
|
|
|
def _params_to_urlencoded(params):
|
|
"""
|
|
Returns a application/x-www-form-urlencoded :class:`str` representing the key/value pairs in
|
|
:attr:`params`.
|
|
|
|
Keys are values are ``str()``'d before calling :meth:`urllib.urlencode`, with the exception of
|
|
unicode objects which are utf8-encoded.
|
|
"""
|
|
def encode(o):
|
|
if isinstance(o, six.binary_type):
|
|
return o
|
|
else:
|
|
if isinstance(o, six.text_type):
|
|
return o.encode('utf-8')
|
|
else:
|
|
return str(o).encode('utf-8')
|
|
|
|
utf8_params = {encode(k): encode(v) for k, v in six.iteritems(params)}
|
|
return url_encode(utf8_params)
|
|
|
|
def _generate_pkce_code_verifier():
|
|
code_verifier = base64.urlsafe_b64encode(os.urandom(PKCE_VERIFIER_LENGTH)).decode('utf-8')
|
|
code_verifier = re.sub('[^a-zA-Z0-9]+', '', code_verifier)
|
|
if len(code_verifier) > PKCE_VERIFIER_LENGTH:
|
|
code_verifier = code_verifier[:128]
|
|
return code_verifier
|
|
|
|
def _generate_pkce_code_challenge(code_verifier):
|
|
code_challenge = hashlib.sha256(code_verifier.encode('utf-8')).digest()
|
|
code_challenge = base64.urlsafe_b64encode(code_challenge).decode('utf-8')
|
|
code_challenge = code_challenge.replace('=', '')
|
|
return code_challenge
|