#8233 Restrict API key usage by source IP

docs/release-notes/version-3.1.md

@@ -2,6 +2,10 @@
 
 ## v3.1.10 (FUTURE)
 
+### Enhancements
+
+* [#8233](https://github.com/netbox-community/netbox/issues/8233) -  Restrict API key usage by source IP
+
 ### Bug Fixes
 
 * [#8820](https://github.com/netbox-community/netbox/issues/8820) - Fix navbar background color in dark mode

netbox/netbox/api/authentication.py

@@ -10,6 +10,11 @@ class TokenAuthentication(authentication.TokenAuthentication):
     A custom authentication scheme which enforces Token expiration times.
     """
     model = Token
+    __request = False
+
+    def authenticate(self, request):
+        self.request = request
+        return super().authenticate(request)
 
     def authenticate_credentials(self, key):
         model = self.get_model()
@@ -18,6 +23,20 @@ class TokenAuthentication(authentication.TokenAuthentication):
         except model.DoesNotExist:
             raise exceptions.AuthenticationFailed("Invalid token")
 
+        # Verify source IP is allowed
+        request = self.request
+        if token.allowed_ips and request:
+            # Replace 'HTTP_X_REAL_IP' with the settings variable choosen in #8867
+            if 'HTTP_X_REAL_IP' in request.META:
+                clientip = request.META['HTTP_X_REAL_IP'].split(",")[0].strip()
+            elif 'REMOTE_ADDR' in request.META:
+                clientip = request.META['REMOTE_ADDR']
+            else:
+                raise exceptions.AuthenticationFailed(f"The request HTTP headers (HTTP_X_REAL_IP, REMOTE_ADDR) are missing or do not contain a valid source IP.")
+
+            if not token.validate_client_ip(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

@@ -321,6 +321,7 @@ INSTALLED_APPS = [
     'wireless',
     'django_rq',  # Must come after extras to allow overriding management commands
     'drf_yasg',
+    'django_better_admin_arrayfield',
 ]
 
 # Middleware

netbox/templates/users/api_tokens.html

@@ -22,11 +22,11 @@
                     </div>
                     <div class="card-body">
                         <div class="row">
-                            <div class="col col-md-4">
+                            <div class="col col-md-3">
                                 <small class="text-muted">Created</small><br />
                                 {{ token.created|annotated_date }}
                             </div>
-                            <div class="col col-md-4">
+                            <div class="col col-md-3">
                                 <small class="text-muted">Expires</small><br />
                                 {% if token.expires %}
                                     {{ token.expires|annotated_date }}
@@ -34,7 +34,7 @@
                                     <span>Never</span>
                                 {% endif %}
                             </div>
-                            <div class="col col-md-4">
+                            <div class="col col-md-3">
                                 <small class="text-muted">Create/Edit/Delete Operations</small><br />
                                 {% if token.write_enabled %}
                                     <span class="badge bg-success">Enabled</span>
@@ -42,6 +42,14 @@
                                     <span class="badge bg-danger">Disabled</span>
                                 {% endif %}
                             </div>
+                            <div class="col col-md-3">
+                                <small class="text-muted">Allowed Source IPs</small><br />
+                                {% if token.allowed_ips %}
+                                    {{ token.allowed_ips }}
+                                {% else %}
+                                    <span>Any</span>
+                                {% endif %}
+                            </div>
                         </div>
                         {% if token.description %}
                             <br /><span>{{ token.description }}</span>

netbox/users/admin/__init__.py

@@ -1,6 +1,7 @@
 from django.contrib import admin
 from django.contrib.auth.admin import UserAdmin as UserAdmin_
 from django.contrib.auth.models import Group, User
+from django_better_admin_arrayfield.admin.mixins import DynamicArrayMixin
 
 from users.models import ObjectPermission, Token
 from . import filters, forms, inlines
@@ -55,10 +56,10 @@ class UserAdmin(UserAdmin_):
 #
 
 @admin.register(Token)
-class TokenAdmin(admin.ModelAdmin):
+class TokenAdmin(admin.ModelAdmin, DynamicArrayMixin):
     form = forms.TokenAdminForm
     list_display = [
-        'key', 'user', 'created', 'expires', 'write_enabled', 'description'
+        'key', 'user', 'created', 'expires', 'write_enabled', 'description', 'allowed_ips'
     ]
 
 

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_ips'
         ]
         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_ips')
 
     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_ips',
         ]
         widgets = {
             'expires': DateTimePicker(),

netbox/users/migrations/0002_token_allowed_ips.py

@@ -0,0 +1,20 @@
+# Generated by Django 3.2.12 on 2022-03-15 13:08
+
+from django.db import migrations
+import django_better_admin_arrayfield.models.fields
+import ipam.fields
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('users', '0001_squashed_0011'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='token',
+            name='allowed_ips',
+            field=django_better_admin_arrayfield.models.fields.ArrayField(base_field=ipam.fields.IPNetworkField(), blank=True, null=True, size=None),
+        ),
+    ]

netbox/users/models.py

@@ -4,17 +4,22 @@ 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
 from django.dispatch import receiver
 from django.utils import timezone
+from django_better_admin_arrayfield.models.fields import ArrayField as betterArrayField
 
 from netbox.models import BigIDModel
+from ipam.fields import IPNetworkField
 from utilities.querysets import RestrictedQuerySet
 from utilities.utils import flatten_dict
 from .constants import *
 
+import ipaddress
+
 
 __all__ = (
     'ObjectPermission',
@@ -203,6 +208,12 @@ class Token(BigIDModel):
         max_length=200,
         blank=True
     )
+    allowed_ips = betterArrayField(
+        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"',
+    )
 
     class Meta:
         pass
@@ -227,6 +238,24 @@ class Token(BigIDModel):
             return False
         return True
 
+    def validate_client_ip(self, raw_ip_address):
+        """
+        Checks that an ip address falls within the allowed ips.
+        """
+        if not self.allowed_ips:
+            return True
+
+        try:
+            ip_address = ipaddress.ip_address(raw_ip_address)
+        except ValueError:
+            raise ValidationError(f"{raw_ip_address} is an invalid IP address")
+
+        for ipnet in self.allowed_ips:
+            if ip_address in ipaddress.ip_network(ipnet):
+                return True
+
+        return False
+
 
 #
 # Permissions

requirements.txt

@@ -28,6 +28,7 @@ social-auth-core==4.2.0
 svgwrite==1.4.1
 tablib==3.2.0
 tzdata==2021.5
+django_better_admin_arrayfield==1.4.2
 
 # Workaround for #7401
 jsonschema==3.2.0