ff52fde8a5
This change ensures that we always store and then check the contents of the OAuth `state` argument against a session-stored CSRF token. Fixes https://www.pivotaltracker.com/story/show/135803615
383 lines
11 KiB
Python
383 lines
11 KiB
Python
import logging
|
|
import datetime
|
|
import json
|
|
|
|
from calendar import timegm
|
|
from email.utils import formatdate
|
|
from functools import partial, wraps
|
|
|
|
from enum import Enum
|
|
from flask import Blueprint, Response, request, make_response, jsonify, session, url_for
|
|
from flask_restful import Resource, abort, Api, reqparse
|
|
from flask_restful.utils.cors import crossdomain
|
|
from jsonschema import validate, ValidationError
|
|
|
|
from app import app, metric_queue
|
|
from data import model
|
|
from auth.permissions import (ReadRepositoryPermission, ModifyRepositoryPermission,
|
|
AdministerRepositoryPermission, UserReadPermission,
|
|
UserAdminPermission)
|
|
from auth import scopes
|
|
from auth.auth_context import get_authenticated_user, get_validated_oauth_token
|
|
from auth.process import process_oauth
|
|
from endpoints.csrf import csrf_protect
|
|
from endpoints.exception import (ApiException, Unauthorized, InvalidRequest, InvalidResponse,
|
|
FreshLoginRequired)
|
|
from endpoints.decorators import check_anon_protection
|
|
from util.metrics.metricqueue import time_decorator
|
|
from util.names import parse_namespace_repository
|
|
from util.pagination import encrypt_page_token, decrypt_page_token
|
|
|
|
logger = logging.getLogger(__name__)
|
|
api_bp = Blueprint('api', __name__)
|
|
api = Api()
|
|
api.init_app(api_bp)
|
|
api.decorators = [csrf_protect(),
|
|
crossdomain(origin='*', headers=['Authorization', 'Content-Type']),
|
|
process_oauth, time_decorator(api_bp.name, metric_queue)]
|
|
|
|
|
|
@api_bp.app_errorhandler(ApiException)
|
|
@crossdomain(origin='*', headers=['Authorization', 'Content-Type'])
|
|
def handle_api_error(error):
|
|
response = Response(json.dumps(error.to_dict()), error.status_code, mimetype='application/json')
|
|
if error.status_code == 401:
|
|
response.headers['WWW-Authenticate'] = ('Bearer error="%s" error_description="%s"' %
|
|
(error.error_type.value, error.error_description))
|
|
return response
|
|
|
|
def resource(*urls, **kwargs):
|
|
def wrapper(api_resource):
|
|
if not api_resource:
|
|
return None
|
|
|
|
api.add_resource(api_resource, *urls, **kwargs)
|
|
return api_resource
|
|
return wrapper
|
|
|
|
|
|
def show_if(value):
|
|
def f(inner):
|
|
if not value:
|
|
return None
|
|
|
|
return inner
|
|
return f
|
|
|
|
|
|
def hide_if(value):
|
|
def f(inner):
|
|
if value:
|
|
return None
|
|
|
|
return inner
|
|
return f
|
|
|
|
|
|
def truthy_bool(param):
|
|
return param not in {False, 'false', 'False', '0', 'FALSE', '', 'null'}
|
|
|
|
|
|
def format_date(date):
|
|
""" Output an RFC822 date format. """
|
|
if date is None:
|
|
return None
|
|
return formatdate(timegm(date.utctimetuple()))
|
|
|
|
|
|
def add_method_metadata(name, value):
|
|
def modifier(func):
|
|
if func is None:
|
|
return None
|
|
|
|
if '__api_metadata' not in dir(func):
|
|
func.__api_metadata = {}
|
|
func.__api_metadata[name] = value
|
|
return func
|
|
return modifier
|
|
|
|
|
|
def method_metadata(func, name):
|
|
if func is None:
|
|
return None
|
|
|
|
if '__api_metadata' in dir(func):
|
|
return func.__api_metadata.get(name, None)
|
|
return None
|
|
|
|
|
|
|
|
nickname = partial(add_method_metadata, 'nickname')
|
|
related_user_resource = partial(add_method_metadata, 'related_user_resource')
|
|
internal_only = add_method_metadata('internal', True)
|
|
|
|
|
|
def path_param(name, description):
|
|
def add_param(func):
|
|
if not func:
|
|
return func
|
|
|
|
if '__api_path_params' not in dir(func):
|
|
func.__api_path_params = {}
|
|
func.__api_path_params[name] = {
|
|
'name': name,
|
|
'description': description
|
|
}
|
|
return func
|
|
return add_param
|
|
|
|
|
|
def query_param(name, help_str, type=reqparse.text_type, default=None,
|
|
choices=(), required=False):
|
|
def add_param(func):
|
|
if '__api_query_params' not in dir(func):
|
|
func.__api_query_params = []
|
|
func.__api_query_params.append({
|
|
'name': name,
|
|
'type': type,
|
|
'help': help_str,
|
|
'default': default,
|
|
'choices': choices,
|
|
'required': required,
|
|
'location': ('args')
|
|
})
|
|
return func
|
|
return add_param
|
|
|
|
def page_support(page_token_kwarg='page_token', parsed_args_kwarg='parsed_args'):
|
|
def inner(func):
|
|
""" Adds pagination support to an API endpoint. The decorated API will have an
|
|
added query parameter named 'next_page'. Works in tandem with the
|
|
modelutil paginate method.
|
|
"""
|
|
@wraps(func)
|
|
@query_param('next_page', 'The page token for the next page', type=str)
|
|
def wrapper(self, *args, **kwargs):
|
|
# Note: if page_token is None, we'll receive the first page of results back.
|
|
page_token = decrypt_page_token(kwargs[parsed_args_kwarg]['next_page'])
|
|
kwargs[page_token_kwarg] = page_token
|
|
|
|
(result, next_page_token) = func(self, *args, **kwargs)
|
|
if next_page_token is not None:
|
|
result['next_page'] = encrypt_page_token(next_page_token)
|
|
|
|
return result
|
|
return wrapper
|
|
return inner
|
|
|
|
def parse_args(kwarg_name='parsed_args'):
|
|
def inner(func):
|
|
@wraps(func)
|
|
def wrapper(self, *args, **kwargs):
|
|
if '__api_query_params' not in dir(func):
|
|
abort(500)
|
|
|
|
parser = reqparse.RequestParser()
|
|
for arg_spec in func.__api_query_params:
|
|
parser.add_argument(**arg_spec)
|
|
kwargs[kwarg_name] = parser.parse_args()
|
|
|
|
return func(self, *args, **kwargs)
|
|
return wrapper
|
|
return inner
|
|
|
|
def parse_repository_name(func):
|
|
@wraps(func)
|
|
def wrapper(repository, *args, **kwargs):
|
|
(namespace, repository) = parse_namespace_repository(repository, app.config['LIBRARY_NAMESPACE'])
|
|
return func(namespace, repository, *args, **kwargs)
|
|
return wrapper
|
|
|
|
|
|
class ApiResource(Resource):
|
|
method_decorators = [check_anon_protection]
|
|
|
|
def options(self):
|
|
return None, 200
|
|
|
|
|
|
class RepositoryParamResource(ApiResource):
|
|
method_decorators = [check_anon_protection, parse_repository_name]
|
|
|
|
|
|
def require_repo_permission(permission_class, scope, allow_public=False):
|
|
def wrapper(func):
|
|
@add_method_metadata('oauth2_scope', scope)
|
|
@wraps(func)
|
|
def wrapped(self, namespace, repository, *args, **kwargs):
|
|
logger.debug('Checking permission %s for repo: %s/%s', permission_class, namespace,
|
|
repository)
|
|
permission = permission_class(namespace, repository)
|
|
if (permission.can() or
|
|
(allow_public and
|
|
model.repository.repository_is_public(namespace, repository))):
|
|
return func(self, namespace, repository, *args, **kwargs)
|
|
raise Unauthorized()
|
|
return wrapped
|
|
return wrapper
|
|
|
|
|
|
require_repo_read = require_repo_permission(ReadRepositoryPermission, scopes.READ_REPO, True)
|
|
require_repo_write = require_repo_permission(ModifyRepositoryPermission, scopes.WRITE_REPO)
|
|
require_repo_admin = require_repo_permission(AdministerRepositoryPermission, scopes.ADMIN_REPO)
|
|
|
|
|
|
def require_user_permission(permission_class, scope=None):
|
|
def wrapper(func):
|
|
@add_method_metadata('oauth2_scope', scope)
|
|
@wraps(func)
|
|
def wrapped(self, *args, **kwargs):
|
|
user = get_authenticated_user()
|
|
if not user:
|
|
raise Unauthorized()
|
|
|
|
logger.debug('Checking permission %s for user %s', permission_class, user.username)
|
|
permission = permission_class(user.username)
|
|
if permission.can():
|
|
return func(self, *args, **kwargs)
|
|
raise Unauthorized()
|
|
return wrapped
|
|
return wrapper
|
|
|
|
|
|
require_user_read = require_user_permission(UserReadPermission, scopes.READ_USER)
|
|
require_user_admin = require_user_permission(UserAdminPermission, scopes.ADMIN_USER)
|
|
|
|
|
|
def verify_not_prod(func):
|
|
@add_method_metadata('enterprise_only', True)
|
|
@wraps(func)
|
|
def wrapped(*args, **kwargs):
|
|
# Verify that we are not running on a production (i.e. hosted) stack. If so, we fail.
|
|
# This should never happen (because of the feature-flag on SUPER_USERS), but we want to be
|
|
# absolutely sure.
|
|
if app.config['SERVER_HOSTNAME'].find('quay.io') >= 0:
|
|
logger.error('!!! Super user method called IN PRODUCTION !!!')
|
|
raise NotFound()
|
|
|
|
return func(*args, **kwargs)
|
|
|
|
return wrapped
|
|
|
|
|
|
def require_fresh_login(func):
|
|
@add_method_metadata('requires_fresh_login', True)
|
|
@wraps(func)
|
|
def wrapped(*args, **kwargs):
|
|
user = get_authenticated_user()
|
|
if not user:
|
|
raise Unauthorized()
|
|
|
|
oauth_token = get_validated_oauth_token()
|
|
if oauth_token:
|
|
return func(*args, **kwargs)
|
|
|
|
logger.debug('Checking fresh login for user %s', user.username)
|
|
|
|
last_login = session.get('login_time', datetime.datetime.min)
|
|
valid_span = datetime.datetime.now() - datetime.timedelta(minutes=10)
|
|
|
|
if not user.password_hash or last_login >= valid_span:
|
|
return func(*args, **kwargs)
|
|
|
|
raise FreshLoginRequired()
|
|
return wrapped
|
|
|
|
|
|
def require_scope(scope_object):
|
|
def wrapper(func):
|
|
@add_method_metadata('oauth2_scope', scope_object)
|
|
@wraps(func)
|
|
def wrapped(*args, **kwargs):
|
|
return func(*args, **kwargs)
|
|
return wrapped
|
|
return wrapper
|
|
|
|
|
|
def validate_json_request(schema_name):
|
|
def wrapper(func):
|
|
@add_method_metadata('request_schema', schema_name)
|
|
@wraps(func)
|
|
def wrapped(self, *args, **kwargs):
|
|
schema = self.schemas[schema_name]
|
|
try:
|
|
json_data = request.get_json()
|
|
if json_data is None:
|
|
raise InvalidRequest('Missing JSON body')
|
|
|
|
validate(json_data, schema)
|
|
return func(self, *args, **kwargs)
|
|
except ValidationError as ex:
|
|
raise InvalidRequest(ex.message)
|
|
return wrapped
|
|
return wrapper
|
|
|
|
|
|
def request_error(exception=None, **kwargs):
|
|
data = kwargs.copy()
|
|
message = 'Request error.'
|
|
if exception:
|
|
message = exception.message
|
|
message = data.pop('message', message)
|
|
raise InvalidRequest(message, data)
|
|
|
|
|
|
def log_action(kind, user_or_orgname, metadata=None, repo=None):
|
|
if not metadata:
|
|
metadata = {}
|
|
|
|
oauth_token = get_validated_oauth_token()
|
|
if oauth_token:
|
|
metadata['oauth_token_id'] = oauth_token.id
|
|
metadata['oauth_token_application_id'] = oauth_token.application.client_id
|
|
metadata['oauth_token_application'] = oauth_token.application.name
|
|
|
|
performer = get_authenticated_user()
|
|
model.log.log_action(kind, user_or_orgname, performer=performer, ip=request.remote_addr,
|
|
metadata=metadata, repository=repo)
|
|
|
|
|
|
def define_json_response(schema_name):
|
|
def wrapper(func):
|
|
@add_method_metadata('response_schema', schema_name)
|
|
@wraps(func)
|
|
def wrapped(self, *args, **kwargs):
|
|
schema = self.schemas[schema_name]
|
|
resp = func(self, *args, **kwargs)
|
|
|
|
if app.config['TESTING']:
|
|
try:
|
|
validate(resp, schema)
|
|
except ValidationError as ex:
|
|
raise InvalidResponse(ex.message)
|
|
|
|
return resp
|
|
return wrapped
|
|
return wrapper
|
|
|
|
|
|
import endpoints.api.billing
|
|
import endpoints.api.build
|
|
import endpoints.api.discovery
|
|
import endpoints.api.error
|
|
import endpoints.api.globalmessages
|
|
import endpoints.api.image
|
|
import endpoints.api.logs
|
|
import endpoints.api.manifest
|
|
import endpoints.api.organization
|
|
import endpoints.api.permission
|
|
import endpoints.api.prototype
|
|
import endpoints.api.repository
|
|
import endpoints.api.repositorynotification
|
|
import endpoints.api.repoemail
|
|
import endpoints.api.repotoken
|
|
import endpoints.api.robot
|
|
import endpoints.api.search
|
|
import endpoints.api.suconfig
|
|
import endpoints.api.superuser
|
|
import endpoints.api.tag
|
|
import endpoints.api.team
|
|
import endpoints.api.trigger
|
|
import endpoints.api.user
|
|
import endpoints.api.secscan
|
|
|