Merge pull request #9590 from netbox-community/8233-api-token-ip

Closes #8233: Restrict API tokens by source IP

Author: jeremystretch

docs/models/users/token.md

@@ -9,4 +9,4 @@ Each token contains a 160-bit key represented as 40 hexadecimal characters. When
 
 By default, a token can be used to perform all actions via the API that a user would be permitted to do via the web UI. Deselecting the "write enabled" option will restrict API requests made with the token to read operations (e.g. GET) only.
 
-Additionally, a token can be set to expire at a specific time. This can be useful if an external client needs to be granted temporary access to NetBox.
+Additionally, a token can be set to expire at a specific time. This can be useful if an external client needs to be granted temporary access to NetBox. Tokens can also be restricted by IP range: If defined, authentication for API clients connecting from an IP address outside these ranges will fail.

docs/release-notes/version-3.3.md

@@ -13,6 +13,8 @@
 
 #### PoE Interface Attributes ([#1099](https://github.com/netbox-community/netbox/issues/1099))
 
+#### Restrict API Tokens by Client IP ([#8233](https://github.com/netbox-community/netbox/issues/8233))
+
 ### Enhancements
 
 * [#1202](https://github.com/netbox-community/netbox/issues/1202) - Support overlapping assignment of NAT IP addresses

netbox/netbox/api/__init__.py

@@ -1,4 +1,4 @@
-from .fields import ChoiceField, ContentTypeField, SerializedPKRelatedField
+from .fields import *
 from .routers import NetBoxRouter
 from .serializers import BulkOperationSerializer, ValidatedModelSerializer, WritableNestedSerializer
 
@@ -7,6 +7,7 @@
     'BulkOperationSerializer',
     'ChoiceField',
     'ContentTypeField',
+    'IPNetworkSerializer',
     'NetBoxRouter',
     'SerializedPKRelatedField',
     'ValidatedModelSerializer',

netbox/netbox/api/authentication.py

@@ -3,14 +3,36 @@
 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):
+        result = super().authenticate(request)
+
+        if result:
+            token = result[1]
+
+            # Enforce source IP restrictions (if any) set on the token
+            if token.allowed_ips:
+                client_ip = get_client_ip(request)
+                if client_ip is None:
+                    raise exceptions.AuthenticationFailed(
+                        "Client IP address could not be determined for validation. Check that the HTTP server is "
+                        "correctly configured to pass the required header(s)."
+                    )
+                if not token.validate_client_ip(client_ip):
+                    raise exceptions.AuthenticationFailed(
+                        f"Source IP {client_ip} is not permitted to authenticate using this token."
+                    )
+
+        return result
+
     def authenticate_credentials(self, key):
         model = self.get_model()
         try:

netbox/netbox/api/fields.py

@@ -1,12 +1,18 @@
 from collections import OrderedDict
 
-import pytz
-from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ObjectDoesNotExist
+from netaddr import IPNetwork
 from rest_framework import serializers
 from rest_framework.exceptions import ValidationError
 from rest_framework.relations import PrimaryKeyRelatedField, RelatedField
 
+__all__ = (
+    'ChoiceField',
+    'ContentTypeField',
+    'IPNetworkSerializer',
+    'SerializedPKRelatedField',
+)
+
 
 class ChoiceField(serializers.Field):
     """
@@ -104,6 +110,17 @@ def to_representation(self, obj):
         return f"{obj.app_label}.{obj.model}"
 
 
+class IPNetworkSerializer(serializers.Serializer):
+    """
+    Representation of an IP network value (e.g. 192.0.2.0/24).
+    """
+    def to_representation(self, instance):
+        return str(instance)
+
+    def to_internal_value(self, value):
+        return IPNetwork(value)
+
+
 class SerializedPKRelatedField(PrimaryKeyRelatedField):
     """
     Extends PrimaryKeyRelatedField to return a serialized object on read. This is useful for representing related

netbox/netbox/tests/test_authentication.py

@@ -1,3 +1,5 @@
+import datetime
+
 from django.conf import settings
 from django.contrib.auth.models import Group, User
 from django.contrib.contenttypes.models import ContentType
@@ -8,10 +10,73 @@
 from rest_framework.test import APIClient
 
 from dcim.models import Site
-from ipam.choices import PrefixStatusChoices
 from ipam.models import Prefix
 from users.models import ObjectPermission, Token
 from utilities.testing import TestCase
