<commit_before># Copyright (c) <NAME>
# See LICENSE for details.
from functools import wraps
from twisted.web.server import NOT_DONE_YET

from oauth2.errors import InvalidTokenError
from oauth2.resource import OAuth2
from oauth2.token import TokenResource


def _getToken(request):
    """
    Helper method to get a token from a request, if it contains any.
    :param request: The request.
    :return: A token that was send with the request or None.
    """
    return request.getHeader(b'Authorization')


def isAuthorized(request, scope, allowInsecureRequestDebug=False):
    """
    Returns True if the token in the request grants access to the given
    scope. The token is validated via the authTokenStorage singleton
    given to the TokenResource instance. If the token is invalid,
    does not grant access to the scope or was not send via a secure
    protocol, False is returned, an error is written to the request
    and the request is closed.
    You can not write to the request if this function returned False!
    :param request: The request.
    :param scope: The scope or list of scopes the token must grant access to.
    :param allowInsecureRequestDebug: Allow requests to originate from
           insecure connections. Only use for local testing!
    :return: True, if the request is authorized, False otherwise.
    """
    if allowInsecureRequestDebug or request.isSecure():
        requestToken = _getToken(request)
        if requestToken is not None and requestToken.startswith(b'Bearer '):
            requestToken = requestToken[7:].decode('utf-8')
            scope = scope if type(scope) == list else [scope]
            if TokenResource.getTokenStorageSingleton().contains(requestToken, scope):
                return True
    request.write(InvalidTokenError('auth token').generate(request))
    request.finish()
    return False


def oauth2(scope, allowInsecureRequestDebug=False):
    """
    Function decorator that checks the first argument, which must
    be a request object, with isAuthorized.
    If the request is authorized, the function is called,
    otherwise the request is closed and NOT_DONE_YET is returned.
    :param scope: The scope or list of scopes the token must grant access to.
    :param allowInsecureRequestDebug: Allow requests to originate from
           insecure connections. Only use for local testing!
    :return: The wrapped function.<commit_msg>Return 401 when a request is not authorized and populate the WWW-Authenticate header<commit_after># Copyright (c) <NAME>
# See LICENSE for details.
from functools import wraps
from twisted.web.server import NOT_DONE_YET

from oauth2.errors import MissingTokenError, InvalidTokenRequestError
from oauth2.resource import OAuth2
from oauth2.token import TokenResource


def _getToken(request):
    """
    Helper method to get a token from a request, if it contains any.
    :param request: The request.
    :return: A token that was send with the request or None.
    """
    return request.getHeader(b'Authorization')


def isAuthorized(request, scope, allowInsecureRequestDebug=False):
    """
    Returns True if the token in the request grants access to the given
    scope. The token is validated via the authTokenStorage singleton
    given to the TokenResource instance. If the token is invalid,
    does not grant access to the scope or was not send via a secure
    protocol, False is returned, an error is written to the request
    and the request is closed.
    You can not write to the request if this function returned False!
    :param request: The request.
    :param scope: The scope or list of scopes the token must grant access to.
    :param allowInsecureRequestDebug: Allow requests to originate from
           insecure connections. Only use for local testing!
    :return: True, if the request is authorized, False otherwise.
    """
    error = None
    scope = scope if type(scope) == list else [scope]
    if allowInsecureRequestDebug or request.isSecure():
        requestToken = _getToken(request)
        if requestToken is None:
            error = MissingTokenError(scope)
        else:
            if requestToken.startswith(b'Bearer '):
                try:
                    requestToken = requestToken[7:].decode('utf-8')
                except UnicodeDecodeError:
                    pass
                else:
                    if TokenResource.getTokenStorageSingleton().contains(requestToken, scope):
                        return True
            error = InvalidTokenRequestError(scope)
    request.write(error.generate(request))
    request.finish()
    return False


def oauth2(scope, allowInsecureRequestDebug=False):
    """
    Function decorator that checks the first argument, which must
    be a request object, with isAuthorized.
    If the request is authorized, the function is called,
    otherwise the request is closed and NOT_DONE_YET is returned.
    :param scope: The scope or list of scopes the token must grant access to.
    :param allowInsecureRequestDebug: Allow requests to originate from
           insecure connections. Only use for local testing!
    :return: The wrapped function.
{'fraction_non_alphanumeric': 0.05843653250773994, 'fraction_numerical': 0.003289473684210526, 'mean_word_length': 3.1618357487922704, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4509278', 'n_tokens_mistral': 1312, 'n_tokens_neox': 1278, 'n_words': 580}