Projet

Général

Profil

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

Benjamin Dauvergne, 06 août 2019 11:35

Télécharger (12 ko)

Voir les différences: 

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


 combo/utils/requests_wrapper.py  | 14 ++++++++++++--
 tests/test_lingo_remote_regie.py | 24 ++++++++++++------------
 tests/test_newsletters_cell.py   | 12 ++++++------
 tests/test_requests.py           | 28 ++++++++++++++--------------
 4 files changed, 44 insertions(+), 34 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_lingo_remote_regie.py
    
  





  83
  83
  
    
        return 'r2d2'

  		




  84
  84
  
    

  		




  85
  85
  
    

  		




  86
  
  
    
@mock.patch('combo.utils.requests_wrapper.RequestsSession.request')

  		




  87
  
  
    
def test_remote_regie_active_invoices_cell(mock_request, remote_regie):

  		




  
  86
  
    
@mock.patch('combo.utils.requests_wrapper.RequestsSession.send')

  		




  
  87
  
    
def test_remote_regie_active_invoices_cell(mock_send, remote_regie):

  		




  88
  88
  
    
    assert remote_regie.is_remote() == True

  		




  89
  89
  
    

  		




  90
  90
  
    
    page = Page(title='xxx', slug='test_basket_cell', template_name='standard')

  		




  ......




  103
  103
  
    
    ws_invoices = {'err': 0, 'data': INVOICES}

  		




  104
  104
  
    
    mock_response = mock.Mock(status_code=200, content=json.dumps(ws_invoices))

  		




  105
  105
  
    
    mock_response.json.return_value = ws_invoices

  		




  106
  
  
    
    mock_request.return_value = mock_response

  		




  
  106
  
    
    mock_send.return_value = mock_response

  		




  107
  107
  
    
    content = cell.render(context)

  		




  108
  108
  
    
    assert 'F-2016-One' in content

  		




  109
  109
  
    
    assert '123.45' in content

  		




  110
  110
  
    

  		




  111
  111
  
    
    assert '?page=%s' % page.pk in content

  		




  112
  112
  
    
    # check if regie webservice has been correctly called

  		




  113
  
  
    
    assert mock_request.call_args[0][0] == 'GET'

  		




  114
  
  
    
    url = mock_request.call_args[0][1]

  		




  
  113
  
    
    assert mock_send.call_args[0][0].method == 'GET'

  		




  
  114
  
    
    url = mock_send.call_args[0][0].url

  		




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

  		




  116
  116
  
    
    assert scheme == 'http'

  		




  117
  117
  
    
    assert netloc == 'example.org'

  		




  ......




  125
  125
  
    
    ws_invoices = {'err': 0, 'data': []}

  		




  126
  126
  
    
    mock_response = mock.Mock(status_code=200, content=json.dumps(ws_invoices))

  		




  127
  127
  
    
    mock_response.json.return_value = ws_invoices

  		




  128
  
  
    
    mock_request.return_value = mock_response

  		




  
  128
  
    
    mock_send.return_value = mock_response

  		




  129
  129
  
    
    content = cell.render(context)

  		




  130
  130
  
    
    assert 'No items yet' in content

  		




  131
  131
  
    

  		




  132
  
  
    
@mock.patch('combo.utils.requests_wrapper.RequestsSession.request')

  		




  133
  
  
    
def test_remote_regie_past_invoices_cell(mock_request, remote_regie):

  		




  
  132
  
    
@mock.patch('combo.utils.requests_wrapper.RequestsSession.send')

  		




  
  133
  
    
