Produits Entr'ouvert » Chrono

# timeout used in python-requests call, in seconds

# we use 28s by default: timeout just before web server, which is usually 30s

REQUESTS_TIMEOUT = 28

# combo - content management system

# Copyright (C) 2015-2018  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import hashlib

import logging

from requests import Response, Session as RequestsSession

from requests.auth import AuthBase

from django.conf import settings

from django.core.cache import cache

from django.utils.encoding import smart_bytes

from django.utils.http import urlencode

from django.utils.six.moves.urllib import parse as urlparse

from django.utils.six import BytesIO

from .signature import sign_url

class NothingInCacheException(Exception):

    pass

class PublikSignature(AuthBase):

    def __init__(self, secret):

        self.secret = secret

    def __call__(self, request):

        request.url = sign_url(request.url, self.secret)

        return request

class Requests(RequestsSession):

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

        remote_service = kwargs.pop('remote_service', None)

        cache_duration = kwargs.pop('cache_duration', 15)

        invalidate_cache = kwargs.pop('invalidate_cache', False)

        user = kwargs.pop('user', None)

        django_request = kwargs.pop('django_request', None)

        without_user = kwargs.pop('without_user', False)

        federation_key = kwargs.pop('federation_key', 'auto')  # 'auto', 'email', 'nameid'

        raise_if_not_cached = kwargs.pop('raise_if_not_cached', False)

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

        # don't use persistent cookies

        self.cookies.clear()

        if remote_service == 'auto':

            remote_service = None

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

            for services in 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:

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

        if remote_service:

            if isinstance(user, dict):

                query_params = user.copy()

            elif not user or not user.is_authenticated:

                if without_user:

                    query_params = {}

                else:

                    query_params = {'NameID': '', 'email': ''}

            else:

                query_params = {}

                if federation_key == 'nameid':

                    query_params['NameID'] = user.get_name_id()

                elif federation_key == 'email':

                    query_params['email'] = user.email

                else:  # 'auto'

                    user_name_id = user.get_name_id()

                    if user_name_id:

                        query_params['NameID'] = user_name_id

                    else:

                        query_params['email'] = user.email

            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('?', 1)

                query += '&' + old_query

            else:

                path = url

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

        if method == 'GET' and cache_duration:

            # handle cache

            cache_key = hashlib.md5(smart_bytes(url)).hexdigest()

            cache_content = cache.get(cache_key)

            if cache_content and not invalidate_cache:

                response = Response()

                response.status_code = 200

                response.raw = BytesIO(smart_bytes(cache_content))

                return response

            elif raise_if_not_cached:

                raise NothingInCacheException()

        if remote_service:  # sign

            kwargs['auth'] = PublikSignature(remote_service.get('secret'))

        kwargs['timeout'] = kwargs.get('timeout') or settings.REQUESTS_TIMEOUT

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

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

            extra = {}

            if django_request:

                extra['request'] = django_request

            if log_errors == 'warn':

                logging.warning(

                    'failed to %s %s (%s)', method, response.request.url, response.status_code, extra=extra

                )

            else:

                logging.error(

                    'failed to %s %s (%s)', method, response.request.url, response.status_code, extra=extra

                )

        if method == 'GET' and cache_duration and (response.status_code // 100 == 2):

            cache.set(cache_key, response.content, cache_duration)

        return response

requests = Requests()

# combo - content management system

# Copyright (C) 2015-2018  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import base64

import datetime

import hmac

import hashlib

import random

from django.conf import settings

from django.utils.encoding import smart_bytes

from django.utils.http import quote, urlencode

from django.utils import six

from django.utils.six.moves.urllib import parse as urlparse

# 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(smart_bytes(key), digestmod=digestmod, msg=smart_bytes(s))

    return hash.digest()

def check_request_signature(django_request, keys=[]):

    query_string = django_request.META['QUERY_STRING']

    if not query_string:

        return False

    orig = django_request.GET.get('orig', '')

    known_services = getattr(settings, 'KNOWN_SERVICES', None)

    if known_services and orig:

        for services in known_services.values():

            for service in services.values():

                if 'verif_orig' in service and service['verif_orig'] == orig:

                    keys.append(service['secret'])

                    break

    return check_query(query_string, keys)

def check_query(query, keys, known_nonce=None, timedelta=30):

    parsed = urlparse.parse_qs(query)

    if not ('signature' in parsed and 'algo' in parsed and 'timestamp' in parsed and 'nonce' in parsed):

        return False

    unsigned_query, signature_content = query.split('&signature=', 1)

    if '&' in signature_content:

        return False  # signature must be the last parameter

    signature = base64.b64decode(parsed['signature'][0])

    algo = parsed['algo'][0]

    timestamp = parsed['timestamp'][0]

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

    nonce = parsed['nonce']

    if known_nonce is not None and known_nonce(nonce):

        return False

    if abs(datetime.datetime.utcnow() - timestamp) > datetime.timedelta(seconds=timedelta):

        return False

    return check_string(unsigned_query, signature, keys, algo=algo)

def check_string(s, signature, keys, algo='sha256'):

    if not isinstance(keys, list):

        keys = [keys]

    for key in keys:

        signature2 = sign_string(s, key, algo=algo)

        if len(signature2) != len(signature):

            continue

        res = 0

        # constant time compare

        for a, b in zip(signature, signature2):

            res |= a ^ b

        if res == 0:

            return True

    return False

KNOWN_SERVICES = {

    'wcs': {

        'default': {

            'title': 'test',

            'url': 'http://example.org',

            'secret': 'chrono',

            'orig': 'chrono',

            'backoffice-menu-url': 'http://example.org/manage/',

        }

    },

}

# -*- coding: utf-8 -*-

import mock

import pytest

from django.contrib.auth.models import AnonymousUser

from django.utils.six.moves.urllib import parse as urlparse

from chrono.utils.requests_wrapper import requests, NothingInCacheException

from chrono.utils.signature import check_query

class MockUser(object):

    email = 'foo@example.net'

    is_authenticated = True

    def get_name_id(self):

        if self.samlized:

            return 'r2d2'

        return None

    def __init__(self, samlized=True):

        self.samlized = samlized

def test_nosign():

    with mock.patch('chrono.utils.requests_wrapper.RequestsSession.send') as send:

        requests.get('http://example.org/foo/bar/')

        assert send.call_args[0][0].url == 'http://example.org/foo/bar/'

def test_sign():

    remote_service = {'url': 'http://example.org', 'secret': 'secret', 'orig': 'myself'}

    with mock.patch('chrono.utils.requests_wrapper.RequestsSession.send') as send:

        requests.get('/foo/bar/', remote_service=remote_service)

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

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

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['orig'][0] == 'myself'

        assert query['email'][0] == ''

        assert query['NameID'][0] == ''

        assert check_query(querystring, 'secret') == True

        requests.get('/foo/bar/', remote_service=remote_service, without_user=True)

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

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

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['orig'][0] == 'myself'

        assert 'email' not in query

        assert 'NameID' not in query

        assert check_query(querystring, 'secret') == True

def test_auto_sign():

    with mock.patch('chrono.utils.requests_wrapper.RequestsSession.send') as send:

        requests.get('http://example.org/foo/bar/', remote_service='auto')

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

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

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['orig'][0] == 'chrono'

        assert check_query(querystring, 'chrono') == True

        requests.get('http://doesnotexist/foo/bar/', remote_service='auto')

        assert send.call_args[0][0].url == 'http://doesnotexist/foo/bar/'

def test_sign_user():

    remote_service = {'url': 'http://example.org', 'secret': 'secret', 'orig': 'myself'}

    with mock.patch('chrono.utils.requests_wrapper.RequestsSession.send') as send:

        user = MockUser(samlized=True)

        requests.get('/foo/bar/', remote_service=remote_service, user=user)

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

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

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['NameID'][0] == 'r2d2'

        assert 'email' not in query

        assert query['orig'][0] == 'myself'

        assert check_query(querystring, 'secret') == True

        requests.get('/foo/bar/', remote_service=remote_service, user=user, federation_key='email')

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

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

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['email'][0] == 'foo@example.net'

        assert 'NameID' not in query

        assert query['orig'][0] == 'myself'

        assert check_query(querystring, 'secret') == True

        user = MockUser(samlized=False)

        requests.get('/foo/bar/', remote_service=remote_service, user=user)

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

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

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert 'NameID' not in query

        assert query['email'][0] == 'foo@example.net'

        assert query['orig'][0] == 'myself'

        assert check_query(querystring, 'secret') == True

def test_sign_anonymous_user():

    remote_service = {'url': 'http://example.org', 'secret': 'secret', 'orig': 'myself'}

    with mock.patch('chrono.utils.requests_wrapper.RequestsSession.send') as send:

        user = AnonymousUser()

        requests.get('/foo/bar/', remote_service=remote_service, user=user)

        url = send.call_args[0][0].url

        assert url.startswith('http://example.org/foo/bar/?')

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

        query = urlparse.parse_qs(querystring, keep_blank_values=True)

        assert query['NameID'][0] == ''

        assert query['email'][0] == ''

        assert query['orig'][0] == 'myself'

        assert check_query(querystring, 'secret') == True

def test_requests_cache():

    with mock.patch('chrono.utils.requests_wrapper.RequestsSession.request') as requests_get:

        requests_get.return_value = mock.Mock(content=b'hello world', status_code=200)

        # default cache, nothing in there

        assert requests.get('http://cache.example.org/').content == b'hello world'

        assert requests_get.call_count == 1

        # now there's something in cache

        assert requests.get('http://cache.example.org/').content == b'hello world'

        assert requests_get.call_count == 1

        # value changed

        requests_get.return_value = mock.Mock(content=b'hello second world', status_code=200)

        assert requests.get('http://cache.example.org/').content == b'hello world'

        assert requests_get.call_count == 1

        # force cache invalidation

        assert (

            requests.get('http://cache.example.org/', invalidate_cache=True).content == b'hello second world'

        )

        assert requests_get.call_count == 2

        # check raise_if_not_cached

        with pytest.raises(NothingInCacheException):

            requests.get('http://cache.example.org/other', raise_if_not_cached=True)

        # check with unicode url

        assert requests.get(u'http://cache.example.org/éléphant').content == b'hello second world'