diff --git a/taiga/base/utils/urls.py b/taiga/base/utils/urls.py index 25a5ad8a..f376482a 100644 --- a/taiga/base/utils/urls.py +++ b/taiga/base/utils/urls.py @@ -54,11 +54,11 @@ class HostnameValueError(Exception): pass -class IpAddresValueError(Exception): +class IpAddresValueError(ValueError): pass -def validate_destination_address(url): +def validate_private_url(url): host = urlparse(url).hostname port = urlparse(url).port @@ -74,5 +74,3 @@ def validate_destination_address(url): raise IpAddresValueError(_("IP Address error")) if ipa.is_private: raise IpAddresValueError("Private IP Address not allowed") - - return True diff --git a/tests/integration/test_webhooks.py b/tests/integration/test_webhooks.py index d4da70c1..23d3d939 100644 --- a/tests/integration/test_webhooks.py +++ b/tests/integration/test_webhooks.py @@ -54,8 +54,9 @@ def test_webhook_action_test_transform_to_json(client, data): response = Mock(status_code=200, headers={}, text="ok") response.elapsed.total_seconds.return_value = 100 - with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock: - client.login(data.project_owner) - response = client.json.post(url) - assert response.status_code == 200 - assert json.loads(response.data["response_data"]) == {"content": "ok"} + with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response), \ + patch("taiga.base.utils.urls.validate_destination_address", return_value=True): + client.login(data.project_owner) + response = client.json.post(url) + assert response.status_code == 200 + assert json.loads(response.data["response_data"]) == {"content": "ok"} diff --git a/tests/integration/test_webhooks_signals.py b/tests/integration/test_webhooks_signals.py index 7c8fe6d9..d41ae8a4 100644 --- a/tests/integration/test_webhooks_signals.py +++ b/tests/integration/test_webhooks_signals.py @@ -44,22 +44,26 @@ def test_new_object_with_one_webhook_signal(settings): response.elapsed.total_seconds.return_value = 100 for obj in objects: - with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock: + with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \ + patch("taiga.base.utils.urls.validate_destination_address", return_value=True): services.take_snapshot(obj, user=obj.owner, comment="test") assert session_send_mock.call_count == 1 for obj in objects: - with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock: + with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \ + patch("taiga.base.utils.urls.validate_destination_address", return_value=True): services.take_snapshot(obj, user=obj.owner) assert session_send_mock.call_count == 0 for obj in objects: - with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock: + with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \ + patch("taiga.base.utils.urls.validate_destination_address", return_value=True): services.take_snapshot(obj, user=obj.owner, comment="test") assert session_send_mock.call_count == 1 for obj in objects: - with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock: + with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \ + patch("taiga.base.utils.urls.validate_destination_address", return_value=True): services.take_snapshot(obj, user=obj.owner, comment="test", delete=True) assert session_send_mock.call_count == 1 @@ -81,22 +85,26 @@ def test_new_object_with_two_webhook_signals(settings): response.elapsed.total_seconds.return_value = 100 for obj in objects: - with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock: + with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \ + patch("taiga.base.utils.urls.validate_destination_address", return_value=True): services.take_snapshot(obj, user=obj.owner, comment="test") assert session_send_mock.call_count == 2 for obj in objects: - with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock: + with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \ + patch("taiga.base.utils.urls.validate_destination_address", return_value=True): services.take_snapshot(obj, user=obj.owner, comment="test") assert session_send_mock.call_count == 2 for obj in objects: - with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock: + with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \ + patch("taiga.base.utils.urls.validate_destination_address", return_value=True): services.take_snapshot(obj, user=obj.owner) assert session_send_mock.call_count == 0 for obj in objects: - with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock: + with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \ + patch("taiga.base.utils.urls.validate_destination_address", return_value=True): services.take_snapshot(obj, user=obj.owner, comment="test", delete=True) assert session_send_mock.call_count == 2 @@ -117,11 +125,13 @@ def test_send_request_one_webhook_signal(settings): response.elapsed.total_seconds.return_value = 100 for obj in objects: - with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock: + with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \ + patch("taiga.base.utils.urls.validate_destination_address", return_value=True): services.take_snapshot(obj, user=obj.owner, comment="test") assert session_send_mock.call_count == 1 for obj in objects: - with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock: + with patch("taiga.webhooks.tasks.requests.Session.send", return_value=response) as session_send_mock, \ + patch("taiga.base.utils.urls.validate_destination_address", return_value=True): services.take_snapshot(obj, user=obj.owner, comment="test", delete=True) assert session_send_mock.call_count == 1 diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index e57ed8f4..c2d5c34c 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -22,7 +22,8 @@ from unittest import mock import django_sites as sites import re -from taiga.base.utils.urls import get_absolute_url, is_absolute_url, build_url +from taiga.base.utils.urls import get_absolute_url, is_absolute_url, build_url, \ + validate_private_url, IpAddresValueError from taiga.base.utils.db import save_in_bulk, update_in_bulk, to_tsquery pytestmark = pytest.mark.django_db @@ -124,7 +125,7 @@ def test_to_tsquery(): ]) def test_validate_bad_destination_address(url): with pytest.raises(IpAddresValueError): - validate_destination_address(url) + validate_private_url(url) @pytest.mark.parametrize("url", [ @@ -138,4 +139,4 @@ def test_validate_bad_destination_address(url): "http://1.1.1.1/", ]) def test_validate_good_destination_address(url): - assert validate_destination_address(url) + assert validate_private_url(url) is None