def test_remote_regie_past_invoices_cell(mock_send, remote_regie):

  		




  134
  134
  
    
    assert remote_regie.is_remote() == True

  		




  135
  135
  
    

  		




  136
  136
  
    
    page = Page(title='xxx', slug='test_basket_cell', template_name='standard')

  		




  ......




  149
  149
  
    
    ws_invoices = {'err': 0, 'data': INVOICES}

  		




  150
  150
  
    
    mock_response = mock.Mock(status_code=200, content=json.dumps(ws_invoices))

  		




  151
  151
  
    
    mock_response.json.return_value = ws_invoices

  		




  152
  
  
    
    mock_request.return_value = mock_response

  		




  
  152
  
    
    mock_send.return_value = mock_response

  		




  153
  153
  
    
    content = cell.render(context)

  		




  154
  154
  
    
    assert 'F-2016-One' in content

  		




  155
  155
  
    
    assert '123.45' in content

  		




  156
  156
  
    

  		




  157
  157
  
    
    # check if regie webservice has been correctly called

  		




  158
  
  
    
    assert mock_request.call_args[0][0] == 'GET'

  		




  159
  
  
    
    url = mock_request.call_args[0][1]

  		




  
  158
  
    
    assert mock_send.call_args[0][0].method == 'GET'

  		




  
  159
  
    
    url = mock_send.call_args[0][0].url

  		




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

  		




  161
  161
  
    
    assert scheme == 'http'

  		




  162
  162
  
    
    assert netloc == 'example.org'

  		




  ......




  170
  170
  
    
    ws_invoices = {'err': 0, 'data': []}

  		




  171
  171
  
    
    mock_response = mock.Mock(status_code=200, content=json.dumps(ws_invoices))

  		




  172
  172
  
    
    mock_response.json.return_value = ws_invoices

  		




  173
  
  
    
    mock_request.return_value = mock_response

  		




  
  173
  
    
    mock_send.return_value = mock_response

  		




  174
  174
  
    
    content = cell.render(context)

  		




  175
  175
  
    
    assert 'No items yet' in content

  		




  176
  176
  
    

  		












  

    
      tests/test_newsletters_cell.py
    
  





  10
  10
  
    
from combo.data.models import Page

  		




  11
  11
  
    
from combo.apps.newsletters.models import NewslettersCell, SubscriptionsSaveError

  		




  12
  12
  
    
from combo.apps.newsletters.forms import NewslettersManageForm

  		




  
  13
  
    
from combo.utils import check_query

  		




  13
  14
  
    

  		




  14
  15
  
    
pytestmark = pytest.mark.django_db

  		




  15
  16
  
    

  		




  ......




  147
  148
  
    
    assert cell.get_subscriptions(user) == expected_subscriptions

  		




  148
  149
  
    
    assert mock_get.call_args[1]['user'].email == USER_EMAIL

  		




  149
  150
  
    

  		




  150
  
  
    
@mock.patch('combo.utils.requests_wrapper.RequestsSession.request')

  		




  151
  
  
    
def test_get_subscriptions_signature_check(mock_get, cell, user):

  		




  
  151
  
    
@mock.patch('combo.utils.requests_wrapper.RequestsSession.send')

  		




  
  152
  
    
def test_get_subscriptions_signature_check(mock_send, cell, user):

  		




  152
  153
  
    
    restrictions = ('mail', 'sms')

  		




  153
  154
  
    
    cell.transports_restrictions = ','.join(restrictions)

  		




  154
  155
  
    
    expected_subscriptions = []

  		




  ......




  160
  161
  
    
    mock_json = mock.Mock(status_code=200)

  		




  161
  162
  
    
    mock_json.json.return_value = {'err': 0,

  		




  162
  163
  
    
                                   'data': SUBSCRIPTIONS}

  		




  163
  
  
    
    mock_get.return_value = mock_json

  		




  
  164
  
    
    mock_send.return_value = mock_json

  		




  164
  165
  
    
    cell.get_subscriptions(user)

  		




  165
  
  
    
    assert 'orig=combo' in mock_get.call_args[0][1]

  		




  166
  
  
    
    assert 'signature=' in mock_get.call_args[0][1]

  		




  167
  
  
    
    assert 'nonce=' in mock_get.call_args[0][1]

  		




  
  166
  
    
    url = mock_send.call_args[0][0].url

  		




  
  167
  
    
    assert check_query(url.split('?', 1)[-1], 'combo')

  		




  168
  168
  
    

  		




  169
  169
  
    

  		




  170
  170
  
    
def mocked_requests_connection_error(*args, **kwargs):

  		












  

    
      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
  
  
    
-