<commit_before>#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import json
import time
import uuid
import motor
import base64
import logging
import email.utils

from jose import jwt
from tornado.ioloop import IOLoop
from tornado.web import RequestHandler, Application, url, Finish, stream_request_body
from tornado import gen, httputil, httpserver, escape
from concurrent.futures import ThreadPoolExecutor
from ipaddress import ip_address

import common
import repo_manage


access_log = logging.getLogger('tornado.access')
app_log = logging.getLogger('tornado.application')
gen_log = logging.getLogger('tornado.general')


class CacusRequestHandler(RequestHandler):
    def set_default_headers(self):
        self.set_header("Access-Control-Allow-Origin", "*")
        self.set_header("Access-Control-Allow-Headers", "x-requested-with, authorization, content-type, accept")
        self.set_header('Access-Control-Allow-Methods', 'POST, PUT, GET, OPTIONS')

    def options(self, *args, **kwargs):
        self.set_status(204)
        self.finish()


class CachedRequestHandler(CacusRequestHandler):

    def prepare(self):
        pass

    @gen.coroutine
    def _cache_expired(self, item, selector):
        """ Checks whether item in DB was updated since client's latest version
        TODO: i don't like this double-quering of DB, first for expiry check, than
        for actual data. Should be redesigned.
        """
        db = self.settings['db']
        latest_dt = None
        result = db.cacus[item].find(selector, {'lastupdated': 1})
        while (yield result.fetch_next):
            dt = result.next_object()['lastupdated']
            if not latest_dt or dt > latest_dt:
                latest_dt = dt

        if not latest_dt:
            raise gen.Return((True, None))

        if_modified = self.request.headers.get('If-Modified-Since')
        if not if_modified:
            raise gen.Return((True, latest_dt))

        cached_dt = email.utils.parsedate(if_modified)
        cached_ts = time.mktime(cached_dt)
        latest_ts = time.mktime(latest_dt.timetuple())
        if latest_ts <= cached_ts:
            raise gen.Return((False, latest_dt))
        else:
            raise gen.Return((True, latest_dt))


# TODO JSON schema?
class ApiRequestHandler(CacusRequestHandler):
    """ Provides JSON body processing and authentication/authorization using JWT """

    def _get_json_request(self):
        if 'application/json' not in self.request.headers.get('Content-type', '').lower():
            self.set_status(400)
            self.write({'success': False, 'msg': 'application/json Content-type expected'})
            raise Finish()
        try:
            app_log.debug("body '%s'",  self.request.body)
            req = json.loads(escape.to_unicode(self.request.body))
        except Exception as e:
            self.set_status(400)
            self.write({'success': False, 'msg': 'invalid JSON: {}'.format(e)})
            raise Finish()

        class JsonRequestData(dict):

            def __init__(self, request, *args, **kwargs):
                self._request = request
                super(JsonRequestData, self).__init__(*args, **kwargs)

            def __getitem__(self, key):
                try:
                    return dict.__getitem__(self, key)
                except KeyError:
                    app_log.error("Missing required argument %s", key)
                    self._request.set_status(400)
                    self._request.write({'success': False, 'msg': "Missing required argument '{}'".format(key)})
                    raise Finish()

        return JsonRequestData(self, req)

    @gen.coroutine
    def _check_token(self, aud):
        config = self.settings['manager'].config

        ip = ip_address(unicode(self.request.remote_ip))
        for net in config['repo_daemon']['privileged_nets']:
            if ip in net:
                # no auth required
                raise gen.Return({})

        try:
            secret = base64.b64decode(config['repo_daemon']['auth_secret'])
            if 'Authorization' not in self.request.headers:
                raise Exception("Authorization required")
            scheme, token = self.request.headers['Authorization'].split(' ')
            if scheme != 'Bearer':
                raise Exception("Use Bearer authorization scheme")
            try:
                claim = jwt.decode(token, secret, audience=aud, algorithms='HS256')
            except jwt.JWTClaimsError as e:
                if 'Invalid audience' in e:
                    claim = jwt.decode(token, secret, audience=common.Cacus.admin_access, algorithms='HS256')
            if 'jti' in claim:
                # check for revoked tokens
                doc = yield self.settings['db'].cacus.access_tokens.find_one({'jti': claim['jti']})
                if doc and doc['revoked']:
                    raise Exception("Token blacklisted")
            else:
                if self.settings['manager'].config['repo_daemon'].get('reject_old_tokens', True):
                    raise Exception("Old token format, please renew")


        except Exception as e:
            self.set_status(401)
            self.write({'success': False, 'msg': str(e)})
            raise Finish()

        app_log.user = claim['sub']
        access_log.user = claim['sub']
        self.settings['manager'].log.user = claim['sub']
        raise gen.Return(claim)

    def on_finish(self):
        app_log.user = None
        access_log.user = None
        self.settings['manager'].log.user = None


