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