+from utilities.testing.api import APITestCase
+
+
+class TokenAuthenticationTestCase(APITestCase):
+
+    @override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
+    def test_token_authentication(self):
+        url = reverse('dcim-api:site-list')
+
+        # Request without a token should return a 403
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 403)
+
+        # Valid token should return a 200
+        token = Token.objects.create(user=self.user)
+        response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token.key}')
+        self.assertEqual(response.status_code, 200)
+
+    @override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
+    def test_token_expiration(self):
+        url = reverse('dcim-api:site-list')
+
+        # Request without a non-expired token should succeed
+        token = Token.objects.create(user=self.user)
+        response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token.key}')
+        self.assertEqual(response.status_code, 200)
+
+        # Request with an expired token should fail
+        token.expires = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
+        token.save()
+        response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token.key}')
+        self.assertEqual(response.status_code, 403)
+
+    @override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
+    def test_token_write_enabled(self):
+        url = reverse('dcim-api:site-list')
+        data = {
+            'name': 'Site 1',
+            'slug': 'site-1',
+        }
+
+        # Request with a write-disabled token should fail
+        token = Token.objects.create(user=self.user, write_enabled=False)
+        response = self.client.post(url, data, format='json', HTTP_AUTHORIZATION=f'Token {token.key}')
+        self.assertEqual(response.status_code, 403)
+
+        # Request with a write-enabled token should succeed
+        token.write_enabled = True
+        token.save()
+        response = self.client.post(url, data, format='json', HTTP_AUTHORIZATION=f'Token {token.key}')
+        self.assertEqual(response.status_code, 403)
+
+    @override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
+    def test_token_allowed_ips(self):
+        url = reverse('dcim-api:site-list')
+
+        # Request from a non-allowed client IP should fail
+        token = Token.objects.create(user=self.user, allowed_ips=['192.0.2.0/24'])
+        response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token.key}', REMOTE_ADDR='127.0.0.1')
+        self.assertEqual(response.status_code, 403)
+
+        # Request with an expired token should fail
+        response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token.key}', REMOTE_ADDR='192.0.2.1')
+        self.assertEqual(response.status_code, 200)
 
 
 class ExternalAuthenticationTestCase(TestCase):

netbox/templates/users/api_tokens.html

@@ -22,27 +22,34 @@
                     </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 }}
                                 {% else %}
                                     <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>
                                 {% else %}
                                     <span class="badge bg-danger">Disabled</span>
                                 {% endif %}
                             </div>
-                        </div>
+                            <div class="col col-md-3">
+                                <small class="text-muted">Allowed Source IPs</small><br />
+                                {% if token.allowed_ips %}
+                                    {{ token.allowed_ips|join:', ' }}
+                                {% else %}
+                                    <span>Any</span>
+                                {% endif %}
+                            </div>                        </div>
                         {% if token.description %}
                             <br /><span>{{ token.description }}</span>
                         {% endif %}

netbox/users/admin/__init__.py

@@ -58,9 +58,13 @@ def get_inlines(self, request, obj):
 class TokenAdmin(admin.ModelAdmin):
     form = forms.TokenAdminForm
     list_display = [
-        'key', 'user', 'created', 'expires', 'write_enabled', 'description'
+        'key', 'user', 'created', 'expires', 'write_enabled', 'description', 'list_allowed_ips'
     ]
 
+    def list_allowed_ips(self, obj):
+        return obj.allowed_ips or 'Any'
+    list_allowed_ips.short_description = "Allowed IPs"
+
 
 #
 # Permissions

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

@@ -2,7 +2,7 @@
 from django.contrib.contenttypes.models import ContentType
 from rest_framework import serializers
 
-from netbox.api import ContentTypeField, SerializedPKRelatedField, ValidatedModelSerializer
+from netbox.api import ContentTypeField, IPNetworkSerializer, SerializedPKRelatedField, ValidatedModelSerializer
 from users.models import ObjectPermission, Token
 from .nested_serializers import *
 
@@ -64,10 +64,19 @@ class TokenSerializer(ValidatedModelSerializer):
     url = serializers.HyperlinkedIdentityField(view_name='users-api:token-detail')
     key = serializers.CharField(min_length=40, max_length=40, allow_blank=True, required=False)
     user = NestedUserSerializer()
+    allowed_ips = serializers.ListField(
+        child=IPNetworkSerializer(),
+        required=False,
+        allow_empty=True,
+        default=[]
+    )
 
     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: