Source code for oauthlib.oauth2.rfc6749.tokens


This module contains methods for adding two types of access tokens to requests.

- Bearer
import hashlib
import hmac
import warnings
from binascii import b2a_base64
from urllib.parse import urlparse

from oauthlib import common
from oauthlib.common import add_params_to_qs, add_params_to_uri

from . import utils

class OAuth2Token(dict):

    def __init__(self, params, old_scope=None):
        self._new_scope = None
        if 'scope' in params and params['scope']:
            self._new_scope = set(utils.scope_to_list(params['scope']))
        if old_scope is not None:
            self._old_scope = set(utils.scope_to_list(old_scope))
            if self._new_scope is None:
                # the rfc says that if the scope hasn't changed, it's optional
                # in params so set the new scope to the old scope
                self._new_scope = self._old_scope
            self._old_scope = self._new_scope

    def scope_changed(self):
        return self._new_scope != self._old_scope

    def old_scope(self):
        return utils.list_to_scope(self._old_scope)

    def old_scopes(self):
        return list(self._old_scope)

    def scope(self):
        return utils.list_to_scope(self._new_scope)

    def scopes(self):
        return list(self._new_scope)

    def missing_scopes(self):
        return list(self._old_scope - self._new_scope)

    def additional_scopes(self):
        return list(self._new_scope - self._old_scope)

