darkhelm/notifications-api
1
0
Fork 0
mirror of https://github.com/GSA/notifications-api.git synced 2026-06-13 17:58:46 -04:00
Code Issues Packages Projects Releases Wiki Activity

Validate recipient for restricted service w/ utils

Browse Source 
Implements
https://github.com/alphagov/notifications-utils/pull/16

Once
https://github.com/alphagov/notifications-admin/pull/376
is merged it will no longer be possible for a user to upload a CSV file
containing recipients that they’re not allowed to send to.

So this commit also removes any restricted service checks in the task,
because any public phone numbers/email addresses no longer have any way
of reach this point if the service is restricted.
This commit is contained in:
Chris Hill-Scott
2016-04-04 15:02:25 +01:00
parent 30f0ca3e43
commit e56aee5d1d
7 changed files with 62 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
app/celery/tasks.py
View File

@@ -49,11 +49,6 @@ from app.models import (
TEMPLATE_TYPE_SMS
)
from app.validation import (
allowed_send_to_email,
allowed_send_to_number
)
@notify_celery.task(name="delete-verify-codes")
def delete_verify_codes():
@@ -197,16 +192,6 @@ def send_sms(service_id, notification_id, encrypted_notification, created_at):
client = firetext_client
try:
status = 'sent'
can_send = True
if not allowed_send_to_number(service, notification['to']):
current_app.logger.info(
"SMS {} failed as restricted service".format(notification_id)
)
status = 'failed'
can_send = False
sent_at = datetime.utcnow()
notification_db_object = Notification(
id=notification_id,
@@ -214,7 +199,7 @@ def send_sms(service_id, notification_id, encrypted_notification, created_at):
to=notification['to'],
service_id=service_id,
job_id=notification.get('job', None),
status=status,
status='sent',
created_at=datetime.strptime(created_at, DATETIME_FORMAT),
sent_at=sent_at,
sent_by=client.get_name()
@@ -222,30 +207,29 @@ def send_sms(service_id, notification_id, encrypted_notification, created_at):
dao_create_notification(notification_db_object, TEMPLATE_TYPE_SMS)
if can_send:
try:
template = Template(
dao_get_template_by_id(notification['template']).__dict__,
values=notification.get('personalisation', {}),
prefix=service.name
)
client.send_sms(
to=validate_and_format_phone_number(notification['to']),
content=template.replaced,
reference=str(notification_id)
)
except FiretextClientException as e:
current_app.logger.error(
"SMS notification {} failed".format(notification_id)
)
current_app.logger.exception(e)
notification_db_object.status = 'failed'
dao_update_notification(notification_db_object)
current_app.logger.info(
"SMS {} created at {} sent at {}".format(notification_id, created_at, sent_at)
try:
template = Template(
dao_get_template_by_id(notification['template']).__dict__,
values=notification.get('personalisation', {}),
prefix=service.name
)
client.send_sms(
to=validate_and_format_phone_number(notification['to']),
content=template.replaced,
reference=str(notification_id)
)
except FiretextClientException as e:
current_app.logger.error(
"SMS notification {} failed".format(notification_id)
)
current_app.logger.exception(e)
notification_db_object.status = 'failed'
dao_update_notification(notification_db_object)
current_app.logger.info(
"SMS {} created at {} sent at {}".format(notification_id, created_at, sent_at)
)
except SQLAlchemyError as e:
current_app.logger.debug(e)
@@ -253,21 +237,9 @@ def send_sms(service_id, notification_id, encrypted_notification, created_at):
@notify_celery.task(name="send-email")
def send_email(service_id, notification_id, subject, from_address, encrypted_notification, created_at):
notification = encryption.decrypt(encrypted_notification)
service = dao_fetch_service_by_id(service_id)
client = aws_ses_client
try:
status = 'sent'
can_send = True
if not allowed_send_to_email(service, notification['to']):
current_app.logger.info(
"Email {} failed as restricted service".format(notification_id)
)
status = 'failed'
can_send = False
sent_at = datetime.utcnow()
notification_db_object = Notification(
id=notification_id,
@@ -275,36 +247,35 @@ def send_email(service_id, notification_id, subject, from_address, encrypted_not
to=notification['to'],
service_id=service_id,
job_id=notification.get('job', None),
status=status,
status='sent',
created_at=datetime.strptime(created_at, DATETIME_FORMAT),
sent_at=sent_at,
sent_by=client.get_name()
)
dao_create_notification(notification_db_object, TEMPLATE_TYPE_EMAIL)
if can_send:
try:
template = Template(
dao_get_template_by_id(notification['template']).__dict__,
values=notification.get('personalisation', {})
)
reference = client.send_email(
from_address,
notification['to'],
subject,
body=template.replaced,
html_body=template.as_HTML_email,
)
update_notification_reference_by_id(notification_id, reference)
except AwsSesClientException as e:
current_app.logger.exception(e)
notification_db_object.status = 'failed'
dao_update_notification(notification_db_object)
current_app.logger.info(
"Email {} created at {} sent at {}".format(notification_id, created_at, sent_at)
try:
template = Template(
dao_get_template_by_id(notification['template']).__dict__,
values=notification.get('personalisation', {})
)
reference = client.send_email(
from_address,
notification['to'],
subject,
body=template.replaced,
html_body=template.as_HTML_email,
)
update_notification_reference_by_id(notification_id, reference)
except AwsSesClientException as e:
current_app.logger.exception(e)
notification_db_object.status = 'failed'
dao_update_notification(notification_db_object)
current_app.logger.info(
"Email {} created at {} sent at {}".format(notification_id, created_at, sent_at)
)
except SQLAlchemyError as e:
current_app.logger.debug(e)

21
app/notifications/rest.py
View File

@@ -1,5 +1,6 @@
from datetime import datetime
import uuid
import itertools
from flask import (
Blueprint,
@@ -11,6 +12,7 @@ from flask import (
)
from utils.template import Template
from utils.recipients import allowed_to_send_to, first_column_heading
from app.clients.sms.firetext import FiretextResponses
from app.clients.email.aws_ses import AwsSesResponses
from app import api_user, encryption, create_uuid, DATETIME_FORMAT, DATE_FORMAT
@@ -28,7 +30,6 @@ from app.schemas import (
notifications_filter_schema
)
from app.celery.tasks import send_sms, send_email
from app.validation import allowed_send_to_number, allowed_send_to_email
notifications = Blueprint('notifications', __name__)
@@ -356,12 +357,21 @@ def send_notification(notification_type):
}
), 400
if service.restricted and not allowed_to_send_to(
notification['to'],
itertools.chain.from_iterable(
[user.mobile_number, user.email_address] for user in service.users
)
):
return jsonify(
result="error", message={
'to': ['Invalid {} for restricted service'.format(first_column_heading[notification_type])]
}
), 400
notification_id = create_uuid()
if notification_type == 'sms':
if not allowed_send_to_number(service, notification['to']):
return jsonify(
result="error", message={'to': ['Invalid phone number for restricted service']}), 400
send_sms.apply_async((
service_id,
notification_id,
@@ -369,9 +379,6 @@ def send_notification(notification_type):
datetime.utcnow().strftime(DATETIME_FORMAT)
), queue='sms')
else:
if not allowed_send_to_email(service, notification['to']):
return jsonify(
result="error", message={'to': ['Email address not permitted for restricted service']}), 400
send_email.apply_async((
service_id,
notification_id,

15
app/validation.py
View File

@@ -1,15 +0,0 @@
from utils.recipients import format_phone_number, validate_phone_number
def allowed_send_to_number(service, to):
if service.restricted and format_phone_number(validate_phone_number(to)) not in [
format_phone_number(validate_phone_number(user.mobile_number)) for user in service.users
]:
return False
return True
def allowed_send_to_email(service, to):
if service.restricted and to not in [user.email_address for user in service.users]:
return False
return True

2
requirements.txt
View File

@@ -21,4 +21,4 @@ monotonic==0.3
git+https://github.com/alphagov/notifications-python-client.git@0.2.6#egg=notifications-python-client==0.2.6
git+https://github.com/alphagov/notifications-utils.git@3.1.3#egg=notifications-utils==3.1.3
git+https://github.com/alphagov/notifications-utils.git@3.2.0#egg=notifications-utils==3.2.0

25
tests/app/celery/test_tasks.py
View File

@@ -343,31 +343,6 @@ def test_should_send_sms_if_restricted_service_and_valid_number(notify_db, notif
)
def test_should_not_send_sms_if_restricted_service_and_invalid_number(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)
notification = {
"template": template.id,
"to": "07700 900849"
}
mocker.patch('app.encryption.decrypt', return_value=notification)
mocker.patch('app.firetext_client.send_sms')
mocker.patch('app.firetext_client.get_name', return_value="firetext")
notification_id = uuid.uuid4()
now = datetime.utcnow()
send_sms(
service.id,
notification_id,
"encrypted-in-reality",
now.strftime(DATETIME_FORMAT)
)
firetext_client.send_sms.assert_not_called()
def test_should_send_email_if_restricted_service_and_valid_email(notify_db, notify_db_session, mocker):
user = sample_user(notify_db, notify_db_session, email="test@restricted.com")
service = sample_service(notify_db, notify_db_session, user=user, restricted=True)

4
tests/app/notifications/test_rest.py
View File

@@ -879,7 +879,7 @@ def test_should_not_send_email_if_restricted_and_not_a_service_user(notify_api,
app.celery.tasks.send_email.apply_async.assert_not_called()
assert response.status_code == 400
assert 'Email address not permitted for restricted service' in json_resp['message']['to']
assert 'Invalid email address for restricted service' in json_resp['message']['to']
def test_should_not_send_email_for_job_if_restricted_and_not_a_service_user(
@@ -915,7 +915,7 @@ def test_should_not_send_email_for_job_if_restricted_and_not_a_service_user(
app.celery.tasks.send_email.apply_async.assert_not_called()
assert response.status_code == 400
assert 'Email address not permitted for restricted service' in json_resp['message']['to']
assert 'Invalid email address for restricted service' in json_resp['message']['to']
@freeze_time("2016-01-01 11:09:00.061258")

65
tests/app/test_validation.py
View File

@@ -1,65 +0,0 @@
from app.models import User, Service
from app.validation import allowed_send_to_number, allowed_send_to_email
def test_allowed_send_to_number_returns_true_for_restricted_service_with_same_number():
mobile_number = '07524609792'
service = _create_service_data(mobile_number)
assert allowed_send_to_number(service, mobile_number)
def test_allowed_send_to_number_returns_false_for_restricted_service_with_different_number():
mobile_number = '00447524609792'
service = _create_service_data(mobile_number)
assert not allowed_send_to_number(service, '+447344609793')
def test_allowed_send_to_number_returns_true_for_unrestricted_service_with_different_number():
mobile_number = '+447524609792'
service = _create_service_data(mobile_number, False)
assert allowed_send_to_number(service, '+447344609793')
def test_allowed_send_to_email__returns_true_for_restricted_service_with_same_email():
email = 'testing@it.gov.uk'
service = _create_service_data(email_address=email)
assert allowed_send_to_email(service, email)
def test_allowed_send_to_email__returns_false_for_restricted_service_with_different_email():
email = 'testing@it.gov.uk'
service = _create_service_data(email_address=email)
assert not allowed_send_to_email(service, 'another@it.gov.uk')
def test_allowed_send_to_email__returns_false_for_restricted_service_with_different_email():
email = 'testing@it.gov.uk'
service = _create_service_data(email_address=email)
assert not allowed_send_to_email(service, 'another@it.gov.uk')
def test_allowed_send_to_email__returns_true_for_unrestricted_service_with_different_email():
email = 'testing@it.gov.uk'
service = _create_service_data(email_address=email, restricted=False)
assert allowed_send_to_number(service, 'another@it.gov.uk')
def _create_service_data(mobile_number='+447524609792', restricted=True, email_address='test_user@it.gov.uk'):
usr = {
'name': 'Test User',
'email_address': email_address,
'password': 'password',
'mobile_number': mobile_number,
'state': 'active'
}
user = User(**usr)
data = {
'name': 'Test service',
'limit': 10,
'active': False,
'restricted': restricted,
'email_from': 'test_service@it.gov.uk'
}
service = Service(**data)
service.users = [user]
return service