HEX
Server: Apache/2.4.52 (Ubuntu)
System: Linux spn-python 5.15.0-89-generic #99-Ubuntu SMP Mon Oct 30 20:42:41 UTC 2023 x86_64
User: arjun (1000)
PHP: 8.1.2-1ubuntu2.20
Disabled: NONE
Upload Files
File: //home/arjun/projects/buyercall_forms/buyercall/buyercall/lib/util_rest.py
from functools import wraps

from flask import (
    make_response,
    request,
    jsonify,
    current_app as app)
from werkzeug.local import Local, LocalProxy


_local = Local()
rest_partnership_account = LocalProxy(lambda: _local.partnership_account)
rest_is_partnership_account = LocalProxy(lambda: _local.is_partnership_account)
rest_is_partnership = LocalProxy(lambda: _local.is_partnership)
rest_partnership = LocalProxy(lambda: _local.partnership)
rest_api_token_id = LocalProxy(lambda: _local.api_token_id)

blocked_debug_request_methods = ['DELETE', 'POST', 'PUT']


def rest_method(wrapped):
    def make_error(text, code):
        response = make_response(text + '\r\n', 401)
        response.headers['WWW-Authenticate'] = (
            'Bearer realm="buyercall", error="{}", '
            'error_description="{}"'
        ).format(code, text)
        return response

    def wrapper(*args, **kwargs):
        from buyercall.blueprints.partnership.models import PartnershipAccount, ApiToken, Partnership

        header = request.headers.get('Authorize', '').split(' ')
        if len(header) != 2 or header[0] != 'Bearer':
            return make_error(
                'Invalid authorization method', 'invalid_request')

        token = header[1].strip()
        api_token_id = ApiToken.check_token(token)

        if api_token_id:
            partnership_account = PartnershipAccount.get_partnership_account_by_token(api_token_id)
            _local.api_token_id = api_token_id

            if partnership_account:
                _local.partnership_account = partnership_account
            else:
                partnership = Partnership.get_partnership_by_token(api_token_id)

                if partnership is not None:
                    _local.partnership = partnership
                else:
                    return make_error(
                        'Invalid credentials', 'invalid_token')
        else:
            return make_error(
                'Invalid credentials', 'invalid_token')

        return wrapped(*args, **kwargs)
    return wrapper


def authenticate(text, code):
    message = {'message': "Authentication error."}
    resp = jsonify(message)
    resp.status_code = 401
    resp.headers['WWW-Authenticate'] = (
            'Bearer realm="buyercall", error="{}", '
            'error_description="{}"'
        ).format(code, text)

    return resp


def unsupported_in_debug_mode():
    message = {'message': "Error processing request. POST, PUT and DELETE requests are disabled in DEBUG mode."}
    resp = jsonify(message)
    resp.status_code = 401

    return resp


def requires_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        from buyercall.blueprints.partnership.models import PartnershipAccount, ApiToken, Partnership

        _local.is_partnership_account = False
        _local.is_partnership = False
        header = request.headers.get('Authorize', '').split(' ')
        api_doc_debug = request.headers.get('Documentation', '').lower()

        if app.config.get('DEBUG', False) is False \
                and request.method in blocked_debug_request_methods \
                and api_doc_debug == "true":
            return unsupported_in_debug_mode()

        if len(header) != 2 or header[0] != 'Bearer':
            return authenticate('Invalid authorization method.', 'invalid_request')

        token = header[1].strip()
        api_token_id = ApiToken.check_token(token)

        if api_token_id is not None:
            retrieved_partnership_account = PartnershipAccount.get_partnership_account_by_token(api_token_id)
            _local.api_token_id = api_token_id

            if retrieved_partnership_account is not None:
                _local.partnership_account = retrieved_partnership_account
                _local.is_partnership_account = True
            else:
                retrieved_partnership = Partnership.get_partnership_by_token(api_token_id)

                if retrieved_partnership is not None:
                    _local.partnership = retrieved_partnership
                    _local.is_partnership = True
        else:
            return authenticate('Invalid credentials.', 'invalid_token')
        return f(*args, **kwargs)
    return decorated