Merge pull request #689 from alphagov/service-whitelist

Service whitelist

app/celery/tasks.py

@@ -1,10 +1,8 @@
-import itertools
 from datetime import (datetime)
 
 from flask import current_app
 from notifications_utils.recipients import (
-    RecipientCSV,
+    RecipientCSV
-    allowed_to_send_to
 )
 from notifications_utils.template import Template
 from sqlalchemy.exc import SQLAlchemyError
@@ -31,6 +29,7 @@ from app.models import (
     KEY_TYPE_NORMAL,
     KEY_TYPE_TEST
 )
+from app.service.utils import service_allowed_to_send_to
 from app.statsd_decorators import statsd
 
 
@@ -181,15 +180,3 @@ def send_email(self, service_id,
                 "RETRY FAILED: task send_email failed for notification {}".format(notification.id),
                 e
             )
-
-
-def service_allowed_to_send_to(recipient, service, key_type):
-    if not service.restricted or key_type == KEY_TYPE_TEST:
-        return True
-
-    return allowed_to_send_to(
-        recipient,
-        itertools.chain.from_iterable(
-            [user.mobile_number, user.email_address] for user in service.users
-        )
-    )

app/dao/dao_utils.py

@@ -1,6 +1,7 @@
 import itertools
 from functools import wraps, partial
 
+from app import db
 from app.history_meta import create_history
 
 
@@ -35,3 +36,7 @@ def version_class(model_class, history_cls=None):
                 db.session.add(h_obj)
         return record_version
     return versioned
+
+
+def dao_rollback():
+    db.session.rollback()

app/dao/service_whitelist_dao.py

@@ -0,0 +1,17 @@
+from app import db
+from app.models import Service, ServiceWhitelist
+
+
+def dao_fetch_service_whitelist(service_id):
+    return ServiceWhitelist.query.filter(
+        ServiceWhitelist.service_id == service_id).all()
+
+
+def dao_add_and_commit_whitelisted_contacts(objs):
+    db.session.add_all(objs)
+    db.session.commit()
+
+
+def dao_remove_service_whitelist(service_id):
+    return ServiceWhitelist.query.filter(
+        ServiceWhitelist.service_id == service_id).delete()

app/models.py

@@ -1,11 +1,18 @@
 import uuid
 import datetime
+
 from sqlalchemy.dialects.postgresql import (
     UUID,
     JSON
 )
-from sqlalchemy import UniqueConstraint, text, ForeignKeyConstraint, and_
+from sqlalchemy import UniqueConstraint, and_
 from sqlalchemy.orm import foreign, remote
+from notifications_utils.recipients import (
+    validate_email_address,
+    validate_phone_number,
+    InvalidPhoneError,
+    InvalidEmailError
+)
 
 from app.encryption import (
     hashpw,
@@ -131,6 +138,47 @@ class Service(db.Model, Versioned):
         default=BRANDING_GOVUK
     )
 
+MOBILE_TYPE = 'mobile'
+EMAIL_TYPE = 'email'
+
+WHITELIST_RECIPIENT_TYPE = [MOBILE_TYPE, EMAIL_TYPE]
+whitelist_recipient_types = db.Enum(*WHITELIST_RECIPIENT_TYPE, name='recipient_type')
+
+
+class ServiceWhitelist(db.Model):
+    __tablename__ = 'service_whitelist'
+
+    id = db.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    service_id = db.Column(UUID(as_uuid=True), db.ForeignKey('services.id'), index=True, nullable=False)
+    service = db.relationship('Service', backref='whitelist')
+    recipient_type = db.Column(whitelist_recipient_types, nullable=False)
+    recipient = db.Column(db.String(255), nullable=False)
+    created_at = db.Column(db.DateTime, default=datetime.datetime.utcnow)
+
+    @classmethod
+    def from_string(cls, service_id, recipient_type, recipient):
+        instance = cls(service_id=service_id, recipient_type=recipient_type)
+
+        try:
+            if recipient_type == MOBILE_TYPE:
+                validate_phone_number(recipient)
+                instance.recipient = recipient
+            elif recipient_type == EMAIL_TYPE:
+                validate_email_address(recipient)
+                instance.recipient = recipient
+            else:
+                raise ValueError('Invalid recipient type')
+        except InvalidPhoneError:
+            raise ValueError('Invalid whitelist: "{}"'.format(recipient))
+        except InvalidEmailError:
+            raise ValueError('Invalid whitelist: "{}"'.format(recipient))
+        else:
+            return instance
+
+        def __repr__(self):
+            return 'Recipient {} of type: {}'.format(self.recipient,
+                                                     self.recipient_type)
+
 
 class ApiKey(db.Model, Versioned):
     __tablename__ = 'api_keys'

