Merge pull request #2654 from alphagov/validate-content-header-for-post-requests

Add validation for missing content-type header

app/schema_validation/__init__.py

@@ -7,7 +7,6 @@ from jsonschema import (Draft7Validator, ValidationError, FormatChecker)
 from notifications_utils.recipients import (validate_phone_number, validate_email_address, InvalidPhoneError,
                                             InvalidEmailError)
 
-
 format_checker = FormatChecker()
 
 

app/v2/notifications/post_notifications.py

@@ -3,7 +3,6 @@ import functools
 
 from flask import request, jsonify, current_app, abort
 from notifications_utils.recipients import try_validate_and_format_phone_number
-from werkzeug.exceptions import BadRequest
 
 from app import api_user, authenticated_service, notify_celery, document_download_client
 from app.celery.letters_pdf_tasks import create_letters_pdf, process_virus_scan_passed
@@ -58,14 +57,16 @@ from app.v2.notifications.notification_schemas import (
     post_letter_request,
     post_precompiled_letter_request
 )
+from app.v2.utils import get_valid_json
 
 
 @v2_notification_blueprint.route('/{}'.format(LETTER_TYPE), methods=['POST'])
 def post_precompiled_letter_notification():
-    if 'content' not in (request.get_json() or {}):
+    request_json = get_valid_json()
+    if 'content' not in (request_json or {}):
         return post_notification(LETTER_TYPE)
 
-    form = validate(request.get_json(), post_precompiled_letter_request)
+    form = validate(request_json, post_precompiled_letter_request)
 
     # Check permission to send letters
     check_service_has_permission(LETTER_TYPE, authenticated_service.permissions)
@@ -99,11 +100,7 @@ def post_precompiled_letter_notification():
 
 @v2_notification_blueprint.route('/<notification_type>', methods=['POST'])
 def post_notification(notification_type):
-    try:
-        request_json = request.get_json()
-    except BadRequest as e:
-        raise BadRequestError(message="Error decoding arguments: {}".format(e.description),
-                              status_code=400)
+    request_json = get_valid_json()
 
     if notification_type == EMAIL_TYPE:
         form = validate(request_json, post_email_request)

app/v2/template/post_template.py

@@ -6,14 +6,21 @@ from app.schema_validation import validate
 from app.utils import get_template_instance
 from app.v2.errors import BadRequestError
 from app.v2.template import v2_template_blueprint
-from app.v2.template.template_schemas import post_template_preview_request, create_post_template_preview_response
+from app.v2.template.template_schemas import (
+    post_template_preview_request,
+    create_post_template_preview_response
+)
+from app.v2.utils import get_valid_json
 
 
 @v2_template_blueprint.route("/<template_id>/preview", methods=['POST'])
 def post_template_preview(template_id):
-    _data = request.get_json()
-    if _data is None:
+    # The payload is empty when there are no place holders in the template.
+    _data = request.get_data(as_text=True)
+    if not _data:
         _data = {}
+    else:
+        _data = get_valid_json()
 
     _data['id'] = template_id
 

app/v2/utils.py

@@ -0,0 +1,13 @@
+from flask import request
+
+from app.v2.errors import BadRequestError
+from werkzeug.exceptions import BadRequest
+
+
+def get_valid_json():
+    try:
+        request_json = request.get_json(force=True)
+    except BadRequest:
+        raise BadRequestError(message="Invalid JSON supplied in POST data",
+                              status_code=400)
+    return request_json or {}

tests/app/v2/notifications/test_post_letter_notifications.py