class StorageHandler(CacusRequestHandler):

    def on_connection_close(self):
        self.dead = True

    @gen.coroutine
    def stream_from_storage(self, key=None, headers=[]):
        self.dead = False
        stream = common.ProxyStream(self, headers=headers)
        self.set_header('Content-Type', 'application/octet-stream')
        # TODO last-modified, content-length and other metadata _should_ be provided!
        try:
            yield self.settings['workers'].submit(self.settings['manager'].storage.get, key, stream)
        except common.NotFound:
            self.set_status(404)
            app_log.error("Key %s was not found at storage", key)
        except common.FatalError as e:
            # self.set_status(500)
            app_log.error("Got error from storage plugin: %s", e)
        self.finish()

    @gen.coroutine
    def get(self, key):
        yield self.stream_from_storage(key)


class ExtStorageHandler(CacusRequestHandler):
    """ Redirects to external location.
    APT should support redirects since version 0.7.21 (see https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=79002)
    """

    @gen.coroutine
    def get(self, url):
        # unescaping being performed here automagical?
        # url = escape.url_unescape(url)
        self.set_header('Location', url)
        self.set_status(302)


class PackagesHandler(CachedRequestHandler, StorageHandler):

    @gen.coroutine
    def get(self, distro=None, comp=None, arch=None):

        db = self.settings['db']
        (expired, dt) = yield self._cache_expired('repos', {'distro': distro, 'component': comp, 'architecture': arch})
        if not expired:
            self.set_status(304)
            return
        if not dt:
            self.set_status(404)
            return
        self.add_header('Last-Modified', httputil.format_timestamp(dt))
        self.set_header('Content-Type', 'application/octet-stream')

        doc = yield db.cacus.repos.find_one({'distro': distro, 'component': comp, 'architecture': arch})
        if doc:
            s = self.settings['config']['repo_daemon']
            if s['proxy_storage']:
                headers = [('Content-Length', doc['size']), ('Last-Modified', httputil.format_timestamp(dt))]
                yield self.stream_from_storage(doc['packages_file'], headers=headers)
            else:
                # we use x-accel-redirect instead of direct proxying via storage plugin to allow
                # user to offload cacus' StorageHandler if current storage allows it
                url = os.path.join(s['repo_base'], s['storage_subdir'], doc['packages_file'])
                app_log.info("Redirecting %s/%s/%s/Packages to %s", distro, comp, arch, url)
                self.add_header("X-Accel-Redirect", url)
                self.set_status(200)
        else:
            self.set_status(404)


class SourcesHandler(CachedRequestHandler, StorageHandler):
    """ Returns Sources repo indice
    Generating on the fly as it's rarely used so no point to slow down
    metadata update by pre-generating and storing this file
    """

    @gen.coroutine
    def get(self, distro=None, comp=None):

        db = self.settings['db']
        (expired, dt) = yield self._cache_expired('components', {'distro': distro, 'component': comp})
        if not expired:
            self.set_status(304)
            return
        if not dt:
            self.set_status(404)
            return
        self.add_header('Last-Modified', httputil.format_timestamp(dt))
        self.set_header('Content-Type', 'application/octet-stream')

        doc = yield db.cacus.components.find_one({'distro': distro, 'component': comp})
        if doc:
            s = self.settings['config']['repo_daemon']
            if s['proxy_storage']:
                headers = [('Content-Length', doc['size']), ('Last-Modified', httputil.format_timestamp(dt))]
                yield self.stream_from_storage(doc['sources_file'], headers=headers)
            else:
                # we use x-accel-redirect instead of direct proxying via storage plugin to allow
                # user to offload cacus' StorageHandler if current storage allows it
                url = os.path.join(s['repo_base'], s['storage_subdir'], doc['sources_file'])
                app_log.info("Redirecting %s/%s/source/Sources to %s", distro, comp, url)
                self.add_header("X-Accel-Redirect", url)
                self.set_status(200)
        else:
            self.set_status(404)


class SourcesFilesHandler(CachedRequestHandler):

    @gen.coroutine
    def get(self, distro=None, comp=None, file=None):
        db = self.settings['db']
        doc = yield db.cacus.repos[distro].find_one({'component': comp, 'sources.name': file},
                                                    {'sources.storage_key': 1, 'sources.name': 1})
        for f in doc['sources']:
            if f['name'] == file:
                s = self.settings['config']['repo_daemon']
                url = os.path.join(s['repo_base'], f['storage_key'])
                app_log.info("Redirecting %s to %s", file, url)
                self.add_header("X-Accel-Redirect", url)
                break
        self.set_status(200)


