From 39f3f82970897844512a3bcec7b90bbc22ac0383 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=81lex=20Hermida?= Date: Wed, 10 Oct 2018 13:56:44 +0200 Subject: [PATCH] Add tests --- taiga/base/utils/urls.py | 33 +++++++++++++++++++++++++++++++++ taiga/webhooks/tasks.py | 8 ++++---- taiga/webhooks/utils.py | 21 --------------------- tests/unit/test_utils.py | 32 ++++++++++++++++++++++++++++++++ 4 files changed, 69 insertions(+), 25 deletions(-) delete mode 100644 taiga/webhooks/utils.py diff --git a/taiga/base/utils/urls.py b/taiga/base/utils/urls.py index a145cd21..25a5ad8a 100644 --- a/taiga/base/utils/urls.py +++ b/taiga/base/utils/urls.py @@ -17,8 +17,13 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +import ipaddress +import socket +from urllib.parse import urlparse + import django_sites as sites from django.core.urlresolvers import reverse as django_reverse +from django.utils.translation import ugettext as _ URL_TEMPLATE = "{scheme}://{domain}/{path}" @@ -43,3 +48,31 @@ def get_absolute_url(path): def reverse(viewname, *args, **kwargs): """Same behavior as django's reverse but uses django_sites to compute absolute url.""" return get_absolute_url(django_reverse(viewname, *args, **kwargs)) + + +class HostnameValueError(Exception): + pass + + +class IpAddresValueError(Exception): + pass + + +def validate_destination_address(url): + host = urlparse(url).hostname + port = urlparse(url).port + + try: + socket_args, *others = socket.getaddrinfo(host, port) + except Exception: + raise HostnameValueError(_("Host access error")) + + destination_address = socket_args[4][0] + try: + ipa = ipaddress.ip_address(destination_address) + except ValueError: + raise IpAddresValueError(_("IP Address error")) + if ipa.is_private: + raise IpAddresValueError("Private IP Address not allowed") + + return True diff --git a/taiga/webhooks/tasks.py b/taiga/webhooks/tasks.py index 62bdd8d8..c6487116 100644 --- a/taiga/webhooks/tasks.py +++ b/taiga/webhooks/tasks.py @@ -22,7 +22,7 @@ import requests from requests.exceptions import RequestException from taiga.base.api.renderers import UnicodeJSONRenderer -from taiga.base.utils import json +from taiga.base.utils import json, urls from taiga.base.utils.db import get_typename_for_model_instance from taiga.celery import app @@ -30,7 +30,7 @@ from .serializers import (EpicSerializer, EpicRelatedUserStorySerializer, UserStorySerializer, IssueSerializer, TaskSerializer, WikiPageSerializer, MilestoneSerializer, HistoryEntrySerializer, UserSerializer) -from . import utils + from .models import WebhookLog @@ -74,8 +74,8 @@ def _send_request(webhook_id, url, key, data): } try: - utils.validate_destination_address(url) - except utils.IpaddresValueError as e: + urls.validate_destination_address(url) + except urls.IpAddresValueError as e: # Error validating url webhook_log = WebhookLog.objects.create(webhook_id=webhook_id, url=url, status=0, diff --git a/taiga/webhooks/utils.py b/taiga/webhooks/utils.py deleted file mode 100644 index 0a87462b..00000000 --- a/taiga/webhooks/utils.py +++ /dev/null @@ -1,21 +0,0 @@ -import ipaddress -import socket -from urllib.parse import urlparse - - -class IpaddresValueError(Exception): - pass - - -def validate_destination_address(url): - host = urlparse(url).hostname - port = urlparse(url).port - socket_args, _ = socket.getaddrinfo(host, port) - destination_address = socket_args[4][0] - - try: - ipa = ipaddress.ip_address(destination_address) - except ValueError: - raise IpaddresValueError(_("IP Address error")) - if ipa.is_private: - raise IpaddresValueError("Not allowed IP Address") diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 0c11354a..e57ed8f4 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -107,3 +107,35 @@ def test_to_tsquery(): expected = re.sub("([0-9])", r"'\1':*", expected) actual = to_tsquery(input) assert actual == expected + + +@pytest.mark.parametrize("url", [ + "http://127.0.0.1", + "http://[::1]", + "http://192.168.0.12", + "http://10.0.0.1", + "https://172.25.0.1", + "https://10.25.23.100", + "ftp://192.168.1.100/", + "http://[::ffff:c0a8:164]/", + "scp://192.168.1.100/", + "http://www.192.168.1.100.xip.io/", + "http://test.local/", +]) +def test_validate_bad_destination_address(url): + with pytest.raises(IpAddresValueError): + validate_destination_address(url) + + +@pytest.mark.parametrize("url", [ + "http://192.167.0.12", + "http://11.0.0.1", + "https://173.25.0.1", + "https://193.24.23.100", + "ftp://173.168.1.100/", + "scp://194.168.1.100/", + "http://www.google.com/", + "http://1.1.1.1/", +]) +def test_validate_good_destination_address(url): + assert validate_destination_address(url)