Projet

Général

Profil

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

Nicolas Roche, 01 février 2021 17:05

Télécharger (8,7 ko)

Voir les différences: 

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


 passerelle/sms/models.py | 37 +++++++++++++++++++++
 tests/test_sms.py        | 71 +++++++++++++++++++++++++++++++++++++++-
 2 files changed, 107 insertions(+), 1 deletion(-)
passerelle/sms/models.py
160 160 
            elif number.startswith(self.default_trunk_prefix):
161 161 
                number = '00' + self.default_country_code + number[len(self.default_trunk_prefix):]
162 162 
            else:
163 163 
                raise APIError('phone number %r is unsupported (no international prefix, '
164 164 
                               'no local trunk prefix)' % number)
165 165 
            numbers.append(number)
166 166 
        return numbers
167 167 

  		




  
  168
  
    
    def authorize_numbers(self, destinations):

  		




  
  169
  
    
        unknown_numbers = set()

  		




  
  170
  
    
        premium_numbers = set()

  		




  
  171
  
    
        regexes = []

  		




  
  172
  
    
        if SMSResource.ALL not in self.authorized:

  		




  
  173
  
    
            if SMSResource.FR_METRO in self.authorized:

  		




  
  174
  
    
                regexes.append(r'0033[67]\d{8}')    # France

  		




  
  175
  
    
            if SMSResource.FR_DOMTOM in self.authorized:

  		




  
  176
  
    
                regexes.append(r'00262262\d{6}')    # Réunion, Mayotte, Terres australe/antarctiques

  		




  
  177
  
    
                regexes.append(r'508508\d{6}')      # Saint-Pierre-et-Miquelon

  		




  
  178
  
    
                regexes.append(r'590590\d{6}')      # Guadeloupe, Saint-Barthélemy, Saint-Martin

  		




  
  179
  
    
                regexes.append(r'594594\d{6}')      # Guyane

  		




  
  180
  
    
                regexes.append(r'596596\d{6}')      # Martinique

  		




  
  181
  
    
                regexes.append(r'00687[67]\d{8}')   # Nouvelle-Calédonie

  		




  
  182
  
    
            if SMSResource.BE_ in self.authorized:

  		




  
  183
  
    
                regexes.append(r'00324[5-9]\d{7}')  # Belgian

  		




  
  184
  
    

  		




  
  185
  
    
            unknown_numbers = set(destinations)

  		




  
  186
  
    
            for value in regexes:

  		




  
  187
  
    
                regex = re.compile(value)

  		




  
  188
  
    
                unknown_numbers = set(dest for dest in unknown_numbers if not regex.match(dest))

  		




  
  189
  
    

  		




  
  190
  
    
            for number in list(unknown_numbers):

  		




  
  191
  
    
                logging.warning('phone number does not match any authorization mask: %s', number)

  		




  
  192
  
    

  		




  
  193
  
    
        if SMSResource.NO_ == self.premium_rate:

  		




  
  194
  
    
            regex = re.compile(r'0033[8]\d{8}')

  		




  
  195
  
    
            premium_numbers = set(dest for dest in destinations if regex.match(dest))

  		




  
  196
  
    
        for number in list(premium_numbers):

  		




  
  197
  
    
            logging.warning('primium rate phone number: %s', number)

  		




  
  198
  
    

  		




  
  199
  
    
        unauthorized_numbers = list(unknown_numbers.union(premium_numbers))

  		




  
  200
  
    
        if unauthorized_numbers != []:

  		




  
  201
  
    
            raise APIError('phone numbers not authorized: %s' % ', '.join(unauthorized_numbers))

  		




  
  202
  
    
        return True

  		




  
  203
  
    

  		




  168
  204
  
    
    @endpoint(perm='can_send_messages', methods=['post'],

  		




  169
  205
  
    
              description=_('Send a SMS message'),

  		




  170
  206
  
    
              parameters={'nostop': {'description': _('Do not send STOP instruction'), 'example_value': '1'}},

  		




  171
  207
  
    
              post={'request_body': {'schema': {'application/json': SEND_SCHEMA}}})

  		




  172
  208
  
    
    def send(self, request, post_data, nostop=None):

  		




  173
  209
  
    
        post_data['message'] = post_data['message'][:self.max_message_length]

  		




  174
  210
  
    
        post_data['to'] = self.clean_numbers(post_data['to'])

  		




  
  211
  
    
        self.authorize_numbers(post_data['to'])

  		




  175
  212
  
    
        logging.info('sending SMS to %r from %r', post_data['to'], post_data['from'])

  		




  176
  213
  
    
        stop = nostop is None  # ?nostop in not in query string

  		




  177
  214
  
    
        self.add_job('send_job',

  		




  178
  215
  
    
                     text=post_data['message'], sender=post_data['from'], destinations=post_data['to'],

  		




  179
  216
  
    
                     stop=stop)

  		




  180
  217
  
    
        return {'err': 0}

  		




  181
  218
  
    

  		




  182
  219
  
    
    def send_job(self, *args, **kwargs):

  		












  

    
      tests/test_sms.py
    
  





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

  		




  11
  11
  
    
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

  		




  12
  12
  
    