class ReleaseHandler(CachedRequestHandler):

    @gen.coroutine
    def get(self, distro=None, gpg=None):
        db = self.settings['db']
        (expired, dt) = yield self._cache_expired('distros', {'distro': distro})
        if not expired:
            self.set_status(304)
            return
        if not dt:
            self.set_status(404)
            return
        self.add_header('Last-Modified', httputil.format_timestamp(dt))
        self.set_header('Content-Type', 'application/octet-stream')

        doc = yield db.cacus.distros.find_one({'distro': distro})
        if gpg:
            self.write(doc['release_gpg'])
        else:
            self.write(doc['release_file'])


class ApiDistroReindexHandler(ApiRequestHandler):

    @gen.coroutine
    def post(self, distro):
        yield self._check_token(distro)
        try:
            yield self.settings['workers'].submit(self.settings['manager'].update_distro_metadata, distro=distro)
        except common.NotFound as e:
            self.set_status(404)
            self.write({'success': False, 'msg': e.message})
            return
        self.write({'success': True, 'msg': 'Reindex complete'})


class ApiDistroShowHandler(ApiRequestHandler):

    @gen.coroutine
    def get(self, distro):
        yield self._check_token(distro or common.Cacus.admin_access)
        if distro:
            selector = {'distro': distro}
        else:
            selector = {}

        result = []
        cursor = self.settings['db'].cacus.distros.find(selector, {'release_file': 0, 'release_gpg': 0})
        while (yield cursor.fetch_next):
            d = cursor.next_object()
            c = self.settings['db'].cacus.components.find({'distro': d['distro']}, {'component': 1}).to_list(None)<commit_msg>api/v1/distro/show now shows all distros available for authenticated user<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import json
import time
import uuid
import motor
import base64
import logging
import email.utils

from jose import jwt
from tornado.ioloop import IOLoop
from tornado.web import RequestHandler, Application, url, Finish, stream_request_body
from tornado import gen, httputil, httpserver, escape
from concurrent.futures import ThreadPoolExecutor
from ipaddress import ip_address

import common
import repo_manage


access_log = logging.getLogger('tornado.access')
app_log = logging.getLogger('tornado.application')
gen_log = logging.getLogger('tornado.general')


class CacusRequestHandler(RequestHandler):
    def set_default_headers(self):
        self.set_header("Access-Control-Allow-Origin", "*")
        self.set_header("Access-Control-Allow-Headers", "x-requested-with, authorization, content-type, accept")
        self.set_header('Access-Control-Allow-Methods', 'POST, PUT, GET, OPTIONS')

    def options(self, *args, **kwargs):
        self.set_status(204)
        self.finish()


class CachedRequestHandler(CacusRequestHandler):

    def prepare(self):
        pass

    @gen.coroutine
    def _cache_expired(self, item, selector):
        """ Checks whether item in DB was updated since client's latest version
        TODO: i don't like this double-quering of DB, first for expiry check, than
        for actual data. Should be redesigned.
        """
        db = self.settings['db']
        latest_dt = None
        result = db.cacus[item].find(selector, {'lastupdated': 1})
        while (yield result.fetch_next):
            dt = result.next_object()['lastupdated']
            if not latest_dt or dt > latest_dt:
                latest_dt = dt

        if not latest_dt:
            raise gen.Return((True, None))

        if_modified = self.request.headers.get('If-Modified-Since')
        if not if_modified:
            raise gen.Return((True, latest_dt))

        cached_dt = email.utils.parsedate(if_modified)
        cached_ts = time.mktime(cached_dt)
        latest_ts = time.mktime(latest_dt.timetuple())
        if latest_ts <= cached_ts:
            raise gen.Return((False, latest_dt))
        else:
            raise gen.Return((True, latest_dt))


