Fix IP anonymization for ipv6 client addresses

magiclink/helpers.py

@@ -10,7 +10,7 @@
 
 from . import settings
 from .models import MagicLink, MagicLinkError
-from .utils import get_client_ip, get_url_path
+from .utils import get_client_ip, get_url_path, anonymize_ip_address
 
 
 def create_magiclink(
@@ -37,7 +37,7 @@ def create_magiclink(
     if settings.REQUIRE_SAME_IP:
         client_ip = get_client_ip(request)
         if client_ip and settings.ANONYMIZE_IP:
-            client_ip = client_ip[:client_ip.rfind('.')+1] + '0'
+            client_ip = anonymize_ip_address(client_ip)
 
     expiry = timezone.now() + timedelta(seconds=settings.AUTH_TIMEOUT)
     magic_link = MagicLink.objects.create(

magiclink/models.py

@@ -12,7 +12,7 @@
 from django.utils import timezone
 
 from . import settings
-from .utils import get_client_ip
+from .utils import get_client_ip, anonymize_ip_address
 
 User = get_user_model()
 
@@ -111,7 +111,7 @@ def validate(
         if settings.REQUIRE_SAME_IP:
             client_ip = get_client_ip(request)
             if client_ip and settings.ANONYMIZE_IP:
-                client_ip = client_ip[:client_ip.rfind('.')+1] + '0'
+                client_ip = anonymize_ip_address(client_ip)
             if self.ip_address != client_ip:
                 self.disable()
                 raise MagicLinkError('IP address is different from the IP '

magiclink/utils.py

@@ -1,6 +1,10 @@
 from django.http import HttpRequest
 from django.urls import reverse
 from django.urls.exceptions import NoReverseMatch
+import ipaddress
+import logging
+
+log = logging.getLogger(__name__)
 
 
 def get_client_ip(request: HttpRequest) -> str:
@@ -11,6 +15,22 @@ def get_client_ip(request: HttpRequest) -> str:
         ip = request.META.get('REMOTE_ADDR')
     return ip
 
+def anonymize_ip_address(ip_address: str) -> str:
+    try:
+        parsed_ip = ipaddress.ip_address(ip_address)
+    except ValueError as err:
+        log.warning(f'Failed to anonymize ip address: {err}')
+        return ip_address
+
+    if isinstance(parsed_ip, ipaddress.IPv4Address):
+        # Anonymize IPv4 by zeroing out the last octet
+        anonymized_ip = ipaddress.IPv4Address(int(parsed_ip) & 0xFFFFFF00)
+
+    elif isinstance(parsed_ip, ipaddress.IPv6Address):
+        # Anonymize IPv6 by zeroing out the last 80 bits
+        anonymized_ip = ipaddress.IPv6Address(int(parsed_ip) & (0xFFFFFFFFFFFFFFFFFFFF << 80))
+
+    return str(anonymized_ip)
 
 def get_url_path(url: str) -> str:
     """

tests/test_helpers.py

@@ -33,6 +33,22 @@ def test_create_magiclink(settings, freezer):
     assert len(magic_link.cookie_value) == 36
     assert magic_link.ip_address == '127.0.0.0'  # Anonymize IP by default
 
+@pytest.mark.django_db
+def test_create_magiclink_ipv6_remote_addr(settings, freezer):
+    freezer.move_to('2000-01-01T00:00:00')
+
+    email = '[email protected]'
+    expiry = timezone.now() + timedelta(seconds=mlsettings.AUTH_TIMEOUT)
+    request = HttpRequest()
+    request.META['REMOTE_ADDR'] = '2001:0db8:85a3:0000:0000:8a2e:0370:7334'
+    magic_link = create_magiclink(email, request)
+    assert magic_link.email == email
+    assert len(magic_link.token) == mlsettings.TOKEN_LENGTH
+    assert magic_link.expiry == expiry
+    assert magic_link.redirect_url == reverse(settings.LOGIN_REDIRECT_URL)
+    assert len(magic_link.cookie_value) == 36
+    assert magic_link.ip_address == '2001:db8:85a3::'  # Anonymize IP by default
+
 
 @pytest.mark.django_db
 def test_create_magiclink_require_same_ip_off_no_ip(settings):

tests/test_models.py

@@ -201,6 +201,20 @@ def test_validate_wrong_ip(user, magic_link):  # NOQA: F811
     assert ml.disabled is True
 
 
+@pytest.mark.django_db
+def test_validate_ipv6_anonymization(user, magic_link):  # NOQA: F811
+    request = HttpRequest()
+    request.META['REMOTE_ADDR'] = '2001:0db8:85a3:0000:0000:8a2e:0370:7334'
+
+    ml = magic_link(request)
+    request.COOKIES[f'magiclink{ml.pk}'] = ml.cookie_value
+    ml.ip_address = '2001:db8:85a3::'
+    ml.save()
+
+    ml_user = ml.validate(request=request, email=user.email.upper())
+    assert ml_user == user
+
+
 @pytest.mark.django_db
 def test_validate_different_browser(user, magic_link):  # NOQA: F811
     request = HttpRequest()

tests/test_utils.py

@@ -1,6 +1,6 @@
 from django.http import HttpRequest
 
-from magiclink.utils import get_client_ip, get_url_path
+from magiclink.utils import get_client_ip, get_url_path, anonymize_ip_address
 
 
 def test_get_client_ip_http_x_forwarded_for():
@@ -29,3 +29,19 @@ def test_get_url_path_with_path():
     url_name = '/test/'
     url = get_url_path(url_name)
     assert url == '/test/'
+
+def test_anonymize_ip_address_ipv4():
+    ipv4 = '127.0.0.1'
+    anonymized_ipv4 = anonymize_ip_address(ipv4)
+    assert anonymized_ipv4 == '127.0.0.0' # last octet zeroed
+
+
+def test_anonymize_ip_address_ipv6():
+    ipv6 = '2001:0db8:85a3:0000:0000:8a2e:0370:7334'
+    anonymized_ipv6 = anonymize_ip_address(ipv6)
+    assert anonymized_ipv6 == '2001:db8:85a3::' # last 80 bits (SLA ID + Interface ID) zeroed
+
+def test_anonymize_ip_address_invalid_value():
+    bad_input = '127'
+    result = anonymize_ip_address(bad_input)
+    assert result == bad_input # returns original input