#8233 Restrict API key usage by source IP

docs/configuration/optional-settings.md

@@ -277,6 +277,15 @@ Note that a plugin must be listed in `PLUGINS` for its configuration to take eff
 
 ---
 
+## PROXY_HEADER_REALIP
+
+Default: HTTP_X_REAL_IP
+
+This parameters sets the HTTP Header that contains the REAL IP of a client that connects through a PROXY. The Real IP is required to validate an API token's Allowed IPRanges.
+Other common values are HTTP_X_FORWARDED_FOR, HTTP_X_CLIENT_IP
+
+---
+
 ## RELEASE_CHECK_URL
 
 Default: None (disabled)

netbox/netbox/api/authentication.py

@@ -11,13 +11,49 @@ class TokenAuthentication(authentication.TokenAuthentication):
     """
     model = Token
 
-    def authenticate_credentials(self, key):
+    def authenticate(self, request):
+        auth = authentication.get_authorization_header(request).split()
+
+        if not auth or auth[0].lower() != self.keyword.lower().encode():
+            return None
+
+        if len(auth) == 1:
+            msg = 'Invalid token header. No credentials provided.'
+            raise exceptions.AuthenticationFailed(msg)
+        elif len(auth) > 2:
+            msg = 'Invalid token header. Token string should not contain spaces.'
+            raise exceptions.AuthenticationFailed(msg)
+
+        try:
+            token = auth[1].decode()
+        except UnicodeError:
+            msg = 'Invalid token header. Token string should not contain invalid characters.'
+            raise exceptions.AuthenticationFailed(msg)
+
+        return self.authenticate_credentials(request,token)
+
+
+    def authenticate_credentials(self, request, key):
         model = self.get_model()
         try:
             token = model.objects.prefetch_related('user').get(key=key)
         except model.DoesNotExist:
             raise exceptions.AuthenticationFailed("Invalid token")
 
+        # Verify source IP is allowed
+        if len(token.allowed_ipranges) > 0:
+
+            if settings.PROXY_HEADER_REALIP in request.META:
+                clientip = request.META[settings.PROXY_HEADER_REALIP].split(",")[0].strip()
+            elif 'REMOTE_ADDR' in request.META:
+                clientip = request.META['REMOTE_ADDR']
+            else:
+                raise exceptions.AuthenticationFailed(f"The request headers ({settings.PROXY_HEADER_REALIP}, REMOTE_ADDR) are missing or do not contain a valid source IP.")
+
+            if not token.validateclientip(clientip):
+                raise exceptions.AuthenticationFailed(f"Source IP {clientip} is not allowed to use this token.")
+
+
         # Enforce the Token's expiration time, if one has been set.
         if token.is_expired:
             raise exceptions.AuthenticationFailed("Token expired")

netbox/netbox/settings.py

@@ -115,6 +115,7 @@ REMOTE_AUTH_STAFF_GROUPS = getattr(configuration, 'REMOTE_AUTH_STAFF_GROUPS', []
 REMOTE_AUTH_STAFF_USERS = getattr(configuration, 'REMOTE_AUTH_STAFF_USERS', [])
 REMOTE_AUTH_GROUP_SEPARATOR = getattr(configuration, 'REMOTE_AUTH_GROUP_SEPARATOR', '|')
 REPORTS_ROOT = getattr(configuration, 'REPORTS_ROOT', os.path.join(BASE_DIR, 'reports')).rstrip('/')
+PROXY_HEADER_REALIP = getattr(configuration, 'PROXY_HEADER_REALIP', 'HTTP_X_REAL_IP')
 RQ_DEFAULT_TIMEOUT = getattr(configuration, 'RQ_DEFAULT_TIMEOUT', 300)
 SCRIPTS_ROOT = getattr(configuration, 'SCRIPTS_ROOT', os.path.join(BASE_DIR, 'scripts')).rstrip('/')
 SESSION_FILE_PATH = getattr(configuration, 'SESSION_FILE_PATH', None)

netbox/templates/users/api_tokens.html

@@ -34,6 +34,14 @@
                                     <span>Never</span>
                                 {% endif %}
                             </div>
+                            <div class="col col-md-4">
+                                <small class="text-muted">Allowed IP Sources</small><br />
+                                {% if token.allowed_ipranges %}
+                                    {{ token.allowed_ipranges }}
+                                {% else %}
+                                    <span>Everywhere</span>
+                                {% endif %}
+                            </div>
                             <div class="col col-md-4">
                                 <small class="text-muted">Create/Edit/Delete Operations</small><br />
                                 {% if token.write_enabled %}

netbox/users/admin/__init__.py

@@ -58,7 +58,7 @@ class UserAdmin(UserAdmin_):
 class TokenAdmin(admin.ModelAdmin):
     form = forms.TokenAdminForm
     list_display = [
-        'key', 'user', 'created', 'expires', 'write_enabled', 'description'
+        'key', 'user', 'created', 'expires', 'write_enabled', 'description', 'allowed_ipranges'
     ]
 
 

netbox/users/admin/forms.py

@@ -51,7 +51,7 @@ class TokenAdminForm(forms.ModelForm):
 
     class Meta:
         fields = [
-            'user', 'key', 'write_enabled', 'expires', 'description'
+            'user', 'key', 'write_enabled', 'expires', 'description', 'allowed_ipranges'
         ]
         model = Token
 

netbox/users/api/serializers.py

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

netbox/users/forms.py

@@ -22,7 +22,7 @@ class TokenForm(BootstrapMixin, forms.ModelForm):
     class Meta:
         model = Token
         fields = [
-            'key', 'write_enabled', 'expires', 'description',
+            'key', 'write_enabled', 'expires', 'description', 'allowed_ipranges',
         ]
         widgets = {
             'expires': DateTimePicker(),

netbox/users/migrations/0002_token_allowed_ipranges.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.12 on 2022-03-11 13:39
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('users', '0001_squashed_0011'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='token',
+            name='allowed_ipranges',
+            field=models.CharField(blank=True, max_length=250),
+        ),
+    ]

netbox/users/models.py

@@ -4,6 +4,7 @@ import os
 from django.contrib.auth.models import Group, User
 from django.contrib.contenttypes.models import ContentType
 from django.contrib.postgres.fields import ArrayField
+from django.core.exceptions import ValidationError
 from django.core.validators import MinLengthValidator
 from django.db import models
 from django.db.models.signals import post_save
@@ -14,6 +15,7 @@ from netbox.models import BigIDModel
 from utilities.querysets import RestrictedQuerySet
 from utilities.utils import flatten_dict
 from .constants import *
+import ipaddress
 
 
 __all__ = (
@@ -203,6 +205,11 @@ class Token(BigIDModel):
         max_length=200,
         blank=True
     )
+    allowed_ipranges = models.CharField(
+        max_length=250,
+        blank=True,
+        help_text='Allowed ip addresses/ranges from where the token can be used (comma separated). Leave blank for no restrictions. Ex: "10.1.1.0/24, 192.168.10.16, 2001:DB8:1::/64"',
+    )
 
     class Meta:
         pass
@@ -214,6 +221,9 @@ class Token(BigIDModel):
     def save(self, *args, **kwargs):
         if not self.key:
             self.key = self.generate_key()
+        if not self.validateipranges(self.allowed_ipranges):
+            raise TypeError(f"{self.allowed_ipranges} contains an invalid ip address, range or prefix")
+
         return super().save(*args, **kwargs)
 
     @staticmethod
@@ -227,6 +237,59 @@ class Token(BigIDModel):
             return False
         return True
 
+    @staticmethod
+    def validateipranges(ip_addresses):
+        """
+        Checks that the value is a comma separated list of IPv4 and/or IPv6 addresses, ranges or subnets.
+        """
+        if len(ip_addresses)==0:
+            return True
+
+        for ip in ip_addresses.split(','):
+            try:
+                if '/' in ip:
+                    iptest=ipaddress.ip_network(ip)
+                elif '-' in ip:
+                    ips=ip.split('-')
+                    ip1=ipaddress.ip_address(ips[0])
+                    ip2=ipaddress.ip_address(ips[1])
+                    if ip1>ip2:
+                        raise ValidationError()
+                else:
+                    iptest=ipaddress.ip_address(ip)
+            except:
+                raise ValidationError(f"{ip} is an invalid value in the Allowed IP Ranges ({ip_addresses})")
+
+        return True
+
+    def validateclientip(self,raw_ip_address):
+        """
+        Checks that an ip address falls within the allowed ip ranges.
+        """
+        if len(self.allowed_ipranges)==0:
+            return True
+
+        try:
+            ip_address=ipaddress.ip_address(raw_ip_address)
+        except:
+            raise ValidationError(f"{raw_ip_address} is an invalid IP address")
+
+        for ip in self.allowed_ipranges.split(','):
+            if '/' in ip:
+                ipnet=ipaddress.ip_network(ip)
+                if ip_address in ipnet:
+                    return True
+            elif '-' in ip:
+                ips=ip.split('-')
+                ip1=ipaddress.ip_address(ips[0])
+                ip2=ipaddress.ip_address(ips[1])
+                if ip_address >= ip1 and ip_address <= ip2:
+                    return True
+            else:
+                ipaddr=ipaddress.ip_address(ip)
+                if ip_address==ipaddr:
+                    return True
+        return False
 
 #
 # Permissions