Produits Entr'ouvert » Corbo

import base64

import datetime

import hmac

import random

from requests import Response, Session as RequestsSession, RequestException

from django.utils.http import urlencode, quote

# Simple signature scheme for query strings

def sign_url(url, key, algo='sha256', timestamp=None, nonce=None):

    parsed = urlparse.urlparse(url)

    new_query = sign_query(parsed.query, key, algo, timestamp, nonce)

    return urlparse.urlunparse(parsed[:4] + (new_query,) + parsed[5:])

def sign_query(query, key, algo='sha256', timestamp=None, nonce=None):

    if timestamp is None:

        timestamp = datetime.datetime.utcnow()

    timestamp = timestamp.strftime('%Y-%m-%dT%H:%M:%SZ')

    if nonce is None:

        nonce = hex(random.getrandbits(128))[2:]

    new_query = query

    if new_query:

        new_query += '&'

    new_query += urlencode((

        ('algo', algo),

        ('timestamp', timestamp),

        ('nonce', nonce)))

    signature = base64.b64encode(sign_string(new_query, key, algo=algo))

    new_query += '&signature=' + quote(signature)

    return new_query

def sign_string(s, key, algo='sha256', timedelta=30):

    digestmod = getattr(hashlib, algo)

    hash = hmac.HMAC(str(key), digestmod=digestmod, msg=s)

    return hash.digest()

class Requests(RequestsSession):

    def request(self, method, url, **kwargs):

        logger = logging.getLogger(__name__)

        log_errors = kwargs.pop('log_errors', True)

        remote_service = None

        scheme, netloc, path, params, query, fragment = urlparse.urlparse(url)

        for services in getattr(settings, 'KNOWN_SERVICES', {}).values():

            for service in services.values():

                remote_url = service.get('url')

                remote_scheme, remote_netloc, r_path, r_params, r_query, r_fragment = \

                        urlparse.urlparse(remote_url)

                if remote_scheme == scheme and remote_netloc == netloc:

                    remote_service = service

                    break

            else:

                continue

            break

        if remote_service:

            # only keeps the path (URI) in url parameter, scheme and netloc are

            # in remote_service

            url = urlparse.urlunparse(('', '', path, params, query, fragment))

        else:

            logger.warning('service not found in settings.KNOWN_SERVICES for %s', url)

        if remote_service:

            query_params = {'orig': remote_service.get('orig')}

            remote_service_base_url = remote_service.get('url')

            scheme, netloc, old_path, params, old_query, fragment = urlparse.urlparse(

                    remote_service_base_url)

            query = urlencode(query_params)

            if '?' in url:

                path, old_query = url.split('?')

                query += '&' + old_query

            else:

                path = url

            url = urlparse.urlunparse((scheme, netloc, path, params, query, fragment))

        if remote_service: # sign

            url = sign_url(url, remote_service.get('secret'))

        response = super(Requests, self).request(method, url, **kwargs)

        if log_errors and (response.status_code // 100 != 2):

            logger.error('failed to %s %s (%s)', method, url, response.status_code)

        return response

requests = Requests()

        except RequestException as e:

@mock.patch('corbo.utils.RequestsSession.request')

def test_sms_send_with_signed_webservice_call(mocked_post, app, categories, announces):

    services = {

        'passerelle': {

            'default': {'title': 'test', 'url': 'http://passerelle.example.org',

                        'secret': 'corbo', 'orig': 'corbo'}

        },

        'corbo': {

            'announces': {

                'title': 'announces', 'url': 'http://corbo.example.org',

                'secret': 'corbo', 'orig': 'corbo',

            }

        }

    }

    for category in categories:

        for i in range(3):

            uuid = uuid4()

            Subscription.objects.create(category=category,

                            identifier='sms:%s' % get_random_number(), uuid=uuid)

    for announce in announces:

        broadcast = Broadcast.objects.get(announce=announce)

        with override_settings(SMS_GATEWAY_URL='http://passerelle.example.org', KNOWN_SERVICES=services):

            mocked_response = mock.Mock(status_code=200)

            mocked_response.json.return_value = {'err': 0, 'err_desc': None, 'data': 'gateway response'}

            mocked_post.return_value = mocked_response

            broadcast.send()

            assert 'http://passerelle.example.org' in mocked_post.call_args[0][1]

            assert 'orig=corbo' in mocked_post.call_args[0][1]

            assert 'signature=' in mocked_post.call_args[0][1]

            assert 'nonce=' in mocked_post.call_args[0][1]

    assert len(records) == 2