0001-api-check-signature-s-nonce-10923.patch

Benjamin Dauvergne, 15 mai 2016 08:04

Voir les différences: en ligne côte à côte

Subject: [PATCH] api: check signature's nonce (#10923)

It's mandatory for security.

 tests/test_api.py       | 51 ++++++++++++++++++++++++++++++++++++++-----------
 wcs/api_utils.py        | 32 +++++++++++++++++++++++++++++++
 wcs/qommon/publisher.py | 22 +++++++++++++++++++++
 3 files changed, 94 insertions(+), 11 deletions(-)

     import datetime
     import time
     import json
     import sys
     from quixote import cleanup, get_publisher
     from wcs.qommon.errors import AccessForbiddenError
-...
         with pytest.raises(AccessForbiddenError):
             test_is_url_signed(utcnow=datetime.datetime(1970, 1, 1, 0, 0, 31))
     def test_is_url_signed_check_nonce(pub, local_user):
         pub.site_options.add_section('api-secrets')
         pub.site_options.set('api-secrets', 'xxx', 'xxx')
         # 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=xxx&email=%s' % urllib.quote(local_user.email),
                 'xxx', duration=1)
         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'))
         time.sleep(3.)
         pub.clean_nonces(delta=0)
         assert not os.listdir(os.path.join(pub.app_dir, 'nonces'))
     def test_get_user_compat_endpoint(pub, local_user):
         signed_url = sign_url(
                 'http://example.net/user?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email),
-...
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'unsigned API call'
         signed_url = sign_url('http://example.net/api/formdefs/test/submit' +
                 '?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email), '1234')
         url = signed_url[len('http://example.net'):]
         resp = get_app(pub).post_json(url, {'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': {}})
         assert resp.json['err'] == 0
         assert data_class.get(resp.json['data']['id']).status == 'wf-new'
         assert data_class.get(resp.json['data']['id']).user_id == str(local_user.id)
-...
         formdef.disabled = True
         formdef.store()
         resp = get_app(pub).post_json(url, {'data': {}}, status=403)
         resp = get_app(pub).post_json(url(), {'data': {}}, status=403)
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'disabled form'
         formdef.disabled = False
         formdef.store()
         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)
         formdef.backoffice_submission_roles = ['xx']
         formdef.store()
         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)
         formdef.backoffice_submission_roles = [role.id]
         formdef.store()
         resp = get_app(pub).post_json(url, {'meta': {'backoffice-submission': True}, 'data': {}})
         resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {}})
         assert data_class.get(resp.json['data']['id']).status == 'wf-new'
         assert data_class.get(resp.json['data']['id']).backoffice_submission is True
         assert data_class.get(resp.json['data']['id']).user_id is None
-...
         formdef.enable_tracking_codes = True
         formdef.store()
         resp = get_app(pub).post_json(url, {'data': {}})
         resp = get_app(pub).post_json(url(), {'data': {}})
         assert data_class.get(resp.json['data']['id']).tracking_code
         resp = get_app(pub).post_json(url, {'meta': {'draft': True}, 'data': {}})
         resp = get_app(pub).post_json(url(), {'meta': {'draft': True}, 'data': {}})
         assert data_class.get(resp.json['data']['id']).status == 'draft'
         resp = get_app(pub).post_json(url, {'meta': {'backoffice-submission': True}, 'data': {},
         resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {},
             'context': {'channel': 'mail', 'comments': 'blah'} })
         assert data_class.get(resp.json['data']['id']).status == 'wf-new'
         assert data_class.get(resp.json['data']['id']).backoffice_submission is True

     import urllib
     import urlparse
     import random
     import os
     import errno
     import calendar
     from quixote import get_request, get_publisher
     from qommon.errors import (AccessForbiddenError, UnknownNameIdAccessForbiddenError)
     import qommon.misc
     DEFAULT_DURATION = 30
     def is_url_signed(utcnow=None):
         if getattr(get_request(), 'signed', False):
             return True
         query_string = get_request().get_query()
         if not query_string:
             return False
-...
         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 not nonce is 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
         return True

                 cls.cronjobs = []
             cls.cronjobs.append(cronjob)
         def clean_nonces(self, delta=60):
             nonce_dir = os.path.join(get_publisher().app_dir, 'nonces')
             now = 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)
         def clean_sessions(self):
             cleaning_lock_file = os.path.join(self.app_dir, 'cleaning_sessions.lock')
             fd = os.open(cleaning_lock_file, os.O_RDONLY | os.O_CREAT, 0666)
-...
         @classmethod
         def register_cronjobs(cls):
             cls.register_cronjob(CronJob(cls.clean_sessions, minutes=range(0, 60, 5)))
             cls.register_cronjob(CronJob(cls.clean_nonces, minutes=range(0, 60, 5)))
             cls.register_cronjob(CronJob(cls.clean_afterjobs, minutes=[random.randint(0, 59)]))
             cls.register_cronjob(CronJob(cls.clean_tempfiles, minutes=[random.randint(0, 59)]))
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-api-check-signature-s-nonce-10923.patch