Validate IP addresses in alert profiles

python/nav/web/alertprofiles/views.py

@@ -24,11 +24,12 @@
 # TODO Filter/filter_groups have owners, check that the account that performs
 # the operation is the owner
 
-from django.http import HttpResponseRedirect
+from django.http import HttpResponseRedirect, QueryDict
 from django.core.exceptions import ObjectDoesNotExist
 from django.db.models import Q
 from django.shortcuts import render
 from django.urls import reverse
+from IPy import IP
 
 from nav.web.utils import SubListView
 
@@ -50,6 +51,7 @@
     AccountAlertQueue,
 )
 from nav.django.utils import get_account, is_admin
+from nav.oidparsers import TypedInetAddress
 from nav.web.message import Messages, new_message
 
 from nav.web.alertprofiles.forms import TimePeriodForm, LanguageForm
@@ -1575,15 +1577,32 @@ def filter_saveexpression(request):
             request, _('You do not own this filter.')
         )
 
+    if match_field.data_type == MatchField.IP:
+        if operator.type == Operator.IN:
+            value_list = request.POST.get('value').split()
+        else:
+            value_list = [request.POST.get('value')]
+        for value in value_list:
+            try:
+                IP(value)
+            except ValueError:
+                new_message(
+                    request,
+                    f"Invalid IP address: {value}",
+                    Messages.ERROR,
+                )
+                request.POST = QueryDict(
+                    f"id={request.POST.get('filter')}&matchfield={request.POST.get('match_field')}"
+                )
+                return filter_addexpression(request=request)
+
     # Get the value
     if operator.type == Operator.IN:
         # If input was a multiple choice list we have to join each option
         # in one string, where each option is separated by a | (pipe).
         # If input was a IP adress we should replace space with | (pipe).
         # FIXME We might want some data checks here
         if match_field.data_type == MatchField.IP:
-            # FIXME We might want to check that it is a valid IP adress.
-            # If we do so, we need to remember both IPv4 and IPv6
             value = request.POST.get('value').replace(' ', '|')
         else:
             value = "|".join([value for value in request.POST.getlist('value')])

tests/integration/web/alertprofiles_test.py

@@ -9,7 +9,15 @@
 from django.urls import reverse
 from nav.compatibility import smart_str
 
-from nav.models.profiles import AlertProfile, Account, AlertPreference
+from nav.models.profiles import (
+    AlertProfile,
+    Account,
+    AlertPreference,
+    Expression,
+    Filter,
+    MatchField,
+    Operator,
+)
 from nav.web.alertprofiles.views import set_active_profile
 
 
@@ -157,6 +165,75 @@ def test_alertprofiles_add_public_filter_should_succeed(client):
     assert response.status_code == 200
 
 
+def test_alertprofiles_add_expression_with_valid_ipv4_address_should_succeed(
+    client, dummy_filter
+):
+    """Tests that an expression with a valid IPv4 address can be added"""
+    ip_match_field = MatchField.objects.get(data_type=MatchField.IP)
+    url = reverse("alertprofiles-filters-saveexpression")
+    data = {
+        "filter": dummy_filter.pk,
+        "match_field": ip_match_field.pk,
+        "operator": Operator.EQUALS,
+        "value": "172.0.0.1",
+    }
+    response = client.post(url, data=data, follow=True)
+    assert response.status_code == 200
+    assert Expression.objects.filter(
+        filter=dummy_filter,
+        match_field=ip_match_field,
+        operator=Operator.EQUALS,
+        value=data["value"],
+    ).exists()
+    assert f"Added expression to filter {dummy_filter}" in smart_str(response.content)
+
+
+def test_alertprofiles_add_expression_with_valid_ipv6_address_should_succeed(
+    client, dummy_filter
+):
+    """Tests that an expression with a valid IPv6 address can be added"""
+    url = reverse("alertprofiles-filters-saveexpression")
+    ip_match_field = MatchField.objects.get(data_type=MatchField.IP)
+    data = {
+        "filter": dummy_filter.pk,
+        "match_field": ip_match_field.pk,
+        "operator": Operator.EQUALS,
+        "value": "2001:db8:3333:4444:5555:6666:7777:8888",
+    }
+    response = client.post(url, data=data, follow=True)
+    assert response.status_code == 200
+    assert Expression.objects.filter(
+        filter=dummy_filter,
+        match_field=ip_match_field,
+        operator=Operator.EQUALS,
+        value=data["value"],
+    ).exists()
+    assert f"Added expression to filter {dummy_filter}" in smart_str(response.content)
+
+
+def test_alertprofiles_add_expression_with_non_valid_ip_address_should_fail(
+    client, dummy_filter
+):
+    """Tests that an expression with a not valid IP address cannot be added"""
+    ip_match_field = MatchField.objects.get(data_type=MatchField.IP)
+    url = reverse("alertprofiles-filters-saveexpression")
+    data = {
+        "filter": dummy_filter.pk,
+        "match_field": ip_match_field.pk,
+        "operator": Operator.EQUALS,
+        "value": "wrong",
+    }
+    response = client.post(url, data=data, follow=True)
+    assert response.status_code == 200
+    assert not Expression.objects.filter(
+        filter=dummy_filter,
+        match_field=ip_match_field,
+        operator=Operator.EQUALS,
+        value=data["value"],
+    ).exists()
+    assert f"Invalid IP address: {data['value']}" in smart_str(response.content)
+
+
 def test_set_accountgroup_permissions_should_not_crash(db, client):
     """Regression test for #2281"""
     url = reverse('alertprofiles-permissions-save')
@@ -191,3 +268,10 @@ def activated_dummy_profile(dummy_profile):
     )
     preference.save()
     return dummy_profile
+
+
+@pytest.fixture(scope="function")
+def dummy_filter():
+    filtr = Filter(name="dummy", owner=Account.objects.get(id=Account.ADMIN_ACCOUNT))
+    filtr.save()
+    return filtr