Produits Entr'ouvert » w.c.s.

import sys

from wcs.api_utils import sign_url, get_secret_and_orig, is_url_signed, DEFAULT_DURATION

from wcs.qommon.errors import AccessForbiddenError

def test_is_url_signed_check_nonce(pub, local_user):

    ORIG = 'xxx'

    KEY = 'xxx'

    pub.site_options.add_section('api-secrets')

    pub.site_options.set('api-secrets', ORIG, KEY)

    # test clean_nonces do not bark when nonces directory is empty

    if os.path.exists(os.path.join(pub.app_dir, 'nonces')):

        shutil.rmtree(os.path.join(pub.app_dir, 'nonces'))

    pub.clean_nonces()

    signed_url = sign_url('?format=json&orig=%s&email=%s'

                          % (ORIG, urllib.quote(local_user.email)), KEY)

    req = HTTPRequest(None, {'SCRIPT_NAME': '/',

                             'SERVER_NAME': 'example.net',

                             'QUERY_STRING': signed_url[1:]})

    req.process_inputs()

    pub.set_app_dir(req)

    pub._set_request(req)

    assert is_url_signed()

    with pytest.raises(AccessForbiddenError):

        req.signed = False

        is_url_signed()

    assert sys.exc_value.public_msg == 'nonce already used'

    # test that clean nonces works

    assert os.listdir(os.path.join(pub.app_dir, 'nonces'))

    pub.clean_nonces(delta=0)

    assert os.listdir(os.path.join(pub.app_dir, 'nonces'))

    pub.clean_nonces(delta=0, now=time.time() + 2 * DEFAULT_DURATION)

    assert not os.listdir(os.path.join(pub.app_dir, 'nonces'))

    def url():

        signed_url = sign_url('http://example.net/api/formdefs/test/submit' +

                '?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email), '1234')

        return signed_url[len('http://example.net'):]

    resp = get_app(pub).post_json(url(), {'data': {}})

    resp = get_app(pub).post_json(url(), {'data': {}}, status=403)

    resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {}}, status=403)

    resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {}}, status=403)

    resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {}})

    resp = get_app(pub).post_json(url(), {'data': {}})

    resp = get_app(pub).post_json(url(), {'meta': {'draft': True}, 'data': {}})

    resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {},

    def url():

        signed_url = sign_url('http://example.net/api/formdefs/test/submit' +

                              '?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email),

                              '1234')

        return signed_url[len('http://example.net'):]

    resp = get_app(pub).post_json(url(), {'data': {}})

    resp = get_app(pub).post_json(url(), {'data': {}}, status=403)

    resp = get_app(pub).post_json(url(), {'data': {}}, status=200)

import os

import errno

import calendar

import qommon.misc

DEFAULT_DURATION = 30

def is_url_signed(utcnow=None, duration=DEFAULT_DURATION):

    if getattr(get_request(), 'signed', False):

        return True

        timestamp = datetime.datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%SZ')

    except ValueError, e:

        raise AccessForbiddenError('invalid timestamp field; %s' % e)

    delta = (utcnow or datetime.datetime.utcnow()).replace(tzinfo=None) - timestamp

    if abs(delta) > datetime.timedelta(seconds=duration):

        raise AccessForbiddenError('timestamp delta is more than %s seconds: %s'

                                   % (duration, delta))

    # check nonce

    nonce = get_request().form.get('nonce')

    if nonce is not None:

        # compute end date of the nonce as an unix timestamp

        until = timestamp + datetime.timedelta(seconds=duration)

        until = calendar.timegm(until.timetuple())

        # normalize nonce

        nonce = nonce[:128]

        nonce = qommon.misc.simplify(nonce).replace('/', '-')

        nonce_dir = os.path.join(get_publisher().app_dir, 'nonces')

        nonce_path = os.path.join(nonce_dir, nonce)

        try:

            try:

                if not os.path.exists(nonce_dir):

                    os.makedirs(nonce_dir)

            except OSError, e:

                if e.errno != errno.EEXIST:

                    raise

            os.open(nonce_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)

            # touch file in the future for cleaning_nonces

            os.utime(nonce_path, (until, until))

        except OSError, e:

            if e.errno != errno.EEXIST:

                raise

            raise AccessForbiddenError('nonce already used')

    get_request().signed = True

    def clean_nonces(self, delta=60, now=None):

        nonce_dir = os.path.join(get_publisher().app_dir, 'nonces')

        now = now or time.time()

        if not os.path.exists(nonce_dir):

            return

        cleaning_lock_file = os.path.join(self.app_dir, 'cleaning_nonces.lock')

        fd = os.open(cleaning_lock_file, os.O_RDONLY | os.O_CREAT, 0666)

        try:

            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)

        except IOError:

            # lock is currently held, that is fine.

            return

        try:

            for nonce in os.listdir(nonce_dir):

                nonce_path = os.path.join(nonce_dir, nonce)

                # we need delta so that os.utime in is_url_signed has time to update file timestamp

                if delta < now - os.stat(nonce_path)[8]:

                    os.unlink(nonce_path)

        finally:

            os.close(fd)

        cls.register_cronjob(CronJob(cls.clean_nonces, minutes=range(0, 60, 5)))