# GNU Affero General Public License for more details.

  		




  13
  13
  
    
#

  		




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

  		




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

  		




  16
  16
  
    
import isodate

  		




  17
  17
  
    
import json

  		




  
  18
  
    
import logging

  		




  18
  19
  
    
import mock

  		




  19
  20
  
    
import pytest

  		




  20
  21
  
    
from requests import RequestException

  		




  21
  22
  
    

  		




  22
  23
  
    
from django.conf import settings

  		




  23
  24
  
    
from django.contrib.contenttypes.models import ContentType

  		




  24
  25
  
    
from django.urls import reverse

  		




  25
  26
  
    
from django.utils.translation import ugettext as _

  		




  26
  27
  
    

  		




  27
  28
  
    
from passerelle.apps.choosit.models import ChoositSMSGateway

  		




  28
  29
  
    
from passerelle.apps.ovh.models import OVHSMSGateway

  		




  29
  
  
    
from passerelle.base.models import ApiUser, AccessRight, Job

  		




  
  30
  
    
from passerelle.base.models import ApiUser, AccessRight, Job, ResourceLog

  		




  30
  31
  
    
from passerelle.sms.models import SMSResource, SMSLog

  		




  31
  32
  
    
from passerelle.utils.jsonresponse import APIError

  		




  32
  33
  
    

  		




  33
  34
  
    
from test_manager import login

  		




  34
  35
  
    

  		




  35
  36
  
    
import utils

  		




  36
  37
  
    

  		




  37
  38
  
    
pytestmark = pytest.mark.django_db

  		




  ......




  49
  50
  
    
    connector.save()

  		




  50
  51
  
    
    assert connector.clean_numbers(['+ 33 12']) == ['003312']

  		




  51
  52
  
    
    assert connector.clean_numbers(['0 0 33 12']) == ['003312']

  		




  52
  53
  
    
    assert connector.clean_numbers(['1 12']) == ['003212']

  		




  53
  54
  
    
    with pytest.raises(APIError, match='phone number %r is unsupported' % '0123'):

  		




  54
  55
  
    
        connector.clean_numbers(['0123'])

  		




  55
  56
  
    

  		




  56
  57
  
    

  		




  
  58
  
    
