Refactor source IP resolution logic

docs/release-notes/version-3.3.md

@@ -21,10 +21,10 @@
 * [#7120](https://github.com/netbox-community/netbox/issues/7120) - Add `termination_date` field to Circuit
 * [#7744](https://github.com/netbox-community/netbox/issues/7744) - Add `status` field to Location
 * [#8222](https://github.com/netbox-community/netbox/issues/8222) - Enable the assignment of a VM to a specific host device within a cluster
+* [#8233](https://github.com/netbox-community/netbox/issues/8233) - Restrict API token access by source IP
 * [#8471](https://github.com/netbox-community/netbox/issues/8471) - Add `status` field to Cluster
 * [#8495](https://github.com/netbox-community/netbox/issues/8495) - Enable custom field grouping
 * [#8995](https://github.com/netbox-community/netbox/issues/8995) - Enable arbitrary ordering of REST API results
-* [#8233](https://github.com/netbox-community/netbox/issues/8233) - Restrict API key access by source IP
 * [#9166](https://github.com/netbox-community/netbox/issues/9166) - Add UI visibility toggle for custom fields
 * [#9582](https://github.com/netbox-community/netbox/issues/9582) - Enable assigning config contexts based on device location
 

netbox/netbox/api/authentication.py

@@ -1,41 +1,37 @@
 from django.conf import settings
-from django.core.exceptions import ValidationError
 from rest_framework import authentication, exceptions
 from rest_framework.permissions import BasePermission, DjangoObjectPermissions, SAFE_METHODS
 
 from users.models import Token
+from utilities.request import get_client_ip
 
 
 class TokenAuthentication(authentication.TokenAuthentication):
     """
-    A custom authentication scheme which enforces Token expiration times.
+    A custom authentication scheme which enforces Token expiration times and source IP restrictions.
     """
     model = Token
 
     def authenticate(self, request):
-        authenticationresult = super().authenticate(request)
+        result = super().authenticate(request)
-        if authenticationresult:
-            token_user, token = authenticationresult
 
-            # Verify source IP is allowed
+        if result:
+            token = result[1]
+
+            # Enforce source IP restrictions (if any) set on the token
             if token.allowed_ips:
-                # Replace 'HTTP_X_REAL_IP' with the settings variable choosen in #8867
+                client_ip = get_client_ip(request)
-                if 'HTTP_X_REAL_IP' in request.META:
+                if client_ip is None:
-                    clientip = request.META['HTTP_X_REAL_IP'].split(",")[0].strip()
+                    raise exceptions.AuthenticationFailed(
-                    http_header = 'HTTP_X_REAL_IP'
+                        "Client IP address could not be determined for validation. Check that the HTTP server is "
-                elif 'REMOTE_ADDR' in request.META:
+                        "correctly configured to pass the required header(s)."
-                    clientip = request.META['REMOTE_ADDR']
+                    )
-                    http_header = 'REMOTE_ADDR'
+                if not token.validate_client_ip(client_ip):
-                else:
+                    raise exceptions.AuthenticationFailed(
-                    raise exceptions.AuthenticationFailed(f"A HTTP header containing the SourceIP (HTTP_X_REAL_IP, REMOTE_ADDR) is missing from the request.")
+                        f"Source IP {client_ip} is not permitted to authenticate using this token."
+                    )
 
-                try:
+        return result
-                    if not token.validate_client_ip(clientip):
-                        raise exceptions.AuthenticationFailed(f"Source IP {clientip} is not allowed to use this token.")
-                except ValidationError as ValidationErrorInfo:
-                    raise exceptions.ValidationError(f"The value in the HTTP Header {http_header} has a ValidationError: {ValidationErrorInfo.message}")
-
-        return authenticationresult
 
     def authenticate_credentials(self, key):
         model = self.get_model()

netbox/users/api/serializers.py

@@ -67,7 +67,10 @@ class TokenSerializer(ValidatedModelSerializer):
 
     class Meta:
         model = Token
-        fields = ('id', 'url', 'display', 'user', 'created', 'expires', 'key', 'write_enabled', 'description', 'allowed_ips')
+        fields = (
+            'id', 'url', 'display', 'user', 'created', 'expires', 'key', 'write_enabled', 'description',
+            'allowed_ips',
+        )
 
     def to_internal_value(self, data):
         if 'key' not in data:

netbox/users/forms.py

@@ -101,11 +101,12 @@ class TokenForm(BootstrapMixin, forms.ModelForm):
         required=False,
         help_text="If no key is provided, one will be generated automatically."
     )
-
     allowed_ips = SimpleArrayField(
         base_field=IPNetworkFormField(),
         required=False,
-        help_text='Allowed IPv4/IPv6 networks from where the token can be used. Leave blank for no restrictions. Ex: "10.1.1.0/24, 192.168.10.16/32, 2001:DB8:1::/64"',
+        label='Allowed IPs',
+        help_text='Allowed IPv4/IPv6 networks from where the token can be used. Leave blank for no restrictions. '
+                  'Ex: "10.1.1.0/24, 192.168.10.16/32, 2001:DB8:1::/64"',
     )
 
     class Meta:

netbox/users/models.py

@@ -223,7 +223,9 @@ class Token(models.Model):
         base_field=IPNetworkField(),
         blank=True,
         null=True,
-        help_text='Allowed IPv4/IPv6 networks from where the token can be used. Leave blank for no restrictions. Ex: "10.1.1.0/24, 192.168.10.16/32, 2001:DB8:1::/64"',
+        verbose_name='Allowed IPs',
+        help_text='Allowed IPv4/IPv6 networks from where the token can be used. Leave blank for no restrictions. '
+                  'Ex: "10.1.1.0/24, 192.168.10.16/32, 2001:DB8:1::/64"',
     )
 
     class Meta:
@@ -249,20 +251,15 @@ class Token(models.Model):
             return False
         return True
 
-    def validate_client_ip(self, raw_ip_address):
+    def validate_client_ip(self, client_ip):
         """
-        Checks that an IP address falls within the allowed IPs.
+        Validate the API client IP address against the source IP restrictions (if any) set on the token.
         """
         if not self.allowed_ips:
             return True
 
-        try:
-            ip_address = ipaddress.ip_address(raw_ip_address)
-        except ValueError as e:
-            raise ValidationError(str(e))
-
         for ip_network in self.allowed_ips:
-            if ip_address in ipaddress.ip_network(ip_network):
+            if client_ip in ipaddress.ip_network(ip_network):
                 return True
 
         return False

netbox/utilities/request.py

@@ -0,0 +1,27 @@
+import ipaddress
+
+__all__ = (
+    'get_client_ip',
+)
+
+
+def get_client_ip(request, additional_headers=()):
+    """
+    Return the client (source) IP address of the given request.
+    """
+    HTTP_HEADERS = (
+        'HTTP_X_REAL_IP',
+        'HTTP_X_FORWARDED_FOR',
+        'REMOTE_ADDR',
+        *additional_headers
+    )
+    for header in HTTP_HEADERS:
+        if header in request.META:
+            client_ip = request.META[header].split(',')[0]
+            try:
+                return ipaddress.ip_address(client_ip)
+            except ValueError:
+                raise ValueError(f"Invalid IP address set for {header}: {client_ip}")
+
+    # Could not determine the client IP address from request headers
+    return None