app/notifications/rest.py

@@ -9,7 +9,7 @@ from flask import (
     json
 )
 
-from notifications_utils.recipients import allowed_to_send_to, first_column_heading
+from notifications_utils.recipients import first_column_heading
 from notifications_utils.template import Template
 from notifications_utils.renderers import PassThrough
 from app.clients.email.aws_ses import get_aws_responses
@@ -27,6 +27,7 @@ from app.notifications.process_client_response import (
     validate_callback_data,
     process_sms_client_response
 )
+from app.service.utils import service_allowed_to_send_to
 from app.schemas import (
     email_notification_schema,
     sms_template_notification_schema,
@@ -252,16 +253,7 @@ def send_notification(notification_type):
         errors = {'content': [message]}
         raise InvalidRequest(errors, status_code=400)
 
-    if all((
+    if not service_allowed_to_send_to(notification['to'], service, api_user.key_type):
-        api_user.key_type != KEY_TYPE_TEST,
-        service.restricted or api_user.key_type == KEY_TYPE_TEAM,
-        not allowed_to_send_to(
-            notification['to'],
-            itertools.chain.from_iterable(
-                [user.mobile_number, user.email_address] for user in service.users
-            )
-        )
-    )):
         if (api_user.key_type == KEY_TYPE_TEAM):
             message = 'Can’t send to this recipient using a team-only API key'
         else:
@@ -276,7 +268,6 @@ def send_notification(notification_type):
 
     notification_id = create_uuid()
     notification.update({"template_version": template.version})
-
     if not _simulated_recipient(notification['to'], notification_type):
         persist_notification(
             service,

app/service/rest.py

@@ -8,6 +8,7 @@ from flask import (
 )
 from sqlalchemy.orm.exc import NoResultFound
 
+from app.dao.dao_utils import dao_rollback
 from app.dao.api_key_dao import (
     save_model_api_key,
     get_model_api_keys,
@@ -26,9 +27,20 @@ from app.dao.services_dao import (
     dao_fetch_weekly_historical_stats_for_service,
     dao_fetch_todays_stats_for_all_services
 )
+from app.dao.service_whitelist_dao import (
+    dao_fetch_service_whitelist,
+    dao_add_and_commit_whitelisted_contacts,
+    dao_remove_service_whitelist
+)
 from app.dao import notifications_dao
 from app.dao.provider_statistics_dao import get_fragment_count
 from app.dao.users_dao import get_model_users
+from app.errors import (
+    register_errors,
+    InvalidRequest
+)
+from app.service import statistics
+from app.service.utils import get_whitelist_objects
 from app.schemas import (
     service_schema,
     api_key_schema,
@@ -39,11 +51,6 @@ from app.schemas import (
     detailed_service_schema
 )
 from app.utils import pagination_links
-from app.errors import (
-    register_errors,
-    InvalidRequest
-)
-from app.service import statistics
 
 service_blueprint = Blueprint('service', __name__)
 register_errors(service_blueprint)
@@ -270,3 +277,36 @@ def get_detailed_services():
             service.statistics = statistics.create_zeroed_stats_dicts()
 
     return detailed_service_schema.dump(services.values(), many=True).data
+
+
+@service_blueprint.route('/<uuid:service_id>/whitelist', methods=['GET'])
+def get_whitelist(service_id):
+    from app.models import (EMAIL_TYPE, MOBILE_TYPE)
+    service = dao_fetch_service_by_id(service_id)
+
+    if not service:
+        raise InvalidRequest("Service does not exist", status_code=404)
+
+    whitelist = dao_fetch_service_whitelist(service.id)
+    return jsonify(
+        email_addresses=[item.recipient for item in whitelist
+                         if item.recipient_type == EMAIL_TYPE],
+        phone_numbers=[item.recipient for item in whitelist
+                       if item.recipient_type == MOBILE_TYPE]
+    )
+
+
+@service_blueprint.route('/<uuid:service_id>/whitelist', methods=['PUT'])
+def update_whitelist(service_id):
+    # doesn't commit so if there are any errors, we preserve old values in db
+    dao_remove_service_whitelist(service_id)
+    try:
+        whitelist_objs = get_whitelist_objects(service_id, request.get_json())
+    except ValueError as e:
+        current_app.logger.exception(e)
+        dao_rollback()
+        msg = '{} is not a valid email address or phone number'.format(str(e))
+        return jsonify(result='error', message=msg), 400
+    else:
+        dao_add_and_commit_whitelisted_contacts(whitelist_objs)
+        return '', 204

app/service/utils.py

@@ -0,0 +1,53 @@
+import itertools
+
+from app.models import (
+    ServiceWhitelist,
+    MOBILE_TYPE, EMAIL_TYPE,
+    KEY_TYPE_TEST, KEY_TYPE_TEAM, KEY_TYPE_NORMAL)
+
+from notifications_utils.recipients import allowed_to_send_to
+
+
+def get_recipients_from_request(request_json, key, type):
+    return [(type, recipient) for recipient in request_json.get(key)]
+
+
+def get_whitelist_objects(service_id, request_json):
+    return [
+        ServiceWhitelist.from_string(service_id, type, recipient)
+        for type, recipient in (
+            get_recipients_from_request(request_json,
+                                        'phone_numbers',
+                                        MOBILE_TYPE) +
+            get_recipients_from_request(request_json,
+                                        'email_addresses',
+                                        EMAIL_TYPE)
+        )
+    ]
+
+
+def service_allowed_to_send_to(recipient, service, key_type):
+    if key_type == KEY_TYPE_TEST:
+        return True
+
+    if key_type == KEY_TYPE_NORMAL and not service.restricted:
+        return True
+
+    team_members = itertools.chain.from_iterable(
+        [user.mobile_number, user.email_address] for user in service.users)
+
+    if key_type == KEY_TYPE_TEAM:
+        return allowed_to_send_to(
+            recipient,
+            team_members
+        )
+
+    if key_type == KEY_TYPE_NORMAL and service.restricted:
+        whitelist_members = [member.recipient for member in service.whitelist]
+        return allowed_to_send_to(
+            recipient,
+            itertools.chain(
+                team_members,
+                whitelist_members
+            )
+        )

migrations/versions/0055_service_whitelist.py

@@ -0,0 +1,31 @@
+"""add service whitelist table
+
+Revision ID: 0055_service_whitelist
+Revises: 0054_perform_drop_status_column
+Create Date: 2016-09-20 12:12:30.838095
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '0055_service_whitelist'
+down_revision = '0054_perform_drop_status_column'
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+def upgrade():
+    op.create_table('service_whitelist',
+        sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column('service_id', postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column('recipient_type', sa.Enum('mobile', 'email', name='recipient_type'), nullable=False),
+        sa.Column('recipient', sa.String(length=255), nullable=True),
+        sa.Column('created_at', sa.DateTime(), nullable=True),
+        sa.ForeignKeyConstraint(['service_id'], ['services.id'], ),
+        sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_service_whitelist_service_id'), 'service_whitelist', ['service_id'], unique=False)
+
+
+def downgrade():
+    op.drop_table('service_whitelist')

tests/app/celery/test_tasks.py

@@ -1,7 +1,8 @@
 import uuid
+import itertools
+import pytest
 from datetime import datetime
 
-import pytest
 from freezegun import freeze_time
 from unittest.mock import ANY
 from sqlalchemy.exc import SQLAlchemyError
@@ -17,7 +18,7 @@ from app.celery.tasks import (
     send_email
 )
 from app.dao import jobs_dao
-from app.models import Notification, KEY_TYPE_TEAM, KEY_TYPE_TEST
+from app.models import Notification, KEY_TYPE_TEAM, KEY_TYPE_TEST, KEY_TYPE_NORMAL
 from tests.app import load_example_csv
 from tests.app.conftest import (
     sample_service,
@@ -455,8 +456,9 @@ def test_should_send_sms_template_to_and_persist_with_job_id(sample_job, sample_
         encryption.encrypt(notification),
         datetime.utcnow().strftime(DATETIME_FORMAT),
         api_key_id=str(sample_api_key.id),
-        key_type=KEY_TYPE_TEAM
+        key_type=KEY_TYPE_NORMAL
     )
+
     provider_tasks.send_sms_to_provider.apply_async.assert_called_once_with(
         (sample_job.service.id,
          notification_id),
@@ -473,14 +475,71 @@ def test_should_send_sms_template_to_and_persist_with_job_id(sample_job, sample_
     assert not persisted_notification.sent_by
     assert persisted_notification.job_row_number == 2
     assert persisted_notification.api_key_id == sample_api_key.id
-    assert persisted_notification.key_type == KEY_TYPE_TEAM
+    assert persisted_notification.key_type == KEY_TYPE_NORMAL
     assert persisted_notification.notification_type == 'sms'
 
 
+def test_should_not_send_email_if_team_key_and_recipient_not_in_team(sample_email_template_with_placeholders,
+                                                                     sample_team_api_key,
+                                                                     mocker):
+    notification = _notification_json(
+        sample_email_template_with_placeholders,
+        "my_email@my_email.com",
+        {"name": "Jo"},
+        row_number=1)
+    notification_id = uuid.uuid4()
+
+    team_members = [user.email_address for user in sample_email_template_with_placeholders.service.users]
+    assert "my_email@my_email.com" not in team_members
+
+    apply_async = mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
+
+    with freeze_time("2016-01-01 11:09:00.00000"):
+        now = datetime.utcnow()
+
+        send_email(
+            sample_email_template_with_placeholders.service_id,
+            notification_id,
+            encryption.encrypt(notification),
+            now.strftime(DATETIME_FORMAT),
+            api_key_id=str(sample_team_api_key.id),
+            key_type=KEY_TYPE_TEAM
+        )
+
+        with pytest.raises(NoResultFound):
+            persisted_notification = Notification.query.filter_by(id=notification_id).one()
+            print(persisted_notification)
+
+    apply_async.not_called()
+
+
+def test_should_not_send_sms_if_team_key_and_recipient_not_in_team(notify_db, notify_db_session, mocker):
+    user = sample_user(notify_db, notify_db_session, mobile_numnber="07700 900205")
+    service = sample_service(notify_db, notify_db_session, user=user, restricted=True)
+    template = sample_template(notify_db, notify_db_session, service=service)
+
+    team_members = [user.mobile_number for user in service.users]
+    assert "07890 300000" not in team_members
+
+    notification = _notification_json(template, "07700 900849")
+    mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
+
+    notification_id = uuid.uuid4()
+    send_sms(
+        service.id,
+        notification_id,
+        encryption.encrypt(notification),
+        datetime.utcnow().strftime(DATETIME_FORMAT)
+    )
+    provider_tasks.send_sms_to_provider.apply_async.assert_not_called()
+    with pytest.raises(NoResultFound):
+        Notification.query.filter_by(id=notification_id).one()
+
+
 def test_should_use_email_template_and_persist(sample_email_template_with_placeholders, sample_api_key, mocker):
     notification = _notification_json(
         sample_email_template_with_placeholders,
-        "my_email@my_email.com",
+        'my_email@my_email.com',
         {"name": "Jo"},
         row_number=1)
     mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
@@ -497,7 +556,7 @@ def test_should_use_email_template_and_persist(sample_email_template_with_placeh
             encryption.encrypt(notification),
             now.strftime(DATETIME_FORMAT),
             api_key_id=str(sample_api_key.id),
-            key_type=KEY_TYPE_TEAM
+            key_type=sample_api_key.key_type
         )
 
     persisted_notification = Notification.query.filter_by(id=notification_id).one()
@@ -516,7 +575,7 @@ def test_should_use_email_template_and_persist(sample_email_template_with_placeh
     assert persisted_notification.personalisation == {'name': 'Jo'}
     assert persisted_notification._personalisation == encryption.encrypt({"name": "Jo"})
     assert persisted_notification.api_key_id == sample_api_key.id
-    assert persisted_notification.key_type == KEY_TYPE_TEAM
+    assert persisted_notification.key_type == KEY_TYPE_NORMAL
     assert persisted_notification.notification_type == 'email'
 
 

tests/app/conftest.py

@@ -19,7 +19,9 @@ from app.models import (
     ProviderStatistics,
     ProviderDetails,
     NotificationStatistics,
-    KEY_TYPE_NORMAL, KEY_TYPE_TEST, KEY_TYPE_TEAM)
+    ServiceWhitelist,
+    KEY_TYPE_NORMAL, KEY_TYPE_TEST, KEY_TYPE_TEAM,
+    MOBILE_TYPE, EMAIL_TYPE)
 from app.dao.users_dao import (save_model_user, create_user_code, create_secret_code)
 from app.dao.services_dao import (dao_create_service, dao_add_user_to_service)
 from app.dao.templates_dao import dao_create_template
@@ -129,9 +131,11 @@ def sample_service(notify_db,
                    user=None,
                    restricted=False,
                    limit=1000,
-                   email_from="sample.service"):
+                   email_from=None):
     if user is None:
         user = sample_user(notify_db, notify_db_session)
+    if email_from is None:
+        email_from = service_name.lower().replace(' ', '.')
     data = {
         'name': service_name,
         'message_limit': limit,
@@ -874,3 +878,20 @@ def already_registered_template(notify_db,
         template = Template(**data)
         db.session.add(template)
     return template
+
+
+@pytest.fixture(scope='function')
+def sample_service_whitelist(notify_db, notify_db_session, service=None, email_address=None, mobile_number=None):
+    if service is None:
+        service = sample_service(notify_db, notify_db_session)
+
+    if email_address:
+        whitelisted_user = ServiceWhitelist.from_string(service.id, EMAIL_TYPE, email_address)
+    elif mobile_number:
+        whitelisted_user = ServiceWhitelist.from_string(service.id, MOBILE_TYPE, mobile_number)
+    else:
+        whitelisted_user = ServiceWhitelist.from_string(service.id, EMAIL_TYPE, 'whitelisted_user@digital.gov.uk')
+
+    notify_db.session.add(whitelisted_user)
+    notify_db.session.commit()
+    return whitelisted_user

tests/app/dao/test_service_whitelist_dao.py

@@ -0,0 +1,56 @@
+import uuid
+
+from app.models import (
+    ServiceWhitelist,
+    MOBILE_TYPE, EMAIL_TYPE)
+
+from app.dao.service_whitelist_dao import (
+    dao_fetch_service_whitelist,
+    dao_add_and_commit_whitelisted_contacts,
+    dao_remove_service_whitelist
+)
+
+from tests.app.conftest import sample_service as create_service
+
+
+def test_fetch_service_whitelist_gets_whitelists(sample_service_whitelist):
+    whitelist = dao_fetch_service_whitelist(sample_service_whitelist.service_id)
+    assert len(whitelist) == 1
+    assert whitelist[0].id == sample_service_whitelist.id
+
+
+def test_fetch_service_whitelist_ignores_other_service(sample_service_whitelist):
+    assert len(dao_fetch_service_whitelist(uuid.uuid4())) == 0
+
+
+def test_add_and_commit_whitelisted_contacts_saves_data(sample_service):
+    whitelist = ServiceWhitelist.from_string(sample_service.id, EMAIL_TYPE, 'foo@example.com')
+
+    dao_add_and_commit_whitelisted_contacts([whitelist])
+
+    db_contents = ServiceWhitelist.query.all()
+    assert len(db_contents) == 1
+    assert db_contents[0].id == whitelist.id
+
+
+def test_remove_service_whitelist_only_removes_for_my_service(notify_db, notify_db_session):
+    service_1 = create_service(notify_db, notify_db_session, service_name="service 1")
+    service_2 = create_service(notify_db, notify_db_session, service_name="service 2")
+    dao_add_and_commit_whitelisted_contacts([
+        ServiceWhitelist.from_string(service_1.id, EMAIL_TYPE, 'service1@example.com'),
+        ServiceWhitelist.from_string(service_2.id, EMAIL_TYPE, 'service2@example.com')
+    ])
+
+    dao_remove_service_whitelist(service_1.id)
+
+    assert service_1.whitelist == []
+    assert len(service_2.whitelist) == 1
+
+
+def test_remove_service_whitelist_does_not_commit(notify_db, sample_service_whitelist):
+    dao_remove_service_whitelist(sample_service_whitelist.service_id)
+
+    # since dao_remove_service_whitelist doesn't commit, we can still rollback its changes
+    notify_db.session.rollback()
+
+    assert ServiceWhitelist.query.count() == 1

tests/app/notifications/rest/test_send_notification.py

@@ -1,7 +1,7 @@
-from datetime import datetime
 import random
 import string
 import pytest
+from datetime import datetime
 
 from flask import (json, current_app)
 from freezegun import freeze_time
@@ -9,7 +9,7 @@ from notifications_python_client.authentication import create_jwt_token
 
 import app
 from app.dao import notifications_dao
-from app.models import ApiKey, KEY_TYPE_TEAM, KEY_TYPE_TEST, Notification, NotificationHistory
+from app.models import ApiKey, KEY_TYPE_NORMAL, KEY_TYPE_TEAM, KEY_TYPE_TEST, Notification, NotificationHistory
 from app.dao.templates_dao import dao_get_all_templates_for_service, dao_update_template
 from app.dao.services_dao import dao_update_service
 from app.dao.api_key_dao import save_model_api_key
@@ -18,7 +18,9 @@ from tests.app.conftest import (
     sample_notification as create_sample_notification,
     sample_service as create_sample_service,
     sample_email_template as create_sample_email_template,
-    sample_template as create_sample_template
+    sample_template as create_sample_template,
+    sample_service_whitelist as create_sample_service_whitelist,
+    sample_api_key as create_sample_api_key
 )
 
 
@@ -998,6 +1000,106 @@ def test_should_not_persist_notification_or_send_sms_if_simulated_number(
     assert Notification.query.count() == 0
 
 
+@pytest.mark.parametrize('notification_type,to, key_type', [
+    ('sms', '07827992635', KEY_TYPE_NORMAL),
+    ('email', 'non_whitelist_recipient@mail.com', KEY_TYPE_NORMAL),
+    ('sms', '07827992635', KEY_TYPE_TEAM),
+    ('email', 'non_whitelist_recipient@mail.com', KEY_TYPE_TEAM)])
+def test_should_not_send_notification_to_non_whitelist_recipient_in_trial_mode(client,
+                                                                               notify_db,
+                                                                               notify_db_session,
+                                                                               notification_type,
+                                                                               to,
+                                                                               key_type,
+                                                                               mocker):
+    service = create_sample_service(notify_db, notify_db_session, limit=2, restricted=True)
+    service_whitelist = create_sample_service_whitelist(notify_db, notify_db_session, service=service)
+
+    if notification_type == 'sms':
+        apply_async = mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
+        template = create_sample_template(notify_db, notify_db_session, service=service)
+
+    elif notification_type == 'email':
+        apply_async = mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
+        template = create_sample_email_template(notify_db, notify_db_session, service=service)
+
+    assert service_whitelist.service_id == service.id
+    assert to not in [member.recipient for member in service.whitelist]
+
+    create_sample_notification(notify_db, notify_db_session, template=template, service=service)
+
+    data = {
+        'to': to,
+        'template': str(template.id)
+    }
+
+    api_key = create_sample_api_key(notify_db, notify_db_session, service, key_type=key_type)
+    auth_header = create_jwt_token(secret=api_key.unsigned_secret, client_id=str(api_key.service_id))
+
+    response = client.post(
+        path='/notifications/{}'.format(notification_type),
+        data=json.dumps(data),
+        headers=[('Content-Type', 'application/json'), ('Authorization', 'Bearer {}'.format(auth_header))])
+
+    expected_response_message = (
+        'Can’t send to this recipient when service is in trial mode '
+        '– see https://www.notifications.service.gov.uk/trial-mode'
+    ) if key_type == KEY_TYPE_NORMAL else ('Can’t send to this recipient using a team-only API key')
+
+    json_resp = json.loads(response.get_data(as_text=True))
+    assert response.status_code == 400
+    assert json_resp['result'] == 'error'
+    assert expected_response_message in json_resp['message']['to']
+    apply_async.assert_not_called()
+
+
+@pytest.mark.parametrize('notification_type,to', [
+    ('sms', '07123123123'),
+    ('email', 'whitelist_recipient@mail.com')])
+def test_should_send_notification_to_whitelist_recipient_in_trial_mode_with_live_key(client,
+                                                                                     notify_db,
+                                                                                     notify_db_session,
+                                                                                     notification_type,
+                                                                                     to,
+                                                                                     mocker):
+    service = create_sample_service(notify_db, notify_db_session, limit=2, restricted=True)
+    if notification_type == 'sms':
+        apply_async = mocker.patch('app.celery.provider_tasks.send_sms_to_provider.apply_async')
+        template = create_sample_template(notify_db, notify_db_session, service=service)
+        service_whitelist = create_sample_service_whitelist(notify_db, notify_db_session,
+                                                            service=service, mobile_number=to)
+    elif notification_type == 'email':
+        apply_async = mocker.patch('app.celery.provider_tasks.send_email_to_provider.apply_async')
+        template = create_sample_email_template(notify_db, notify_db_session, service=service)
+        service_whitelist = create_sample_service_whitelist(notify_db, notify_db_session,
+                                                            service=service, email_address=to)
+
+    assert service_whitelist.service_id == service.id
+    assert to in [member.recipient for member in service.whitelist]
+
+    create_sample_notification(notify_db, notify_db_session, template=template, service=service)
+
+    data = {
+        'to': to,
+        'template': str(template.id)
+    }
+
+    sample_live_key = create_sample_api_key(notify_db, notify_db_session, service)
+    auth_header = create_jwt_token(secret=sample_live_key.unsigned_secret, client_id=str(sample_live_key.service_id))
+
+    response = client.post(
+        path='/notifications/{}'.format(notification_type),
+        data=json.dumps(data),
+        headers=[('Content-Type', 'application/json'), ('Authorization', 'Bearer {}'.format(auth_header))])
+
+    json_resp = json.loads(response.get_data(as_text=True))
+    assert response.status_code == 201
+    assert json_resp['data']['notification']['id']
+    assert json_resp['data']['body'] == template.content
+    assert json_resp['data']['template_version'] == template.version
+    apply_async.called
+
+
 @pytest.mark.parametrize(
     'notification_type, template_type, to', [
         ('email', 'sms', 'notify@digital.cabinet-office.gov.uk'),

tests/app/service/test_service_whitelist.py

@@ -0,0 +1,95 @@
+import uuid
+import json
+
+from tests import create_authorization_header
+
+from app.models import (
+    ServiceWhitelist,
+    MOBILE_TYPE, EMAIL_TYPE)
+
+from app.dao.service_whitelist_dao import dao_add_and_commit_whitelisted_contacts
+
+
+def test_get_whitelist_returns_data(client, sample_service_whitelist):
+    service_id = sample_service_whitelist.service_id
+
+    response = client.get('service/{}/whitelist'.format(service_id), headers=[create_authorization_header()])
+    assert response.status_code == 200
+    assert json.loads(response.get_data(as_text=True)) == {
+        'email_addresses': [sample_service_whitelist.recipient],
+        'phone_numbers': []
+    }
+
+
+def test_get_whitelist_separates_emails_and_phones(client, sample_service):
+    dao_add_and_commit_whitelisted_contacts([
+        ServiceWhitelist.from_string(sample_service.id, EMAIL_TYPE, 'service@example.com'),
+        ServiceWhitelist.from_string(sample_service.id, MOBILE_TYPE, '07123456789')
+    ])
+
+    response = client.get('service/{}/whitelist'.format(sample_service.id), headers=[create_authorization_header()])
+    assert response.status_code == 200
+    assert json.loads(response.get_data(as_text=True)) == {
+        'email_addresses': ['service@example.com'],
+        'phone_numbers': ['07123456789']
+    }
+
+
+def test_get_whitelist_404s_with_unknown_service_id(client):
+    path = 'service/{}/whitelist'.format(uuid.uuid4())
+
+    response = client.get(path, headers=[create_authorization_header()])
+    assert response.status_code == 404
+    json_resp = json.loads(response.get_data(as_text=True))
+    assert json_resp['result'] == 'error'
+    assert json_resp['message'] == 'No result found'
+
+
+def test_get_whitelist_returns_no_data(client, sample_service):
+    path = 'service/{}/whitelist'.format(sample_service.id)
+
+    response = client.get(path, headers=[create_authorization_header()])
+
+    assert response.status_code == 200
+    assert json.loads(response.get_data(as_text=True)) == {'email_addresses': [], 'phone_numbers': []}
+
+
+def test_update_whitelist_replaces_old_whitelist(client, sample_service_whitelist):
+    data = {
+        'email_addresses': ['foo@bar.com'],
+        'phone_numbers': ['07123456789']
+    }
+
+    response = client.put(
+        'service/{}/whitelist'.format(sample_service_whitelist.service_id),
+        data=json.dumps(data),
+        headers=[('Content-Type', 'application/json'), create_authorization_header()]
+    )
+
+    assert response.status_code == 204
+    whitelist = ServiceWhitelist.query.order_by(ServiceWhitelist.recipient).all()
+    assert len(whitelist) == 2
+    assert whitelist[0].recipient == '07123456789'
+    assert whitelist[1].recipient == 'foo@bar.com'
+
+
+def test_update_whitelist_doesnt_remove_old_whitelist_if_error(client, sample_service_whitelist):
+
+    data = {
+        'email_addresses': [''],
+        'phone_numbers': ['07123456789']
+    }
+
+    response = client.put(
+        'service/{}/whitelist'.format(sample_service_whitelist.service_id),
+        data=json.dumps(data),
+        headers=[('Content-Type', 'application/json'), create_authorization_header()]
+    )
+
+    assert response.status_code == 400
+    assert json.loads(response.get_data(as_text=True)) == {
+        'result': 'error',
+        'message': 'Invalid whitelist: "" is not a valid email address or phone number'
+    }
+    whitelist = ServiceWhitelist.query.one()
+    assert whitelist.id == sample_service_whitelist.id

tests/app/test_model.py

@@ -1,7 +1,12 @@
 from datetime import datetime
 
+import pytest
+
 from app import DATETIME_FORMAT
-from app.models import Notification
+from app.models import (
+    Notification,
+    ServiceWhitelist,
+    MOBILE_TYPE, EMAIL_TYPE)
 
 
 def test_should_build_notification_from_minimal_set_of_api_derived_params(notify_api):
@@ -70,3 +75,32 @@ def test_should_build_notification_from_full_set_of_api_derived_params(notify_ap
     assert notification.notification_type == 'SMS'
     assert notification.api_key_id == 'api_key_id'
     assert notification.key_type == 'key_type'
+
+
+@pytest.mark.parametrize('mobile_number', [
+    '07700 900678',
+    '+44 7700 900678'
+])
+def test_should_build_service_whitelist_from_mobile_number(mobile_number):
+    service_whitelist = ServiceWhitelist.from_string('service_id', MOBILE_TYPE, mobile_number)
+
+    assert service_whitelist.recipient == mobile_number
+
+
+@pytest.mark.parametrize('email_address', [
+    'test@example.com'
+])
+def test_should_build_service_whitelist_from_email_address(email_address):
+    service_whitelist = ServiceWhitelist.from_string('service_id', EMAIL_TYPE, email_address)
+
+    assert service_whitelist.recipient == email_address
+
+
+@pytest.mark.parametrize('contact, recipient_type', [
+    ('', None),
+    ('07700dsadsad', MOBILE_TYPE),
+    ('gmail.com', EMAIL_TYPE)
+])
+def test_should_not_build_service_whitelist_from_invalid_contact(recipient_type, contact):
+    with pytest.raises(ValueError):
+        ServiceWhitelist.from_string('service_id', recipient_type, contact)