# TODO JSON schema?
class ApiRequestHandler(CacusRequestHandler):
    """ Provides JSON body processing and authentication/authorization using JWT """

    def _get_json_request(self):
        if 'application/json' not in self.request.headers.get('Content-type', '').lower():
            self.set_status(400)
            self.write({'success': False, 'msg': 'application/json Content-type expected'})
            raise Finish()
        try:
            app_log.debug("body '%s'",  self.request.body)
            req = json.loads(escape.to_unicode(self.request.body))
        except Exception as e:
            self.set_status(400)
            self.write({'success': False, 'msg': 'invalid JSON: {}'.format(e)})
            raise Finish()

        class JsonRequestData(dict):

            def __init__(self, request, *args, **kwargs):
                self._request = request
                super(JsonRequestData, self).__init__(*args, **kwargs)

            def __getitem__(self, key):
                try:
                    return dict.__getitem__(self, key)
                except KeyError:
                    app_log.error("Missing required argument %s", key)
                    self._request.set_status(400)
                    self._request.write({'success': False, 'msg': "Missing required argument '{}'".format(key)})
                    raise Finish()

        return JsonRequestData(self, req)

    @gen.coroutine
    def _check_token(self, aud):
        config = self.settings['manager'].config

        ip = ip_address(unicode(self.request.remote_ip))
        for net in config['repo_daemon']['privileged_nets']:
            if ip in net:
                # no auth required
                raise gen.Return({'privileged': True})

        try:
            secret = base64.b64decode(config['repo_daemon']['auth_secret'])
            if 'Authorization' not in self.request.headers:
                raise Exception("Authorization required")
            scheme, token = self.request.headers['Authorization'].split(' ')
            if scheme != 'Bearer':
                raise Exception("Use Bearer authorization scheme")
            try:
                if aud is not None:
                    claim = jwt.decode(token, secret, audience=aud, algorithms='HS256')
                else:
                    claim = jwt.decode(token, secret, algorithms='HS256', options={'verify_aud': False})
            except jwt.JWTClaimsError as e:
                if 'Invalid audience' in e:
                    claim = jwt.decode(token, secret, audience=common.Cacus.admin_access, algorithms='HS256')
            if 'jti' in claim:
                # check for revoked tokens
                doc = yield self.settings['db'].cacus.access_tokens.find_one({'jti': claim['jti']})
                if doc and doc['revoked']:
                    raise Exception("Token blacklisted")
            else:
                if self.settings['manager'].config['repo_daemon'].get('reject_old_tokens', True):
                    raise Exception("Old token format, please renew")


        except Exception as e:
            self.set_status(401)
            self.write({'success': False, 'msg': str(e)})
            raise Finish()

        app_log.user = claim['sub']
        access_log.user = claim['sub']
        self.settings['manager'].log.user = claim['sub']
        raise gen.Return(claim)

    def on_finish(self):
        app_log.user = None
        access_log.user = None
        self.settings['manager'].log.user = None


class StorageHandler(CacusRequestHandler):

    def on_connection_close(self):
        self.dead = True

    @gen.coroutine
    def stream_from_storage(self, key=None, headers=[]):
        self.dead = False
        stream = common.ProxyStream(self, headers=headers)
        self.set_header('Content-Type', 'application/octet-stream')
        # TODO last-modified, content-length and other metadata _should_ be provided!
        try:
            yield self.settings['workers'].submit(self.settings['manager'].storage.get, key, stream)
        except common.NotFound:
            self.set_status(404)
            app_log.error("Key %s was not found at storage", key)
        except common.FatalError as e:
            # self.set_status(500)
            app_log.error("Got error from storage plugin: %s", e)
        self.finish()

    @gen.coroutine
    def get(self, key):
        yield self.stream_from_storage(key)


class ExtStorageHandler(CacusRequestHandler):
    """ Redirects to external location.
    APT should support redirects since version 0.7.21 (see https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=79002)
    """

    @gen.coroutine
    def get(self, url):
        # unescaping being performed here automagical?
        # url = escape.url_unescape(url)
        self.set_header('Location', url)
        self.set_status(302)


class PackagesHandler(CachedRequestHandler, StorageHandler):

    @gen.coroutine
    def get(self, distro=None, comp=None, arch=None):

        db = self.settings['db']
        (expired, dt) = yield self._cache_expired('repos', {'distro': distro, 'component': comp, 'architecture': arch})
        if not expired:
            self.set_status(304)
            return
        if not dt:
            self.set_status(404)
            return
        self.add_header('Last-Modified', httputil.format_timestamp(dt))
        self.set_header('Content-Type', 'application/octet-stream')

        doc = yield db.cacus.repos.find_one({'distro': distro, 'component': comp, 'architecture': arch})
        if doc:
            s = self.settings['config']['repo_daemon']
            if s['proxy_storage']:
                headers = [('Content-Length', doc['size']), ('Last-Modified', httputil.format_timestamp(dt))]
                yield self.stream_from_storage(doc['packages_file'], headers=headers)
            else:
                # we use x-accel-redirect instead of direct proxying via storage plugin to allow
                # user to offload cacus' StorageHandler if current storage allows it
                url = os.path.join(s['repo_base'], s['storage_subdir'], doc['packages_file'])
                app_log.info("Redirecting %s/%s/%s/Packages to %s", distro, comp, arch, url)
                self.add_header("X-Accel-Redirect", url)
                self.set_status(200)
        else:
            self.set_status(404)


