Projet

Général

Profil

0001-requests_wrapper-sign-URL-of-prepared-requests-35225.patch

Benjamin Dauvergne, 05 août 2019 19:09

Télécharger (6,53 ko)

Voir les différences: 

Subject: [PATCH] requests_wrapper: sign URL of prepared requests (#35225)


 combo/utils/requests_wrapper.py | 14 ++++++++++++--
 tests/test_requests.py          | 28 ++++++++++++++--------------
 2 files changed, 26 insertions(+), 16 deletions(-)
combo/utils/requests_wrapper.py
18 18 
import logging
19 19 

  		




  20
  20
  
    
from requests import Response, Session as RequestsSession

  		




  
  21
  
    
from requests.auth import AuthBase

  		




  21
  22
  
    

  		




  22
  23
  
    
from django.conf import settings

  		




  23
  24
  
    
from django.core.cache import cache

  		




  ......




  32
  33
  
    
    pass

  		




  33
  34
  
    

  		




  34
  35
  
    

  		




  
  36
  
    
class PublikSignature(AuthBase):

  		




  
  37
  
    
    def __init__(self, secret):

  		




  
  38
  
    
        self.secret = secret

  		




  
  39
  
    

  		




  
  40
  
    
    def __call__(self, request):

  		




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

  		




  
  42
  
    
        return request

  		




  
  43
  
    

  		




  
  44
  
    

  		




  35
  45
  
    
class Requests(RequestsSession):

  		




  36
  46
  
    

  		




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

  		




  ......




  117
  127
  
    
            elif raise_if_not_cached:

  		




  118
  128
  
    
                raise NothingInCacheException()

  		




  119
  129
  
    

  		




  120
  
  
    
        if remote_service: # sign

  		




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

  		




  
  130
  
    
        if remote_service:  # sign

  		




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

  		




  122
  132
  
    

  		




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

  		




  124
  134
  
    

  		












  

    
      tests/test_requests.py
    
  





  23
  23
  
    

  		




  24
  24
  
    

  		




  25
  25
  
    
def test_nosign():

  		




  26
  
  
    
    with mock.patch('combo.utils.requests_wrapper.RequestsSession.request') as request:

  		




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

  		




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

  		




  28
  
  
    
        assert request.call_args[0][1] == 'http://example.org/foo/bar/'

  		




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

  		




  29
  29
  
    

  		




  30
  30
  
    
def test_sign():

  		




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

  		




  32
  
  
    
    with mock.patch('combo.utils.requests_wrapper.RequestsSession.request') as request:

  		




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

  		




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

  		




  34
  
  
    
        url = request.call_args[0][1]

  		




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

  		




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

  		




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

  		




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

  		




  ......




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

  		




  42
  42
  
    

  		




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

  		




  44
  
  
    
        url = request.call_args[0][1]

  		




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

  		




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

  		




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

  		




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

  		




  ......




  52
  52
  
    

  		




  53
  53
  
    

  		




  54
  54
  
    
def test_auto_sign():

  		




  55
  
  
    
    with mock.patch('combo.utils.requests_wrapper.RequestsSession.request') as request:

  		




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

  		




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

  		




  57
  
  
    
        url = request.call_args[0][1]

  		




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

  		




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

  		




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

  		




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

  		




  ......




  62
  62
  
    
        assert check_query(querystring, 'combo') == True

  		




  63
  63
  
    

  		




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

  		




  65
  
  
    
        assert request.call_args[0][1] == 'http://doesnotexist/foo/bar/'

  		




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

  		




  66
  66
  
    

  		




  67
  67
  
    

  		




  68
  68
  
    
def test_sign_user():

  		




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

  		




  70
  
  
    
    with mock.patch('combo.utils.requests_wrapper.RequestsSession.request') as request:

  		




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

  		




  71
  71
  
    

  		




  72
  72
  
    
        user = MockUser(samlized=True)

  		




  73
  73
  
    

  		




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

  		




  75
  
  
    
        url = request.call_args[0][1]

  		




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

  		




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

  		




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

  		




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

  		




  ......




  83
  83
  
    

  		




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

  		




  85
  85
  
    
                     federation_key='email')

  		




  86
  
  
    
        url = request.call_args[0][1]

  		




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

  		




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

  		




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

  		




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

  		




  ......




  96
  96
  
    
        user = MockUser(samlized=False)

  		




  97
  97
  
    

  		




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

  		




  99
  
  
    
        url = request.call_args[0][1]

  		




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

  		




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

  		




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

  		




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

  		




  ......




  108
  108
  
    

  		




  109
  109
  
    
def test_sign_anonymous_user():

  		




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

  		




  111
  
  
    
    with mock.patch('combo.utils.requests_wrapper.RequestsSession.request') as request:

  		




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

  		




  112
  112
  
    

  		




  113
  113
  
    
        user = AnonymousUser()

  		




  114
  114
  
    

  		




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

  		




  116
  
  
    
        url = request.call_args[0][1]

  		




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

  		




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

  		




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

  		




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

  		




  120
  
  
    
-