def prepare_mac_header(token, uri, key, http_method,
    """Add an `MAC Access Authentication`_ signature to headers.

    Unlike OAuth 1, this HMAC signature does not require inclusion of the
    request payload/body, neither does it use a combination of client_secret
    and token_secret but rather a mac_key provided together with the access

    Currently two algorithms are supported, "hmac-sha-1" and "hmac-sha-256",
    `extension algorithms`_ are not supported.

    Example MAC Authorization header, linebreaks added for clarity

    Authorization: MAC id="h480djs93hd8",

    .. _`MAC Access Authentication`:
    .. _`extension algorithms`:

    :param token:
    :param uri: Request URI.
    :param key: MAC given provided by token endpoint.
    :param http_method: HTTP Request method.
    :param nonce:
    :param headers: Request headers as a dictionary.
    :param body:
    :param ext:
    :param hash_algorithm: HMAC algorithm provided by token endpoint.
    :param issue_time: Time when the MAC credentials were issued (datetime).
    :param draft: MAC authentication specification version.
    :return: headers dictionary with the authorization field added.
    http_method = http_method.upper()
    host, port = utils.host_from_uri(uri)

    if hash_algorithm.lower() == 'hmac-sha-1':
        h = hashlib.sha1
    elif hash_algorithm.lower() == 'hmac-sha-256':
        h = hashlib.sha256
        raise ValueError('unknown hash algorithm')

    if draft == 0:
        nonce = nonce or '{}:{}'.format(utils.generate_age(issue_time),
        ts = common.generate_timestamp()
        nonce = common.generate_nonce()

    sch, net, path, par, query, fra = urlparse(uri)

    request_uri = path + '?' + query if query else path

    # Hash the body/payload
    if body is not None and draft == 0:
        body = body.encode('utf-8')
        bodyhash = b2a_base64(h(body).digest())[:-1].decode('utf-8')
        bodyhash = ''

    # Create the normalized base string
    base = []
    if draft == 0:
    if draft == 0:
    base.append(ext or '')
    base_string = '\n'.join(base) + '\n'

    # hmac struggles with unicode strings -
    if isinstance(key, str):
        key = key.encode('utf-8')
    sign =, base_string.encode('utf-8'), h)
    sign = b2a_base64(sign.digest())[:-1].decode('utf-8')

    header = []
    header.append('MAC id="%s"' % token)
    if draft != 0:
        header.append('ts="%s"' % ts)
    header.append('nonce="%s"' % nonce)
    if bodyhash:
        header.append('bodyhash="%s"' % bodyhash)
    if ext:
        header.append('ext="%s"' % ext)
    header.append('mac="%s"' % sign)

    headers = headers or {}
    headers['Authorization'] = ', '.join(header)
    return headers

def prepare_bearer_uri(token, uri):
    """Add a `Bearer Token`_ to the request URI.
    Not recommended, use only if client can't use authorization header or body.

    .. _`Bearer Token`:

    :param token:
    :param uri:
    return add_params_to_uri(uri, [(('access_token', token))])

def prepare_bearer_headers(token, headers=None):
    """Add a `Bearer Token`_ to the request URI.
    Recommended method of passing bearer tokens.

    Authorization: Bearer h480djs93hd8

    .. _`Bearer Token`:

    :param token:
    :param headers:
    headers = headers or {}
    headers['Authorization'] = 'Bearer %s' % token
    return headers

def prepare_bearer_body(token, body=''):
    """Add a `Bearer Token`_ to the request body.


    .. _`Bearer Token`:

    :param token:
    :param body:
    return add_params_to_qs(body, [(('access_token', token))])

def random_token_generator(request, refresh_token=False):
    :param request: OAuthlib request.
    :type request: oauthlib.common.Request
    :param refresh_token:
    return common.generate_token()

def signed_token_generator(private_pem, **kwargs):
    :param private_pem:
    def signed_token_generator(request): = kwargs
        return common.generate_signed_token(private_pem, request)

    return signed_token_generator

def get_token_from_header(request):
    Helper function to extract a token from the request header.

    :param request: OAuthlib request.
    :type request: oauthlib.common.Request
    :return: Return the token or None if the Authorization header is malformed.
    token = None

    if 'Authorization' in request.headers:
        split_header = request.headers.get('Authorization').split()
        if len(split_header) == 2 and split_header[0].lower() == 'bearer':
            token = split_header[1]
        token = request.access_token

    return token

class TokenBase:
    __slots__ = ()

    def __call__(self, request, refresh_token=False):
        raise NotImplementedError('Subclasses must implement this method.')

    def validate_request(self, request):
        :param request: OAuthlib request.
        :type request: oauthlib.common.Request
        raise NotImplementedError('Subclasses must implement this method.')

    def estimate_type(self, request):
        :param request: OAuthlib request.
        :type request: oauthlib.common.Request
        raise NotImplementedError('Subclasses must implement this method.')

[docs]class BearerToken(TokenBase): __slots__ = ( 'request_validator', 'token_generator', 'refresh_token_generator', 'expires_in' ) def __init__(self, request_validator=None, token_generator=None, expires_in=None, refresh_token_generator=None): self.request_validator = request_validator self.token_generator = token_generator or random_token_generator self.refresh_token_generator = ( refresh_token_generator or self.token_generator ) self.expires_in = expires_in or 3600
[docs] def create_token(self, request, refresh_token=False, **kwargs): """ Create a BearerToken, by default without refresh token. :param request: OAuthlib request. :type request: oauthlib.common.Request :param refresh_token: """ if "save_token" in kwargs: warnings.warn("`save_token` has been deprecated, it was not called internally." "If you do, call `request_validator.save_token()` instead.", DeprecationWarning) expires_in = self.expires_in(request) if callable(self.expires_in) else self.expires_in request.expires_in = expires_in token = { 'access_token': self.token_generator(request), 'expires_in': expires_in, 'token_type': 'Bearer', } # If provided, include - this is optional in some cases but # there is currently no mechanism to coordinate issuing a token for only a subset of the requested scopes so # all tokens issued are for the entire set of requested scopes. if request.scopes is not None: token['scope'] = ' '.join(request.scopes) if refresh_token: if (request.refresh_token and not self.request_validator.rotate_refresh_token(request)): token['refresh_token'] = request.refresh_token else: token['refresh_token'] = self.refresh_token_generator(request) token.update(request.extra_credentials or {}) return OAuth2Token(token)
[docs] def validate_request(self, request): """ :param request: OAuthlib request. :type request: oauthlib.common.Request """ token = get_token_from_header(request) return self.request_validator.validate_bearer_token( token, request.scopes, request)
[docs] def estimate_type(self, request): """ :param request: OAuthlib request. :type request: oauthlib.common.Request """ if request.headers.get('Authorization', '').split(' ')[0].lower() == 'bearer': return 9 elif request.access_token is not None: return 5 else: return 0