class SourcesHandler(CachedRequestHandler, StorageHandler):
    """ Returns Sources repo indice
    Generating on the fly as it's rarely used so no point to slow down
    metadata update by pre-generating and storing this file
    """

    @gen.coroutine
    def get(self, distro=None, comp=None):

        db = self.settings['db']
        (expired, dt) = yield self._cache_expired('components', {'distro': distro, 'component': comp})
        if not expired:
            self.set_status(304)
            return
        if not dt:
            self.set_status(404)
            return
        self.add_header('Last-Modified', httputil.format_timestamp(dt))
        self.set_header('Content-Type', 'application/octet-stream')

        doc = yield db.cacus.components.find_one({'distro': distro, 'component': comp})
        if doc:
            s = self.settings['config']['repo_daemon']
            if s['proxy_storage']:
                headers = [('Content-Length', doc['size']), ('Last-Modified', httputil.format_timestamp(dt))]
                yield self.stream_from_storage(doc['sources_file'], headers=headers)
            else:
                # we use x-accel-redirect instead of direct proxying via storage plugin to allow
                # user to offload cacus' StorageHandler if current storage allows it
                url = os.path.join(s['repo_base'], s['storage_subdir'], doc['sources_file'])
                app_log.info("Redirecting %s/%s/source/Sources to %s", distro, comp, url)
                self.add_header("X-Accel-Redirect", url)
                self.set_status(200)
        else:
            self.set_status(404)


class SourcesFilesHandler(CachedRequestHandler):

    @gen.coroutine
    def get(self, distro=None, comp=None, file=None):
        db = self.settings['db']
        doc = yield db.cacus.repos[distro].find_one({'component': comp, 'sources.name': file},
                                                    {'sources.storage_key': 1, 'sources.name': 1})
        for f in doc['sources']:
            if f['name'] == file:
                s = self.settings['config']['repo_daemon']
                url = os.path.join(s['repo_base'], f['storage_key'])
                app_log.info("Redirecting %s to %s", file, url)
                self.add_header("X-Accel-Redirect", url)
                break
        self.set_status(200)


class ReleaseHandler(CachedRequestHandler):

    @gen.coroutine
    def get(self, distro=None, gpg=None):
        db = self.settings['db']
        (expired, dt) = yield self._cache_expired('distros', {'distro': distro})
        if not expired:
            self.set_status(304)
            return
        if not dt:
            self.set_status(404)
            return
        self.add_header('Last-Modified', httputil.format_timestamp(dt))
        self.set_header('Content-Type', 'application/octet-stream')

        doc = yield db.cacus.distros.find_one({'distro': distro})
        if gpg:
            self.write(doc['release_gpg'])
        else:
            self.write(doc['release_file'])


class ApiDistroReindexHandler(ApiRequestHandler):

    @gen.coroutine
    def post(self, distro):
        yield self._check_token(distro)
        try:
            yield self.settings['workers'].submit(self.settings['manager'].update_distro_metadata, distro=distro)
        except common.NotFound as e:
            self.set_status(404)
            self.write({'success': False, 'msg': e.message})
            return
        self.write({'success': True, 'msg': 'Reindex complete'})


class ApiDistroShowHandler(ApiRequestHandler):

    @gen.coroutine
    def get(self, distro):
        claim = yield self._check_token(distro)
        if distro:
            if (claim['aud'] == common.Cacus.admin_access or 'privileged' in claim or distro in claim['aud']):
                selector = {'distro': distro}
            else:
                self.set_status(401)
                self.write({'success': False, 'msg': 'Access denied'})
                return
        else:
            if 'privileged' in claim or claim['aud'] == common.Cacus.admin_access:
                selector = {}
            else:
                selector = {'distro': {'$in': claim['aud']}}

        result = []
        cursor = self.settings['db'].cacus.distros.find(selector, {'release_file': 0, 'release_gpg': 0})
        while (yield cursor.fetch_next):
            d = cursor.next_object()
            c = self.settings['db'].cacus.components.find({'distro': d['distro']}, {'component': 1}).to_list(None)
{'fraction_non_alphanumeric': 0.12033227233910411, 'fraction_numerical': 0.0071034857542116294, 'mean_word_length': 2.358493025681137, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2088761', 'n_tokens_mistral': 7593, 'n_tokens_neox': 7236, 'n_words': 1702}