diff --git a/app/broadcast_message/rest.py b/app/broadcast_message/rest.py index ba4858a1d..addd41231 100644 --- a/app/broadcast_message/rest.py +++ b/app/broadcast_message/rest.py @@ -38,8 +38,13 @@ def _parse_nullable_datetime(dt): def _update_broadcast_message(broadcast_message, new_status, updating_user): + if updating_user not in broadcast_message.service.users: + abort( + 400, + f'User {updating_user.id} cannot approve broadcast {broadcast_message.id} from other service' + ) + # TODO: Restrict status transitions - # TODO: validate that the user belongs to the same service, isn't the creator, has permissions, etc if new_status == BroadcastStatusType.BROADCASTING: # TODO: Remove this platform admin shortcut when the feature goes live if updating_user == broadcast_message.created_by and not updating_user.platform_admin: diff --git a/tests/app/broadcast_message/test_rest.py b/tests/app/broadcast_message/test_rest.py index 0cbfb7a13..a1c3eda49 100644 --- a/tests/app/broadcast_message/test_rest.py +++ b/tests/app/broadcast_message/test_rest.py @@ -269,6 +269,7 @@ def test_update_broadcast_message_status_stores_cancelled_by_and_cancelled_at(ad t = create_template(sample_service, BROADCAST_TYPE) bm = create_broadcast_message(t, status=BroadcastStatusType.BROADCASTING) canceller = create_user(email='canceller@gov.uk') + sample_service.users.append(canceller) response = admin_request.post( 'broadcast_message.update_broadcast_message_status', @@ -291,6 +292,7 @@ def test_update_broadcast_message_status_stores_approved_by_and_approved_at_and_ t = create_template(sample_service, BROADCAST_TYPE) bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL) approver = create_user(email='approver@gov.uk') + sample_service.users.append(approver) mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async') response = admin_request.post( @@ -353,3 +355,25 @@ def test_update_broadcast_message_status_allows_platform_admin_to_approve_own_me assert response['created_by_id'] == str(user.id) assert response['approved_by_id'] == str(user.id) mock_task.assert_called_once_with(kwargs={'broadcast_message_id': str(bm.id)}, queue='notify-internal-tasks') + + +def test_update_broadcast_message_status_rejects_approval_from_user_not_on_that_service( + admin_request, + sample_service, + mocker +): + t = create_template(sample_service, BROADCAST_TYPE) + bm = create_broadcast_message(t, status=BroadcastStatusType.PENDING_APPROVAL) + approver = create_user(email='approver@gov.uk') + mock_task = mocker.patch('app.celery.broadcast_message_tasks.send_broadcast_message.apply_async') + + response = admin_request.post( + 'broadcast_message.update_broadcast_message_status', + _data={'status': BroadcastStatusType.BROADCASTING, 'created_by': str(approver.id)}, + service_id=t.service_id, + broadcast_message_id=bm.id, + _expected_status=400 + ) + + assert mock_task.called is False + assert f'cannot approve broadcast' in response['message']