@@ -496,3 +496,23 @@ def test_post_letter_notification_throws_error_for_invalid_postage(client, notif
     assert resp_json['errors'][0]['message'] == "postage invalid. It must be either first or second."
 
     assert not Notification.query.first()
+
+
+@pytest.mark.parametrize('content_type',
+                         ['application/json', 'application/text'])
+def test_post_letter_notification_when_payload_is_invalid_json_returns_400(
+        client, sample_service, content_type):
+    auth_header = create_authorization_header(service_id=sample_service.id)
+    payload_not_json = {
+        "template_id": "dont-convert-to-json",
+    }
+    response = client.post(
+        path='/v2/notifications/letter',
+        data=payload_not_json,
+        headers=[('Content-Type', content_type), auth_header],
+    )
+
+    assert response.status_code == 400
+    error_msg = json.loads(response.get_data(as_text=True))["errors"][0]["message"]
+
+    assert error_msg == 'Invalid JSON supplied in POST data'

tests/app/v2/notifications/test_post_notifications.py

@@ -871,3 +871,67 @@ def test_post_notification_returns_400_when_get_json_throws_exception(client, sa
         data="[",
         headers=[('Content-Type', 'application/json'), auth_header])
     assert response.status_code == 400
+
+
+@pytest.mark.parametrize('notification_type, content_type',
+                         [('email', 'application/json'),
+                          ('email', 'application/text'),
+                          ('sms', 'application/json'),
+                          ('sms', 'application/text')]
+                         )
+def test_post_notification_when_payload_is_invalid_json_returns_400(
+        client, sample_service, notification_type, content_type):
+    auth_header = create_authorization_header(service_id=sample_service.id)
+    payload_not_json = {
+        "template_id": "dont-convert-to-json",
+    }
+    response = client.post(
+        path='/v2/notifications/{}'.format(notification_type),
+        data=payload_not_json,
+        headers=[('Content-Type', content_type), auth_header],
+    )
+
+    assert response.status_code == 400
+    error_msg = json.loads(response.get_data(as_text=True))["errors"][0]["message"]
+
+    assert error_msg == 'Invalid JSON supplied in POST data'
+
+
+@pytest.mark.parametrize('notification_type', ['email', 'sms'])
+def test_post_notification_returns_201_when_content_type_is_missing_but_payload_is_valid_json(
+        client, sample_service, notification_type, mocker):
+    template = create_template(service=sample_service, template_type=notification_type)
+    mocker.patch('app.celery.provider_tasks.deliver_{}.apply_async'.format(notification_type))
+    auth_header = create_authorization_header(service_id=sample_service.id)
+
+    valid_json = {
+        "template_id": str(template.id),
+    }
+    if notification_type == 'email':
+        valid_json.update({"email_address": sample_service.users[0].email_address})
+    else:
+        valid_json.update({"phone_number": "+447700900855"})
+    response = client.post(
+        path='/v2/notifications/{}'.format(notification_type),
+        data=json.dumps(valid_json),
+        headers=[auth_header],
+    )
+    assert response.status_code == 201
+
+
+@pytest.mark.parametrize('notification_type', ['email', 'sms'])
+def test_post_email_notification_when_data_is_empty_returns_400(
+        client, sample_service, notification_type):
+    auth_header = create_authorization_header(service_id=sample_service.id)
+    data = None
+    response = client.post(
+        path='/v2/notifications/{}'.format(notification_type),
+        data=json.dumps(data),
+        headers=[('Content-Type', 'application/json'), auth_header],
+    )
+    error_msg = json.loads(response.get_data(as_text=True))["errors"][0]["message"]
+    assert response.status_code == 400
+    if notification_type == 'sms':
+        assert error_msg == 'phone_number is a required property'
+    else:
+        assert error_msg == 'email_address is a required property'

tests/app/v2/template/test_post_template.py

@@ -152,3 +152,42 @@ def test_post_template_with_non_existent_template_id_returns_404(client, fake_uu
         ],
         "status_code": 404
     }
+
+
+def test_post_template_returns_200_without_personalisation(client, sample_template):
+    response = client.post(
+        path='/v2/template/{}/preview'.format(sample_template.id),
+        data=None,
+        headers=[('Content-Type', 'application/json'),
+                 create_authorization_header(service_id=sample_template.service_id)]
+    )
+    assert response.status_code == 200
+
+
+def test_post_template_returns_200_without_personalisation_and_missing_content_header(client, sample_template):
+    response = client.post(
+        path='/v2/template/{}/preview'.format(sample_template.id),
+        data=None,
+        headers=[create_authorization_header(service_id=sample_template.service_id)]
+    )
+    assert response.status_code == 200
+
+
+def test_post_template_returns_200_without_personalisation_as_valid_json_and_missing_content_header(
+        client, sample_template
+):
+    response = client.post(
+        path='/v2/template/{}/preview'.format(sample_template.id),
+        data=json.dumps(None),
+        headers=[create_authorization_header(service_id=sample_template.service_id)]
+    )
+    assert response.status_code == 200
+
+
+def test_post_template_returns_200_with_valid_json_and_missing_content_header(client, sample_template):
+    response = client.post(
+        path='/v2/template/{}/preview'.format(sample_template.id),
+        data=json.dumps(valid_personalisation),
+        headers=[create_authorization_header(service_id=sample_template.service_id)]
+    )
+    assert response.status_code == 200