0002-sms-authorize-numbers-using-masks-39650.patch

Nicolas Roche, 29 mai 2020 12:35

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

Subject: [PATCH 2/2] sms: authorize numbers using masks (#39650)

 passerelle/sms/models.py | 38 +++++++++++++++++++++
 tests/test_sms.py        | 71 +++++++++++++++++++++++++++++++++++++++-
 2 files changed, 108 insertions(+), 1 deletion(-)

     # -*- coding: utf-8 -*-
     # passerelle - uniform access to multiple data sources and services
     # Copyright (C) 2020  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.
+    #
-...
                 elif number.startswith(self.default_trunk_prefix):
                     number = '00' + self.default_country_code + number[len(self.default_trunk_prefix):]
                 else:
                     raise APIError('phone number %r is unsupported (no international prefix, '
                                    'no local trunk prefix)' % number)
                 numbers.append(number)
             return numbers
         def authorize_numbers(self, destinations):
             unknown_numbers = set()
             premium_numbers = set()
             regexes = []
             if SMSResource.ALL not in self.authorized:
                 if SMSResource.FR_METRO in self.authorized:
                     regexes.append(r'0033[67]\d{8}')    # France
                 if SMSResource.FR_DOMTOM in self.authorized:
                     regexes.append(r'00262262\d{6}')    # Réunion, Mayotte, Terres australe/antarctiques
                     regexes.append(r'508508\d{6}')      # Saint-Pierre-et-Miquelon
                     regexes.append(r'590590\d{6}')      # Guadeloupe, Saint-Barthélemy, Saint-Martin
                     regexes.append(r'594594\d{6}')      # Guyane
                     regexes.append(r'596596\d{6}')      # Martinique
                     regexes.append(r'00687[67]\d{8}')   # Nouvelle-Calédonie
                 if SMSResource.BE_ in self.authorized:
                     regexes.append(r'00324[5-9]\d{7}')  # Belgian
                 unknown_numbers = set(destinations)
                 for value in regexes:
                     regex = re.compile(value)
                     unknown_numbers = set(dest for dest in unknown_numbers if not regex.match(dest))
                 for number in list(unknown_numbers):
                     logging.warning('phone number does not match any authorization mask: %s', number)
             if SMSResource.NO_ == self.premium_rate:
                 regex = re.compile(r'0033[8]\d{8}')
                 premium_numbers = set(dest for dest in destinations if regex.match(dest))
             for number in list(premium_numbers):
                 logging.warning('primium rate phone number: %s', number)
             unauthorized_numbers = list(unknown_numbers.union(premium_numbers))
             if unauthorized_numbers != []:
                 raise APIError('phone numbers not authorized: %s' % ', '.join(unauthorized_numbers))
             return True
         @endpoint(perm='can_send_messages', methods=['post'])
         def send(self, request, *args, **kwargs):
             try:
                 data = json_loads(request.body)
                 assert isinstance(data, dict), 'JSON payload is not a dict'
                 assert 'message' in data, 'missing "message" in JSON payload'
                 assert 'from' in data, 'missing "from" in JSON payload'
                 assert 'to' in data, 'missing "to" in JSON payload'
-...
                 assert isinstance(data['from'], six.text_type), 'from is not a string'
                 assert all(map(lambda x: isinstance(x, six.text_type), data['to'])), \
                     'to is not a list of strings'
             except (ValueError, AssertionError) as e:
                 raise APIError('Payload error: %s' % e)
             data['message'] = data['message'][:self.max_message_length]
             data['to'] = self.clean_numbers(data['to'])
             logging.info('sending SMS to %r from %r', data['to'], data['from'])
             self.authorize_numbers(data['to'])
             stop = not bool('nostop' in request.GET)
             self.add_job('send_job',
                          text=data['message'], sender=data['from'], destinations=data['to'],
                          stop=stop)
             return {'err': 0}
         def send_job(self, *args, **kwargs):
             self.send_msg(**kwargs)

     # 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 isodate
     import logging
     import mock
     import pytest
     from requests import RequestException
     from django.contrib.contenttypes.models import ContentType
     from django.utils.translation import ugettext as _
     from passerelle.apps.choosit.models import ChoositSMSGateway
     from passerelle.apps.ovh.models import OVHSMSGateway
     from passerelle.base.models import ApiUser, AccessRight, Job
     from passerelle.base.models import ApiUser, AccessRight, Job, ResourceLog
     from passerelle.sms.models import SMSResource, SMSLog
     from passerelle.utils.jsonresponse import APIError
     from test_manager import login, admin_user
     import utils
     pytestmark = pytest.mark.django_db
-...
         connector.save()
         assert connector.clean_numbers(['+ 33 12']) == ['003312']
         assert connector.clean_numbers(['0 0 33 12']) == ['003312']
         assert connector.clean_numbers(['1 12']) == ['003212']
         with pytest.raises(APIError, match='phone number %r is unsupported' % '0123'):
             connector.clean_numbers(['0123'])
     def test_authorize_numbers():
         connector = OVHSMSGateway()
         # premium-rate
         assert connector.premium_rate == SMSResource.NO_
         number = '0033' + '8' + '12345678'
         with pytest.raises(APIError, match='phone numbers not authorized: %s' % number):
             assert connector.authorize_numbers([number])
         connector.premium_rate = SMSResource.YES
         connector.save()
         # All country
         assert connector.authorized == [SMSResource.ALL]
         number = '0033' + '1' + '12345678'
         assert connector.authorize_numbers([number])
         connector.authorized = [SMSResource.FR_METRO]
         connector.save()
         with pytest.raises(APIError, match='phone numbers not authorized: %s' % number):
             connector.authorize_numbers([number])
         # France
         number = '0033' + '6' + '12345678'
         assert connector.authorize_numbers([number])
         connector.authorized = [SMSResource.FR_DOMTOM]
         connector.save()
         with pytest.raises(APIError, match='phone numbers not authorized: %s' % number):
             connector.authorize_numbers([number])
         # Dom-Tom
         number = '596596' + '123456'
         assert connector.authorize_numbers([number])
         connector.authorized = [SMSResource.BE_]
         connector.save()
         with pytest.raises(APIError, match='phone numbers not authorized: %s' % number):
             connector.authorize_numbers([number])
         # Belgian
         number = '0032' + '45' + '1234567'
         assert connector.authorize_numbers([number])
         connector.authorized = [SMSResource.FR_METRO]
         connector.save()
         with pytest.raises(APIError, match='phone numbers not authorized: %s' % number):
             connector.authorize_numbers([number])
     @pytest.fixture(params=klasses)
     def connector(request, db):
         klass = request.param
         kwargs = getattr(klass, 'TEST_DEFAULTS', {}).get('create_kwargs', {})
         kwargs.update({
             'title': klass.__name__,
             'slug': klass.__name__.lower(),
             'description': klass.__name__,
-...
             'There were errors processing your form.'
         resp.html.find('div', {'class': 'error'}).text.strip() == 'This field is required.'
         resp.form['authorized'] = [SMSResource.FR_METRO, SMSResource.FR_DOMTOM]
         resp = resp.form.submit()
         resp = resp.follow()
         assert _('France mainland (+33 [67])') in [
             x.text for x in resp.html.find_all('p')
             if x.text.startswith(_('Authorized Countries'))][0]
         path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)
         payload = {
             'message': 'plop',
             'from': '+33699999999',
             'to': ['+33688888888'],
+        }
         app.post_json(path, params=payload)
         with mock.patch.object(type(connector), 'send_msg') as send_function:
             send_function.return_value = {}
             connector.jobs()
         assert SMSLog.objects.count() == 1
         payload['to'][0] = '+33188888888'
         SMSLog.objects.all().delete()
         app.post_json(path, params=payload)
         with mock.patch.object(type(connector), 'send_msg') as send_function:
             send_function.return_value = {}
             connector.jobs()
         assert not SMSLog.objects.count()
         assert ResourceLog.objects.filter(levelno=logging.WARNING).count() == 1
         assert ResourceLog.objects.filter(levelno=30)[0].extra['exception'] == \
             'phone numbers not authorized: 0033188888888'
+    -

Projet

Général

Profil

Produits Entr'ouvert » Passerelle

0002-sms-authorize-numbers-using-masks-39650.patch