96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
""" Various decorators for endpoint and API handlers. """
|
|
|
|
from functools import wraps
|
|
from flask import abort, request, make_response
|
|
|
|
import features
|
|
|
|
from app import app
|
|
from auth.auth_context import (
|
|
get_validated_oauth_token, get_authenticated_user, get_validated_token, get_grant_context)
|
|
from util.names import parse_namespace_repository
|
|
|
|
|
|
def parse_repository_name(include_tag=False,
|
|
ns_kwarg_name='namespace_name',
|
|
repo_kwarg_name='repo_name',
|
|
tag_kwarg_name='tag_name',
|
|
incoming_repo_kwarg='repository'):
|
|
""" Decorator which parses the repository name found in the incoming_repo_kwarg argument,
|
|
and applies its pieces to the decorated function.
|
|
"""
|
|
def inner(func):
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
repo_name_components = parse_namespace_repository(kwargs[incoming_repo_kwarg],
|
|
app.config['LIBRARY_NAMESPACE'],
|
|
include_tag=include_tag)
|
|
del kwargs[incoming_repo_kwarg]
|
|
kwargs[ns_kwarg_name] = repo_name_components[0]
|
|
kwargs[repo_kwarg_name] = repo_name_components[1]
|
|
if include_tag:
|
|
kwargs[tag_kwarg_name] = repo_name_components[2]
|
|
return func(*args, **kwargs)
|
|
return wrapper
|
|
return inner
|
|
|
|
|
|
def param_required(param_name, allow_body=False):
|
|
""" Marks a route as requiring a parameter with the given name to exist in the request's arguments
|
|
or (if allow_body=True) in its body values. If the parameter is not present, the request will
|
|
fail with a 400.
|
|
"""
|
|
def wrapper(wrapped):
|
|
@wraps(wrapped)
|
|
def decorated(*args, **kwargs):
|
|
if param_name not in request.args:
|
|
if not allow_body or param_name not in request.values:
|
|
abort(make_response('Required param: %s' % param_name, 400))
|
|
return wrapped(*args, **kwargs)
|
|
return decorated
|
|
return wrapper
|
|
|
|
|
|
def anon_allowed(func):
|
|
""" Marks a method to allow anonymous access where it would otherwise be disallowed. """
|
|
func.__anon_allowed = True
|
|
return func
|
|
|
|
|
|
def anon_protect(func):
|
|
""" Marks a method as requiring some form of valid user auth before it can be executed. """
|
|
func.__anon_protected = True
|
|
return check_anon_protection(func)
|
|
|
|
|
|
def check_anon_protection(func):
|
|
""" Validates a method as requiring some form of valid user auth before it can be executed. """
|
|
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
# Skip if anonymous access is allowed.
|
|
if features.ANONYMOUS_ACCESS or '__anon_allowed' in dir(func):
|
|
return func(*args, **kwargs)
|
|
|
|
# Check for validated context. If none exists, fail with a 401.
|
|
if (get_authenticated_user() or get_validated_oauth_token() or get_validated_token() or
|
|
get_grant_context()):
|
|
return func(*args, **kwargs)
|
|
|
|
abort(401)
|
|
|
|
return wrapper
|
|
|
|
|
|
def route_show_if(value):
|
|
""" Adds/shows the decorated route if the given value is True. """
|
|
|
|
def decorator(f):
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
if not value:
|
|
abort(404)
|
|
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
return decorator
|