def test_authorize_numbers():

  		




  
  59
  
    
    connector = OVHSMSGateway()

  		




  
  60
  
    

  		




  
  61
  
    
    # premium-rate

  		




  
  62
  
    
    assert connector.premium_rate == SMSResource.NO_

  		




  
  63
  
    
    number = '0033' + '8' + '12345678'

  		




  
  64
  
    
    with pytest.raises(APIError, match='phone numbers not authorized: %s' % number):

  		




  
  65
  
    
        assert connector.authorize_numbers([number])

  		




  
  66
  
    
    connector.premium_rate = SMSResource.YES

  		




  
  67
  
    
    connector.save()

  		




  
  68
  
    

  		




  
  69
  
    
    # All country

  		




  
  70
  
    
    assert connector.authorized == [SMSResource.ALL]

  		




  
  71
  
    
    number = '0033' + '1' + '12345678'

  		




  
  72
  
    
    assert connector.authorize_numbers([number])

  		




  
  73
  
    
    connector.authorized = [SMSResource.FR_METRO]

  		




  
  74
  
    
    connector.save()

  		




  
  75
  
    
    with pytest.raises(APIError, match='phone numbers not authorized: %s' % number):

  		




  
  76
  
    
        connector.authorize_numbers([number])

  		




  
  77
  
    

  		




  
  78
  
    
    # France

  		




  
  79
  
    
    number = '0033' + '6' + '12345678'

  		




  
  80
  
    
    assert connector.authorize_numbers([number])

  		




  
  81
  
    
    connector.authorized = [SMSResource.FR_DOMTOM]

  		




  
  82
  
    
    connector.save()

  		




  
  83
  
    
    with pytest.raises(APIError, match='phone numbers not authorized: %s' % number):

  		




  
  84
  
    
        connector.authorize_numbers([number])

  		




  
  85
  
    

  		




  
  86
  
    
    # Dom-Tom

  		




  
  87
  
    
    number = '596596' + '123456'

  		




  
  88
  
    
    assert connector.authorize_numbers([number])

  		




  
  89
  
    
    connector.authorized = [SMSResource.BE_]

  		




  
  90
  
    
    connector.save()

  		




  
  91
  
    
    with pytest.raises(APIError, match='phone numbers not authorized: %s' % number):

  		




  
  92
  
    
        connector.authorize_numbers([number])

  		




  
  93
  
    

  		




  
  94
  
    
    # Belgian

  		




  
  95
  
    
    number = '0032' + '45' + '1234567'

  		




  
  96
  
    
    assert connector.authorize_numbers([number])

  		




  
  97
  
    
    connector.authorized = [SMSResource.FR_METRO]

  		




  
  98
  
    
    connector.save()

  		




  
  99
  
    
    with pytest.raises(APIError, match='phone numbers not authorized: %s' % number):

  		




  
  100
  
    
        connector.authorize_numbers([number])

  		




  
  101
  
    

  		




  
  102
  
    

  		




  57
  103
  
    
@pytest.fixture(params=klasses)

  		




  58
  104
  
    
def connector(request, db):

  		




  59
  105
  
    
    klass = request.param

  		




  60
  106
  
    
    kwargs = getattr(klass, 'TEST_DEFAULTS', {}).get('create_kwargs', {})

  		




  61
  107
  
    
    kwargs.update({

  		




  62
  108
  
    
        'title': klass.__name__,

  		




  63
  109
  
    
        'slug': klass.__name__.lower(),

  		




  64
  110
  
    
        'description': klass.__name__,

  		




  ......




  440
  486
  
    
        'There were errors processing your form.'

  		




  441
  487
  
    
    assert resp.html.find('div', {'class': 'error'}).text.strip() == 'This field is required.'

  		




  442
  488
  
    
    resp.form['authorized'] = [SMSResource.FR_METRO, SMSResource.FR_DOMTOM]

  		




  443
  489
  
    
    resp = resp.form.submit()

  		




  444
  490
  
    
    resp = resp.follow()

  		




  445
  491
  
    
    assert _('France mainland (+33 [67])') in [

  		




  446
  492
  
    
        x.text for x in resp.html.find_all('p')

  		




  447
  493
  
    
        if x.text.startswith(_('Authorized Countries'))][0]

  		




  
  494
  
    

  		




  
  495
  
    
    path = '/%s/%s/send/' % (connector.get_connector_slug(), connector.slug)

  		




  
  496
  
    
    payload = {

  		




  
  497
  
    
        'message': 'plop',

  		




  
  498
  
    
        'from': '+33699999999',

  		




  
  499
  
    
        'to': ['+33688888888'],

  		




  
  500
  
    
    }

  		




  
  501
  
    
    app.post_json(path, params=payload)

  		




  
  502
  
    
    with mock.patch.object(type(connector), 'send_msg') as send_function:

  		




  
  503
  
    
        send_function.return_value = {}

  		




  
  504
  
    
        connector.jobs()

  		




  
  505
  
    
    assert SMSLog.objects.count() == 1

  		




  
  506
  
    

  		




  
  507
  
    
    payload['to'][0] = '+33188888888'

  		




  
  508
  
    
    SMSLog.objects.all().delete()

  		




  
  509
  
    
    app.post_json(path, params=payload)

  		




  
  510
  
    
    with mock.patch.object(type(connector), 'send_msg') as send_function:

  		




  
  511
  
    
        send_function.return_value = {}

  		




  
  512
  
    
        connector.jobs()

  		




  
  513
  
    
    assert not SMSLog.objects.count()

  		




  
  514
  
    
    assert ResourceLog.objects.filter(levelno=logging.WARNING).count() == 1

  		




  
  515
  
    
    assert ResourceLog.objects.filter(levelno=30)[0].extra['exception'] == \

  		




  
  516
  
    
        'phone numbers not authorized: 0033188888888'

  		




  448
  
  
    
-