14884 log error on form validation fail

Closes #20492: Disable API token plaintext retrieval

.github/ISSUE_TEMPLATE/02-bug_report.yaml

@@ -35,9 +35,9 @@ body:
       label: Python Version
       description: What version of Python are you currently running?
       options:
-        - "3.10"
-        - "3.11"
         - "3.12"
+        - "3.13"
+        - "3.14"
     validations:
       required: true
   - type: textarea

.github/workflows/ci.yml

@@ -31,7 +31,7 @@ jobs:
       NETBOX_CONFIGURATION: netbox.configuration_testing
     strategy:
       matrix:
-        python-version: ['3.10', '3.11', '3.12']
+        python-version: ['3.12', '3.13']
         node-version: ['20.x']
     services:
       redis:

docs/administration/authentication/overview.md

@@ -2,7 +2,7 @@
 
 ## Local Authentication
 
-Local user accounts and groups can be created in NetBox under the "Authentication" section in the "Admin" menu. This section is available only to users with the "staff" permission enabled.
+Local user accounts and groups can be created in NetBox under the "Authentication" section in the "Admin" menu.
 
 At a minimum, each user account must have a username and password set. User accounts may also denote a first name, last name, and email address. [Permissions](../permissions.md) may also be assigned to individual users and/or groups as needed.
 

docs/configuration/remote-authentication.md

@@ -127,19 +127,3 @@ The list of groups that promote an remote User to Superuser on Login. If group i
 Default: `[]` (Empty list)
 
 The list of users that get promoted to Superuser on Login. If user isn't present in list on next Login, the Role gets revoked. (Requires `REMOTE_AUTH_ENABLED` and `REMOTE_AUTH_GROUP_SYNC_ENABLED` )
-
----
-
-## REMOTE_AUTH_STAFF_GROUPS
-
-Default: `[]` (Empty list)
-
-The list of groups that promote an remote User to Staff on Login. If group isn't present on next Login, the Role gets revoked. (Requires `REMOTE_AUTH_ENABLED` and `REMOTE_AUTH_GROUP_SYNC_ENABLED` )
-
----
-
-## REMOTE_AUTH_STAFF_USERS
-
-Default: `[]` (Empty list)
-
-The list of users that get promoted to Staff on Login. If user isn't present in list on next Login, the Role gets revoked. (Requires `REMOTE_AUTH_ENABLED` and `REMOTE_AUTH_GROUP_SYNC_ENABLED` )

docs/configuration/required-parameters.md

@@ -23,6 +23,31 @@ ALLOWED_HOSTS = ['*']
 
 ---
 
+## API_TOKEN_PEPPERS
+
+!!! info "This parameter was introduced in NetBox v4.5."
+
+[Cryptographic peppers](https://en.wikipedia.org/wiki/Pepper_(cryptography)) are employed to generate hashes of sensitive values on the server. This parameter defines the peppers used to hash v2 API tokens in NetBox. You must define at least one pepper before creating a v2 API token. See the [API documentation](../integrations/rest-api.md#authentication) for further information about how peppers are used.
+
+```python
+API_TOKEN_PEPPERS = {
+    # DO NOT USE THIS EXAMPLE PEPPER IN PRODUCTION
+    1: 'kp7ht*76fiQAhUi5dHfASLlYUE_S^gI^(7J^K5M!LfoH@vl&b_',
+}
+```
+
+!!! warning "Peppers are sensitive"
+    Treat pepper values as extremely sensitive. Consider populating peppers from environment variables at initialization time rather than defining them in the configuration file, if feasible. 
+
+Peppers must be at least 50 characters in length and should comprise a random string with a diverse character set. Consider using the Python script at `$INSTALL_ROOT/netbox/generate_secret_key.py` to generate a pepper value.
+
+It is recommended to start with a pepper ID of `1`. Additional peppers can be introduced later as needed to begin rotating token hashes.
+
+!!! tip
+    Although NetBox will run without `API_TOKEN_PEPPERS` defined, the use of v2 API tokens will be unavailable.
+
+---
+
 ## DATABASE
 
 !!! warning "Legacy Configuration Parameter"

docs/configuration/security.md

@@ -1,16 +1,5 @@
 # Security & Authentication Parameters
 
-## ALLOW_TOKEN_RETRIEVAL
-
-Default: `False`
-
-!!! note
-    The default value of this parameter changed from `True` to `False` in NetBox v4.3.0.
-
-If disabled, the values of API tokens will not be displayed after each token's initial creation. A user **must** record the value of a token prior to its creation, or it will be lost. Note that this affects _all_ users, regardless of assigned permissions.
-
----
-
 ## ALLOWED_URL_SCHEMES
 
 !!! tip "Dynamic Configuration Parameter"

docs/customization/custom-scripts.md

@@ -131,17 +131,6 @@ self.log_info(f"Running as user {username} (IP: {ip_address})...")
 
 For a complete list of available request parameters, please see the [Django documentation](https://docs.djangoproject.com/en/stable/ref/request-response/).
 
-## Reading Data from Files
-
-The Script class provides two convenience methods for reading data from files:
-
-* `load_yaml`
-* `load_json`
-
-These two methods will load data in YAML or JSON format, respectively, from files within the local path (i.e. `SCRIPTS_ROOT`).
-
-**Note:** These convenience methods are deprecated and will be removed in NetBox v4.4.  These only work if running scripts within the local path, they will not work if using a storage other than ScriptFileSystemStorage.
-
 ## Logging
 
 The Script object provides a set of convenient functions for recording messages at different severity levels:

docs/development/getting-started.md

@@ -7,7 +7,7 @@ Getting started with NetBox development is pretty straightforward, and should fe
 * A Linux system or compatible environment
 * A PostgreSQL server, which can be installed locally [per the documentation](../installation/1-postgresql.md)
 * A Redis server, which can also be [installed locally](../installation/2-redis.md)
-* Python 3.10 or later
+* Python 3.12 or later
 
 ### 1. Fork the Repo
 

docs/features/api-integration.md

@@ -8,7 +8,7 @@ NetBox's REST API, powered by the [Django REST Framework](https://www.django-res
 
 ```no-highlight
 curl -s -X POST \
--H "Authorization: Token $TOKEN" \
+-H "Authorization: Bearer $TOKEN" \
 -H "Content-Type: application/json" \
 http://netbox/api/ipam/prefixes/ \
 --data '{"prefix": "192.0.2.0/24", "site": {"name": "Branch 12"}}'

docs/features/user-preferences.md

@@ -34,9 +34,6 @@ Sets the default number of rows displayed on paginated tables.
 ### Paginator placement
 Controls where pagination controls are rendered relative to a table.
 
-### HTMX navigation (experimental)
-Enables partial‑page navigation for supported views. Disable this preference if unexpected behavior is observed.
-
 ### Striped table rows
 Toggles alternating row backgrounds on tables.
 

docs/installation/3-netbox.md

@@ -6,8 +6,8 @@ This section of the documentation discusses installing and configuring the NetBo
 
 Begin by installing all system packages required by NetBox and its dependencies.
 
-!!! warning "Python 3.10 or later required"
-    NetBox supports Python 3.10, 3.11, and 3.12.
+!!! warning "Python 3.12 or later required"
+    NetBox supports only Python 3.12 or later.
 
 ```no-highlight
 sudo apt install -y python3 python3-pip python3-venv python3-dev \
@@ -15,7 +15,7 @@ build-essential libxml2-dev libxslt1-dev libffi-dev libpq-dev \
 libssl-dev zlib1g-dev
 ```
 
-Before continuing, check that your installed Python version is at least 3.10:
+Before continuing, check that your installed Python version is at least 3.12:
 
 ```no-highlight
 python3 -V
@@ -120,6 +120,23 @@ If you are not yet sure what the domain name and/or IP address of the NetBox ins
 ALLOWED_HOSTS = ['*']
 ```
 
+### API_TOKEN_PEPPERS
+
+Define at least one random cryptographic pepper, identified by a numeric ID starting at 1. This will be used to generate SHA256 checksums for API tokens.
+
+```python
+API_TOKEN_PEPPERS = {
+    # DO NOT USE THIS EXAMPLE PEPPER IN PRODUCTION
+    1: 'kp7ht*76fiQAhUi5dHfASLlYUE_S^gI^(7J^K5M!LfoH@vl&b_',
+}
+```
+
+!!! tip
+    As with [`SECRET_KEY`](#secret_key) below, you can use the `generate_secret_key.py` script to generate a random pepper:
+    ```no-highlight
+    python3 ../generate_secret_key.py
+    ```
+
 ### DATABASES
 
 This parameter holds the PostgreSQL database configuration details. The default database must be defined; additional databases may be defined as needed e.g. by plugins.
@@ -235,10 +252,10 @@ Once NetBox has been configured, we're ready to proceed with the actual installa
 sudo /opt/netbox/upgrade.sh
 ```
 
-Note that **Python 3.10 or later is required** for NetBox v4.0 and later releases. If the default Python installation on your server is set to a lesser version,  pass the path to the supported installation as an environment variable named `PYTHON`. (Note that the environment variable must be passed _after_ the `sudo` command.)
+Note that **Python 3.12 or later is required** for NetBox v4.5 and later releases. If the default Python installation on your server is set to a lesser version,  pass the path to the supported installation as an environment variable named `PYTHON`. (Note that the environment variable must be passed _after_ the `sudo` command.)
 
 ```no-highlight
-sudo PYTHON=/usr/bin/python3.10 /opt/netbox/upgrade.sh
+sudo PYTHON=/usr/bin/python3.12 /opt/netbox/upgrade.sh
 ```
 
 !!! note

docs/installation/4a-gunicorn.md

@@ -60,6 +60,3 @@ You should see output similar to the following:
     If the NetBox service fails to start, issue the command `journalctl -eu netbox` to check for log messages that may indicate the problem.
 
 Once you've verified that the WSGI workers are up and running, move on to HTTP server setup.
-
-!!! note
-    There is a bug in the current stable release of gunicorn (v21.2.0) where automatic restarts of the worker processes can result in 502 errors under heavy load. (See [gunicorn bug #3038](https://github.com/benoitc/gunicorn/issues/3038) for more detail.) Users who encounter this issue may opt to downgrade to an earlier, unaffected release of gunicorn (`pip install gunicorn==20.1.0`). Note, however, that this earlier release does not officially support Python 3.11.

docs/installation/6-ldap.md

@@ -121,7 +121,6 @@ AUTH_LDAP_MIRROR_GROUPS = True
 # Define special user types using groups. Exercise great caution when assigning superuser status.
 AUTH_LDAP_USER_FLAGS_BY_GROUP = {
     "is_active": "cn=active,ou=groups,dc=example,dc=com",
-    "is_staff": "cn=staff,ou=groups,dc=example,dc=com",
     "is_superuser": "cn=superuser,ou=groups,dc=example,dc=com"
 }
 
@@ -134,7 +133,6 @@ AUTH_LDAP_CACHE_TIMEOUT = 3600
 ```
 
 * `is_active` - All users must be mapped to at least this group to enable authentication. Without this, users cannot log in.
-* `is_staff` - Users mapped to this group are enabled for access to the administration tools; this is the equivalent of checking the "staff status" box on a manually created user. This doesn't grant any specific permissions.
 * `is_superuser` - Users mapped to this group will be granted superuser status. Superusers are implicitly granted all permissions.
 
 !!! warning
@@ -248,7 +246,6 @@ AUTH_LDAP_MIRROR_GROUPS = True
 # Define special user types using groups. Exercise great caution when assigning superuser status.
 AUTH_LDAP_USER_FLAGS_BY_GROUP = {
     "is_active": "cn=active,ou=groups,dc=example,dc=com",
-    "is_staff": "cn=staff,ou=groups,dc=example,dc=com",
     "is_superuser": "cn=superuser,ou=groups,dc=example,dc=com"
 }
 

docs/installation/index.md

@@ -27,7 +27,7 @@ The following sections detail how to set up a new instance of NetBox:
 
 | Dependency | Supported Versions |
 |------------|--------------------|
-| Python     | 3.10, 3.11, 3.12   |
+| Python     | 3.12, 3.13, 3.14   |
 | PostgreSQL | 14+                |
 | Redis      | 4.0+               |
 

docs/installation/upgrading.md

@@ -19,7 +19,7 @@ NetBox requires the following dependencies:
 
 | Dependency | Supported Versions |
 |------------|--------------------|
-| Python     | 3.10, 3.11, 3.12   |
+| Python     | 3.12, 3.13, 3.14   |
 | PostgreSQL | 14+                |
 | Redis      | 4.0+               |
 
@@ -27,6 +27,7 @@ NetBox requires the following dependencies:
 
 | NetBox Version | Python min | Python max | PostgreSQL min | Redis min |                                       Documentation                                       |
 |:--------------:|:----------:|:----------:|:--------------:|:---------:|:-----------------------------------------------------------------------------------------:|
+|      4.5       |    3.12    |    3.14    |       14       |    4.0    | [Link](https://github.com/netbox-community/netbox/blob/v4.5.0/docs/installation/index.md) |
 |      4.4       |    3.10    |    3.12    |       14       |    4.0    | [Link](https://github.com/netbox-community/netbox/blob/v4.4.0/docs/installation/index.md) |
 |      4.3       |    3.10    |    3.12    |       14       |    4.0    | [Link](https://github.com/netbox-community/netbox/blob/v4.3.0/docs/installation/index.md) |
 |      4.2       |    3.10    |    3.12    |       13       |    4.0    | [Link](https://github.com/netbox-community/netbox/blob/v4.2.0/docs/installation/index.md) |
@@ -130,7 +131,7 @@ sudo ./upgrade.sh
     If the default version of Python is not at least 3.10, you'll need to pass the path to a supported Python version as an environment variable when calling the upgrade script. For example:
 
     ```no-highlight
-    sudo PYTHON=/usr/bin/python3.10 ./upgrade.sh
+    sudo PYTHON=/usr/bin/python3.12 ./upgrade.sh
     ```
 
 !!! note

docs/integrations/rest-api.md

@@ -80,7 +80,7 @@ Likewise, the site, rack, and device objects are located under the "DCIM" applic
 
 The full hierarchy of available endpoints can be viewed by navigating to the API root in a web browser.
 
-Each model generally has two views associated with it: a list view and a detail view. The list view is used to retrieve a list of multiple objects and to create new objects. The detail view is used to retrieve, update, or delete an single existing object. All objects are referenced by their numeric primary key (`id`).
+Each model generally has two views associated with it: a list view and a detail view. The list view is used to retrieve a list of multiple objects and to create new objects. The detail view is used to retrieve, update, or delete a single existing object. All objects are referenced by their numeric primary key (`id`).
 
 * `/api/dcim/devices/` - List existing devices or create a new device
 * `/api/dcim/devices/123/` - Retrieve, update, or delete the device with ID 123
@@ -653,18 +653,22 @@ The NetBox REST API primarily employs token-based authentication. For convenienc
 
 ### Tokens
 
-A token is a unique identifier mapped to a NetBox user account. Each user may have one or more tokens which he or she can use for authentication when making REST API requests. To create a token, navigate to the API tokens page under your user profile.
+A token is a secret, unique identifier mapped to a NetBox user account. Each user may have one or more tokens which he or she can use for authentication when making REST API requests. To create a token, navigate to the API tokens page under your user profile. When creating a token, NetBox will automatically populate a randomly-generated token value.
+
+!!! note "Tokens cannot be retrieved once created"
+    Once a token has been created, its plaintext value cannot be retrieved. For this reason, you must take care to securely record the token locally immediately upon its creation. If a token plaintext is lost, it cannot be recovered: A new token must be created.
 
 By default, all users can create and manage their own REST API tokens under the user control panel in the UI or via the REST API. This ability can be disabled by overriding the [`DEFAULT_PERMISSIONS`](../configuration/security.md#default_permissions) configuration parameter.
 
-Each token contains a 160-bit key represented as 40 hexadecimal characters. When creating a token, you'll typically leave the key field blank so that a random key will be automatically generated. However, NetBox allows you to specify a key in case you need to restore a previously deleted token to operation.
-
 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.
 
-!!! info "Restricting Token Retrieval"
-    The ability to retrieve the key value of a previously-created API token can be restricted by disabling the [`ALLOW_TOKEN_RETRIEVAL`](../configuration/security.md#allow_token_retrieval) configuration parameter.
+#### v1 and v2 Tokens
 
-### Restricting Write Operations
+Beginning with NetBox v4.5, two versions of API token are supported, denoted as v1 and v2. Users are strongly encouraged to create only v2 tokens and to discontinue the use of v1 tokens. Support for v1 tokens will be removed in a future NetBox release.
+
+v2 API tokens offer much stronger security. The token plaintext given at creation time is hashed together with a configured [cryptographic pepper](../configuration/required-parameters.md#api_token_peppers) to generate a unique checksum. This checksum is irreversible; the token plaintext is never stored on the server and thus cannot be retrieved even with database-level access.
+
+#### Restricting Write Operations
 
 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.
 
@@ -681,10 +685,22 @@ It is possible to provision authentication tokens for other users via the REST A
 
 ### Authenticating to the API
 
-An authentication token is attached to a request by setting the `Authorization` header to the string `Token` followed by a space and the user's token:
+An authentication token is included with a request in its `Authorization` header. The format of the header value depends on the version of token in use. v2 tokens use the following form, concatenating the token's prefix (`nbt_`) and key with its plaintext value, separated by a period:
 
 ```
-$ curl -H "Authorization: Token $TOKEN" \
+Authorization: Bearer nbt_<key>.<token>
+```
+
+Legacy v1 tokens use the prefix `Token` rather than `Bearer`, and include only the token plaintext. (v1 tokens do not have a key.)
+
+```
+Authorization: Token <token>
+```
+
+Below is an example REST API request utilizing a v2 token.
+
+```
+$ curl -H "Authorization: Bearer nbt_4F9DAouzURLb.zjebxBPzICiPbWz0Wtx0fTL7bCKXKGTYhNzkgC2S" \
 -H "Accept: application/json; indent=4" \
 https://netbox/api/dcim/sites/
 {

docs/plugins/development/index.md

@@ -173,12 +173,12 @@ classifiers=[
     'Intended Audience :: Developers',
     'Natural Language :: English',
     "Programming Language :: Python :: 3 :: Only",
-    'Programming Language :: Python :: 3.10',
-    'Programming Language :: Python :: 3.11',
     'Programming Language :: Python :: 3.12',
+    'Programming Language :: Python :: 3.13',
+    'Programming Language :: Python :: 3.14',
 ]
 
-requires-python = ">=3.10.0"
+requires-python = ">=3.12.0"
 
 ```
 
@@ -195,7 +195,7 @@ python3 -m venv ~/.virtualenvs/my_plugin
 You can make NetBox available within this environment by creating a path file pointing to its location. This will add NetBox to the Python path upon activation. (Be sure to adjust the command below to specify your actual virtual environment path, Python version, and NetBox installation.)
 
 ```shell
-echo /opt/netbox/netbox > $VENV/lib/python3.10/site-packages/netbox.pth
+echo /opt/netbox/netbox > $VENV/lib/python3.12/site-packages/netbox.pth
 ```
 
 ## Development Installation

docs/plugins/development/navigation.md

@@ -64,14 +64,17 @@ item1 = PluginMenuItem(
 
 A `PluginMenuItem` has the following attributes:
 
-| Attribute       | Required | Description                                                                                              |
-|-----------------|----------|----------------------------------------------------------------------------------------------------------|
-| `link`          | Yes      | Name of the URL path to which this menu item links                                                       |
-| `link_text`     | Yes      | The text presented to the user                                                                           |
-| `permissions`   | -        | A list of permissions required to display this link                                                      |
-| `auth_required` | -        | Display only for authenticated users                                                                     |
-| `staff_only`    | -        | Display only for users who have `is_staff` set to true (any specified permissions will also be required) |
-| `buttons`       | -        | An iterable of PluginMenuButton instances to include                                                     |
+| Attribute       | Required | Description                                          |
+|-----------------|----------|------------------------------------------------------|
+| `link`          | Yes      | Name of the URL path to which this menu item links   |
+| `link_text`     | Yes      | The text presented to the user                       |
+| `permissions`   | -        | A list of permissions required to display this link  |
+| `auth_required` | -        | Display only for authenticated users                 |
+| `staff_only`    | -        | Display only for superusers                          |
+| `buttons`       | -        | An iterable of PluginMenuButton instances to include |
+
+!!! note "Changed in NetBox v4.5"
+    In releases prior to NetBox v4.5, `staff_only` restricted display of a menu item to only users with `is_staff` set to True. In NetBox v4.5, the `is_staff` flag was removed from the user model. Menu items with `staff_only` set to True are now displayed only for superusers.
 
 ## Menu Buttons
 

netbox/account/tables.py

@@ -1,57 +0,0 @@
-from django.utils.translation import gettext as _
-
-from account.models import UserToken
-from netbox.tables import NetBoxTable, columns
-
-__all__ = (
-    'UserTokenTable',
-)
-
-
-TOKEN = """<samp><span id="token_{{ record.pk }}">{{ record }}</span></samp>"""
-
-ALLOWED_IPS = """{{ value|join:", " }}"""
-
-COPY_BUTTON = """
-{% if settings.ALLOW_TOKEN_RETRIEVAL %}
-  {% copy_content record.pk prefix="token_" color="success" %}
-{% endif %}
-"""
-
-
-class UserTokenTable(NetBoxTable):
-    """
-    Table for users to manager their own API tokens under account views.
-    """
-    key = columns.TemplateColumn(
-        verbose_name=_('Key'),
-        template_code=TOKEN,
-    )
-    write_enabled = columns.BooleanColumn(
-        verbose_name=_('Write Enabled')
-    )
-    created = columns.DateTimeColumn(
-        timespec='minutes',
-        verbose_name=_('Created'),
-    )
-    expires = columns.DateTimeColumn(
-        timespec='minutes',
-        verbose_name=_('Expires'),
-    )
-    last_used = columns.DateTimeColumn(
-        verbose_name=_('Last Used'),
-    )
-    allowed_ips = columns.TemplateColumn(
-        verbose_name=_('Allowed IPs'),
-        template_code=ALLOWED_IPS
-    )
-    actions = columns.ActionsColumn(
-        actions=('edit', 'delete'),
-        extra_buttons=COPY_BUTTON
-    )
-
-    class Meta(NetBoxTable.Meta):
-        model = UserToken
-        fields = (
-            'pk', 'id', 'key', 'description', 'write_enabled', 'created', 'expires', 'last_used', 'allowed_ips',
-        )

netbox/account/views.py

@@ -26,8 +26,9 @@ from extras.tables import BookmarkTable, NotificationTable, SubscriptionTable
 from netbox.authentication import get_auth_backend_display, get_saml_idps
 from netbox.config import get_config
 from netbox.views import generic
-from users import forms, tables
+from users import forms
 from users.models import UserConfig
+from users.tables import TokenTable
 from utilities.request import safe_for_redirect
 from utilities.string import remove_linebreaks
 from utilities.views import register_model_view
@@ -328,7 +329,8 @@ class UserTokenListView(LoginRequiredMixin, View):
 
     def get(self, request):
         tokens = UserToken.objects.filter(user=request.user)
-        table = tables.UserTokenTable(tokens)
+        table = TokenTable(tokens)
+        table.columns.hide('user')
         table.configure(request)
 
         return render(request, 'account/token_list.html', {
@@ -343,11 +345,9 @@ class UserTokenView(LoginRequiredMixin, View):
 
     def get(self, request, pk):
         token = get_object_or_404(UserToken.objects.filter(user=request.user), pk=pk)
-        key = token.key if settings.ALLOW_TOKEN_RETRIEVAL else None
 
         return render(request, 'account/token.html', {
             'object': token,
-            'key': key,
         })
 
 

netbox/core/api/views.py

@@ -9,7 +9,6 @@ from drf_spectacular.utils import OpenApiParameter, extend_schema
 from rest_framework import viewsets
 from rest_framework.decorators import action
 from rest_framework.exceptions import PermissionDenied
-from rest_framework.permissions import IsAdminUser
 from rest_framework.response import Response
 from rest_framework.routers import APIRootView
 from rest_framework.viewsets import ReadOnlyModelViewSet
@@ -24,7 +23,7 @@ from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
 from netbox.api.metadata import ContentTypeMetadata
 from netbox.api.pagination import LimitOffsetListPagination
 from netbox.api.viewsets import NetBoxModelViewSet, NetBoxReadOnlyModelViewSet
-
+from utilities.api import IsSuperuser
 from . import serializers
 
 
@@ -100,7 +99,7 @@ class BaseRQViewSet(viewsets.ViewSet):
     """
     Base class for RQ view sets. Provides a list() method. Subclasses must implement get_data().
     """
-    permission_classes = [IsAdminUser]
+    permission_classes = [IsSuperuser]
     serializer_class = None
 
     def get_data(self):

netbox/core/models/contenttypes.py

@@ -1,3 +0,0 @@
-# TODO: Remove this module in NetBox v4.5
-# Provided for backward compatibility
-from .object_types import *

netbox/core/tests/test_api.py

@@ -8,6 +8,7 @@ from rq.job import Job as RQ_Job, JobStatus
 from rq.registry import FailedJobRegistry, StartedJobRegistry
 
 from rest_framework import status
+from users.constants import TOKEN_PREFIX
 from users.models import Token, User
 from utilities.testing import APITestCase, APIViewTestCases, TestCase
 from utilities.testing.utils import disable_logging
@@ -107,14 +108,14 @@ class ObjectTypeTest(APITestCase):
     def test_list_objects(self):
         object_type_count = ObjectType.objects.count()
 
-        response = self.client.get(reverse('extras-api:objecttype-list'), **self.header)
+        response = self.client.get(reverse('core-api:objecttype-list'), **self.header)
         self.assertHttpStatus(response, status.HTTP_200_OK)
         self.assertEqual(response.data['count'], object_type_count)
 
     def test_get_object(self):
         object_type = ObjectType.objects.first()
 
-        url = reverse('extras-api:objecttype-detail', kwargs={'pk': object_type.pk})
+        url = reverse('core-api:objecttype-detail', kwargs={'pk': object_type.pk})
         self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_200_OK)
 
 
@@ -134,12 +135,9 @@ class BackgroundTaskTestCase(TestCase):
         Create a user and token for API calls.
         """
         # Create the test user and assign permissions
-        self.user = User.objects.create_user(username='testuser')
-        self.user.is_staff = True
-        self.user.is_active = True
-        self.user.save()
+        self.user = User.objects.create_user(username='testuser', is_active=True)
         self.token = Token.objects.create(user=self.user)
-        self.header = {'HTTP_AUTHORIZATION': f'Token {self.token.key}'}
+        self.header = {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{self.token.key}.{self.token.token}'}
 
         # Clear all queues prior to running each test
         get_queue('default').connection.flushall()
@@ -150,13 +148,11 @@ class BackgroundTaskTestCase(TestCase):
         url = reverse('core-api:rqqueue-list')
 
         # Attempt to load view without permission
-        self.user.is_staff = False
-        self.user.save()
         response = self.client.get(url, **self.header)
         self.assertEqual(response.status_code, 403)
 
         # Load view with permission
-        self.user.is_staff = True
+        self.user.is_superuser = True
         self.user.save()
         response = self.client.get(url, **self.header)
         self.assertEqual(response.status_code, 200)
@@ -165,7 +161,16 @@ class BackgroundTaskTestCase(TestCase):
         self.assertIn('low', str(response.content))
 
     def test_background_queue(self):
-        response = self.client.get(reverse('core-api:rqqueue-detail', args=['default']), **self.header)
+        url = reverse('core-api:rqqueue-detail', args=['default'])
+
+        # Attempt to load view without permission
+        response = self.client.get(url, **self.header)
+        self.assertEqual(response.status_code, 403)
+
+        # Load view with permission
+        self.user.is_superuser = True
+        self.user.save()
+        response = self.client.get(url, **self.header)
         self.assertEqual(response.status_code, 200)
         self.assertIn('default', str(response.content))
         self.assertIn('oldest_job_timestamp', str(response.content))
@@ -174,8 +179,16 @@ class BackgroundTaskTestCase(TestCase):
     def test_background_task_list(self):
         queue = get_queue('default')
         queue.enqueue(self.dummy_job_default)
+        url = reverse('core-api:rqtask-list')
 
-        response = self.client.get(reverse('core-api:rqtask-list'), **self.header)
+        # Attempt to load view without permission
+        response = self.client.get(url, **self.header)
+        self.assertEqual(response.status_code, 403)
+
+        # Load view with permission
+        self.user.is_superuser = True
+        self.user.save()
+        response = self.client.get(url, **self.header)
         self.assertEqual(response.status_code, 200)
         self.assertIn('origin', str(response.content))
         self.assertIn('core.tests.test_api.BackgroundTaskTestCase.dummy_job_default()', str(response.content))
@@ -183,8 +196,16 @@ class BackgroundTaskTestCase(TestCase):
     def test_background_task(self):
         queue = get_queue('default')
         job = queue.enqueue(self.dummy_job_default)
+        url = reverse('core-api:rqtask-detail', args=[job.id])
 
-        response = self.client.get(reverse('core-api:rqtask-detail', args=[job.id]), **self.header)
+        # Attempt to load view without permission
+        response = self.client.get(url, **self.header)
+        self.assertEqual(response.status_code, 403)
+
+        # Load view with permission
+        self.user.is_superuser = True
+        self.user.save()
+        response = self.client.get(url, **self.header)
         self.assertEqual(response.status_code, 200)
         self.assertIn(str(job.id), str(response.content))
         self.assertIn('origin', str(response.content))
@@ -194,45 +215,65 @@ class BackgroundTaskTestCase(TestCase):
     def test_background_task_delete(self):
         queue = get_queue('default')
         job = queue.enqueue(self.dummy_job_default)
+        url = reverse('core-api:rqtask-delete', args=[job.id])
 
-        response = self.client.post(reverse('core-api:rqtask-delete', args=[job.id]), **self.header)
+        # Attempt to load view without permission
+        response = self.client.get(url, **self.header)
+        self.assertEqual(response.status_code, 403)
+
+        # Load view with permission
+        self.user.is_superuser = True
+        self.user.save()
+        response = self.client.post(url, **self.header)
         self.assertEqual(response.status_code, 200)
         self.assertFalse(RQ_Job.exists(job.id, connection=queue.connection))
         queue = get_queue('default')
         self.assertNotIn(job.id, queue.job_ids)
 
     def test_background_task_requeue(self):
-        queue = get_queue('default')
-
         # Enqueue & run a job that will fail
+        queue = get_queue('default')
         job = queue.enqueue(self.dummy_job_failing)
         worker = get_worker('default')
         with disable_logging():
             worker.work(burst=True)
         self.assertTrue(job.is_failed)
+        url = reverse('core-api:rqtask-requeue', args=[job.id])
+
+        # Attempt to requeue the job without permission
+        response = self.client.post(url, **self.header)
+        self.assertEqual(response.status_code, 403)
 
         # Re-enqueue the failed job and check that its status has been reset
-        response = self.client.post(reverse('core-api:rqtask-requeue', args=[job.id]), **self.header)
+        self.user.is_superuser = True
+        self.user.save()
+        response = self.client.post(url, **self.header)
         self.assertEqual(response.status_code, 200)
         job = RQ_Job.fetch(job.id, queue.connection)
         self.assertFalse(job.is_failed)
 
     def test_background_task_enqueue(self):
-        queue = get_queue('default')
-
         # Enqueue some jobs that each depends on its predecessor
+        queue = get_queue('default')
         job = previous_job = None
         for _ in range(0, 3):
             job = queue.enqueue(self.dummy_job_default, depends_on=previous_job)
             previous_job = job
+        url = reverse('core-api:rqtask-enqueue', args=[job.id])
 
         # Check that the last job to be enqueued has a status of deferred
         self.assertIsNotNone(job)
         self.assertEqual(job.get_status(), JobStatus.DEFERRED)
         self.assertIsNone(job.enqueued_at)
 
+        # Attempt to force-enqueue the job without permission
+        response = self.client.post(url, **self.header)
+        self.assertEqual(response.status_code, 403)
+
         # Force-enqueue the deferred job
-        response = self.client.post(reverse('core-api:rqtask-enqueue', args=[job.id]), **self.header)
+        self.user.is_superuser = True
+        self.user.save()
+        response = self.client.post(url, **self.header)
         self.assertEqual(response.status_code, 200)
 
         # Check that job's status is updated correctly
@@ -242,19 +283,27 @@ class BackgroundTaskTestCase(TestCase):
 
     def test_background_task_stop(self):
         queue = get_queue('default')
-
         worker = get_worker('default')
         job = queue.enqueue(self.dummy_job_default)
         worker.prepare_job_execution(job)
-
+        url = reverse('core-api:rqtask-stop', args=[job.id])
         self.assertEqual(job.get_status(), JobStatus.STARTED)
-        response = self.client.post(reverse('core-api:rqtask-stop', args=[job.id]), **self.header)
+
+        # Attempt to stop the task without permission
+        response = self.client.post(url, **self.header)
+        self.assertEqual(response.status_code, 403)
+
+        # Stop the task
+        self.user.is_superuser = True
+        self.user.save()
+        response = self.client.post(url, **self.header)
         self.assertEqual(response.status_code, 200)
         with disable_logging():
             worker.monitor_work_horse(job, queue)  # Sets the job as Failed and removes from Started
         started_job_registry = StartedJobRegistry(queue.name, connection=queue.connection)
         self.assertEqual(len(started_job_registry), 0)
 
+        # Verify that the task was cancelled
         canceled_job_registry = FailedJobRegistry(queue.name, connection=queue.connection)
         self.assertEqual(len(canceled_job_registry), 1)
         self.assertIn(job.id, canceled_job_registry)
@@ -262,19 +311,34 @@ class BackgroundTaskTestCase(TestCase):
     def test_worker_list(self):
         worker1 = get_worker('default', name=uuid.uuid4().hex)
         worker1.register_birth()
-
         worker2 = get_worker('high')
         worker2.register_birth()
+        url = reverse('core-api:rqworker-list')
 
-        response = self.client.get(reverse('core-api:rqworker-list'), **self.header)
+        # Attempt to fetch the worker list without permission
+        response = self.client.get(url, **self.header)
+        self.assertEqual(response.status_code, 403)
+
+        # Fetch the worker list
+        self.user.is_superuser = True
+        self.user.save()
+        response = self.client.get(url, **self.header)
         self.assertEqual(response.status_code, 200)
         self.assertIn(str(worker1.name), str(response.content))
 
     def test_worker(self):
         worker1 = get_worker('default', name=uuid.uuid4().hex)
         worker1.register_birth()
+        url = reverse('core-api:rqworker-detail', args=[worker1.name])
 
-        response = self.client.get(reverse('core-api:rqworker-detail', args=[worker1.name]), **self.header)
+        # Attempt to fetch a worker without permission
+        response = self.client.get(url, **self.header)
+        self.assertEqual(response.status_code, 403)
+
+        # Fetch the worker
+        self.user.is_superuser = True
+        self.user.save()
+        response = self.client.get(url, **self.header)
         self.assertEqual(response.status_code, 200)
         self.assertIn(str(worker1.name), str(response.content))
         self.assertIn('birth_date', str(response.content))

netbox/core/tests/test_views.py

@@ -158,7 +158,7 @@ class BackgroundTaskTestCase(TestCase):
 
     def setUp(self):
         super().setUp()
-        self.user.is_staff = True
+        self.user.is_superuser = True
         self.user.is_active = True
         self.user.save()
 
@@ -171,13 +171,13 @@ class BackgroundTaskTestCase(TestCase):
         url = reverse('core:background_queue_list')
 
         # Attempt to load view without permission
-        self.user.is_staff = False
+        self.user.is_superuser = False
         self.user.save()
         response = self.client.get(url)
         self.assertEqual(response.status_code, 403)
 
         # Load view with permission
-        self.user.is_staff = True
+        self.user.is_superuser = True
         self.user.save()
         response = self.client.get(url)
         self.assertEqual(response.status_code, 200)
@@ -356,7 +356,7 @@ class SystemTestCase(TestCase):
     def setUp(self):
         super().setUp()
 
-        self.user.is_staff = True
+        self.user.is_superuser = True
         self.user.save()
 
     def test_system_view_default(self):

netbox/core/views.py

@@ -372,7 +372,7 @@ class ConfigRevisionRestoreView(ContentTypePermissionRequiredMixin, View):
 class BaseRQView(UserPassesTestMixin, View):
 
     def test_func(self):
-        return self.request.user.is_staff
+        return self.request.user.is_superuser
 
 
 class BackgroundQueueListView(TableMixin, BaseRQView):
@@ -555,7 +555,7 @@ class WorkerView(BaseRQView):
 class SystemView(UserPassesTestMixin, View):
 
     def test_func(self):
-        return self.request.user.is_staff
+        return self.request.user.is_superuser
 
     def get(self, request):
 
@@ -638,7 +638,7 @@ class BasePluginView(UserPassesTestMixin, View):
     CACHE_KEY_CATALOG_ERROR = 'plugins-catalog-error'
 
     def test_func(self):
-        return self.request.user.is_staff
+        return self.request.user.is_superuser
 
     def get_cached_plugins(self, request):
         catalog_plugins = {}

netbox/dcim/api/serializers_/cables.py

@@ -1,10 +1,8 @@
-from django.contrib.contenttypes.models import ContentType
 from drf_spectacular.types import OpenApiTypes
 from drf_spectacular.utils import extend_schema_field
 from rest_framework import serializers
 
 from dcim.choices import *
-from dcim.constants import *
 from dcim.models import Cable, CablePath, CableTermination
 from netbox.api.fields import ChoiceField, ContentTypeField
 from netbox.api.serializers import BaseModelSerializer, GenericObjectSerializer, NetBoxModelSerializer
@@ -51,9 +49,11 @@ class TracedCableSerializer(BaseModelSerializer):
 
 class CableTerminationSerializer(NetBoxModelSerializer):
     termination_type = ContentTypeField(
-        queryset=ContentType.objects.filter(CABLE_TERMINATION_MODELS)
+        read_only=True,
+    )
+    termination = serializers.SerializerMethodField(
+        read_only=True,
     )
-    termination = serializers.SerializerMethodField(read_only=True)
 
     class Meta:
         model = CableTermination
@@ -61,6 +61,8 @@ class CableTerminationSerializer(NetBoxModelSerializer):
             'id', 'url', 'display', 'cable', 'cable_end', 'termination_type', 'termination_id',
             'termination', 'created', 'last_updated',
         ]
+        read_only_fields = fields
+        brief_fields = ('id', 'url', 'display', 'cable', 'cable_end', 'termination_type', 'termination_id')
 
     @extend_schema_field(serializers.JSONField(allow_null=True))
     def get_termination(self, obj):

netbox/dcim/api/views.py

@@ -16,7 +16,7 @@ from extras.api.mixins import ConfigContextQuerySetMixin, RenderConfigMixin
 from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired
 from netbox.api.metadata import ContentTypeMetadata
 from netbox.api.pagination import StripCountAnnotationsPaginator
-from netbox.api.viewsets import NetBoxModelViewSet, MPTTLockedMixin
+from netbox.api.viewsets import NetBoxModelViewSet, MPTTLockedMixin, NetBoxReadOnlyModelViewSet
 from netbox.api.viewsets.mixins import SequentialBulkCreatesMixin
 from utilities.api import get_serializer_for_model
 from utilities.query_functions import CollateAsChar
@@ -563,7 +563,7 @@ class CableViewSet(NetBoxModelViewSet):
     filterset_class = filtersets.CableFilterSet
 
 
-class CableTerminationViewSet(NetBoxModelViewSet):
+class CableTerminationViewSet(NetBoxReadOnlyModelViewSet):
     metadata_class = ContentTypeMetadata
     queryset = CableTermination.objects.all()
     serializer_class = serializers.CableTerminationSerializer

netbox/dcim/tests/test_api.py

@@ -2376,6 +2376,33 @@ class CableTest(APIViewTestCases.APIViewTestCase):
         ]
 
 
+class CableTerminationTest(
+    APIViewTestCases.GetObjectViewTestCase,
+    APIViewTestCases.ListObjectsViewTestCase,
+):
+    model = CableTermination
+    brief_fields = ['cable', 'cable_end', 'display', 'id', 'termination_id', 'termination_type', 'url']
+
+    @classmethod
+    def setUpTestData(cls):
+        device1 = create_test_device('Device 1')
+        device2 = create_test_device('Device 2')
+
+        interfaces = []
+        for device in (device1, device2):
+            for i in range(0, 10):
+                interfaces.append(Interface(device=device, type=InterfaceTypeChoices.TYPE_1GE_FIXED, name=f'eth{i}'))
+        Interface.objects.bulk_create(interfaces)
+
+        cables = (
+            Cable(a_terminations=[interfaces[0]], b_terminations=[interfaces[10]], label='Cable 1'),
+            Cable(a_terminations=[interfaces[1]], b_terminations=[interfaces[11]], label='Cable 2'),
+            Cable(a_terminations=[interfaces[2]], b_terminations=[interfaces[12]], label='Cable 3'),
+        )
+        for cable in cables:
+            cable.save()
+
+
 class ConnectedDeviceTest(APITestCase):
 
     @classmethod

netbox/extras/api/urls.py

@@ -1,10 +1,8 @@
 from django.urls import include, path
 
-from core.api.views import ObjectTypeViewSet
 from netbox.api.routers import NetBoxRouter
 from . import views
 
-
 router = NetBoxRouter()
 router.APIRootView = views.ExtrasRootView
 
@@ -29,9 +27,6 @@ router.register('config-context-profiles', views.ConfigContextProfileViewSet)
 router.register('config-templates', views.ConfigTemplateViewSet)
 router.register('scripts', views.ScriptViewSet, basename='script')
 
-# TODO: Remove in NetBox v4.5
-router.register('object-types', ObjectTypeViewSet)
-
 app_name = 'extras-api'
 urlpatterns = [
     path('dashboard/', views.DashboardView.as_view(), name='dashboard'),

netbox/extras/events.py

@@ -95,7 +95,11 @@ def process_event_rules(event_rules, object_type, event_type, data, username=Non
             continue
 
         # Compile event data
-        event_data = event_rule.action_data or {}
+        if event_rule.action_type == EventRuleActionChoices.SCRIPT:
+            event_data = {}
+        else:
+            event_data = event_rule.action_data or {}
+
         event_data.update(data)
 
         # Webhooks
@@ -131,6 +135,20 @@ def process_event_rules(event_rules, object_type, event_type, data, username=Non
         elif event_rule.action_type == EventRuleActionChoices.SCRIPT:
             # Resolve the script from action parameters
             script = event_rule.action_object.python_class()
+            if event_rule.action_data:
+                form = script.as_form(event_rule.action_data)
+                if form.is_valid():
+                    form.cleaned_data.pop('_schedule_at')
+                    form.cleaned_data.pop('_interval')
+                    form.cleaned_data.pop('_commit')
+                    event_data.update(form.cleaned_data)
+                else:
+                    logger.error(
+                        _("Processing event rule {event_rule} failed - Cannot validate script form: {errors}").format(
+                            event_rule=event_rule, errors=form.errors
+                        )
+                    )
+                    continue
 
             # Enqueue a Job to record the script's execution
             from extras.jobs import ScriptJob

netbox/extras/scripts.py

@@ -1,12 +1,9 @@
 import inspect
-import json
 import logging
 import os
 import re
 
-import yaml
 from django import forms
-from django.conf import settings
 from django.core.files.storage import storages
 from django.core.validators import RegexValidator
 from django.utils import timezone
@@ -488,7 +485,7 @@ class BaseScript:
         if self.fieldsets:
             fieldsets.extend(self.fieldsets)
         else:
-            fields = list(name for name, _ in self._get_vars().items())
+            fields = list(name for name, __ in self._get_vars().items())
             fieldsets.append((_('Script Data'), fields))
 
         # Append the default fieldset if defined in the Meta class
@@ -580,40 +577,6 @@ class BaseScript:
         self._log(message, obj, level=LogLevelChoices.LOG_FAILURE)
         self.failed = True
 
-    #
-    # Convenience functions
-    #
-
-    def load_yaml(self, filename):
-        """
-        Return data from a YAML file
-        """
-        # TODO: DEPRECATED: Remove this method in v4.5
-        self._log(
-            _("load_yaml is deprecated and will be removed in v4.5"),
-            level=LogLevelChoices.LOG_WARNING
-        )
-        file_path = os.path.join(settings.SCRIPTS_ROOT, filename)
-        with open(file_path, 'r') as datafile:
-            data = yaml.load(datafile, Loader=yaml.SafeLoader)
-
-        return data
-
-    def load_json(self, filename):
-        """
-        Return data from a JSON file
-        """
-        # TODO: DEPRECATED: Remove this method in v4.5
-        self._log(
-            _("load_json is deprecated and will be removed in v4.5"),
-            level=LogLevelChoices.LOG_WARNING
-        )
-        file_path = os.path.join(settings.SCRIPTS_ROOT, filename)
-        with open(file_path, 'r') as datafile:
-            data = json.load(datafile)
-
-        return data
-
     #
     # Legacy Report functionality
     #

netbox/extras/tests/test_event_rules.py

@@ -363,7 +363,7 @@ class EventRuleTest(APITestCase):
             body = json.loads(request.body)
             self.assertEqual(body['event'], 'created')
             self.assertEqual(body['timestamp'], job.kwargs['timestamp'])
-            self.assertEqual(body['model'], 'site')
+            self.assertEqual(body['object_type'], 'dcim.site')
             self.assertEqual(body['username'], 'testuser')
             self.assertEqual(body['request_id'], str(request_id))
             self.assertEqual(body['data']['name'], 'Site 1')

netbox/extras/tests/test_scripts.py

@@ -1,5 +1,3 @@
-import logging
-import tempfile
 from datetime import date, datetime, timezone
 from decimal import Decimal
 
@@ -9,7 +7,6 @@ from netaddr import IPAddress, IPNetwork
 
 from dcim.models import DeviceRole
 from extras.scripts import *
-from utilities.testing import disable_logging
 
 CHOICES = (
     ('ff0000', 'Red'),
@@ -35,35 +32,6 @@ JSON_DATA = """
 """
 
 
-class ScriptTest(TestCase):
-
-    def test_load_yaml(self):
-        datafile = tempfile.NamedTemporaryFile()
-        datafile.write(bytes(YAML_DATA, 'UTF-8'))
-        datafile.seek(0)
-
-        with disable_logging(level=logging.WARNING):
-            data = Script().load_yaml(datafile.name)
-        self.assertEqual(data, {
-            'Foo': 123,
-            'Bar': 456,
-            'Baz': ['A', 'B', 'C'],
-        })
-
-    def test_load_json(self):
-        datafile = tempfile.NamedTemporaryFile()
-        datafile.write(bytes(JSON_DATA, 'UTF-8'))
-        datafile.seek(0)
-
-        with disable_logging(level=logging.WARNING):
-            data = Script().load_json(datafile.name)
-        self.assertEqual(data, {
-            'Foo': 123,
-            'Bar': 456,
-            'Baz': ['A', 'B', 'C'],
-        })
-
-
 class ScriptVariablesTest(TestCase):
 
     def test_stringvar(self):

netbox/extras/webhooks.py

@@ -52,7 +52,6 @@ def send_webhook(event_rule, object_type, event_type, data, timestamp, username,
         'event': WEBHOOK_EVENT_TYPES.get(event_type, event_type),
         'timestamp': timestamp,
         'object_type': '.'.join(object_type.natural_key()),
-        'model': object_type.model,
         'username': username,
         'request_id': request.id if request else None,
         'data': data,
@@ -100,7 +99,7 @@ def send_webhook(event_rule, object_type, event_type, data, timestamp, username,
         'data': body.encode('utf8'),
     }
     logger.info(
-        f"Sending {params['method']} request to {params['url']} ({context['model']} {context['event']})"
+        f"Sending {params['method']} request to {params['url']} ({context['object_type']} {context['event']})"
     )
     logger.debug(params)
     try:

netbox/netbox/api/authentication.py

@@ -2,47 +2,90 @@ import logging
 
 from django.conf import settings
 from django.utils import timezone
-from rest_framework import authentication, exceptions
+from drf_spectacular.extensions import OpenApiAuthenticationExtension
+from rest_framework import exceptions
+from rest_framework.authentication import BaseAuthentication, get_authorization_header
 from rest_framework.permissions import BasePermission, DjangoObjectPermissions, SAFE_METHODS
 
 from netbox.config import get_config
+from users.constants import TOKEN_PREFIX
 from users.models import Token
 from utilities.request import get_client_ip
 
+V1_KEYWORD = 'Token'
+V2_KEYWORD = 'Bearer'
 
-class TokenAuthentication(authentication.TokenAuthentication):
+
+class TokenAuthentication(BaseAuthentication):
     """
     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()
+        # Authorization header is not present; ignore
+        if not (auth := get_authorization_header(request).split()):
+            return
+        # Unrecognized header; ignore
+        if auth[0].lower() not in (V1_KEYWORD.lower().encode(), V2_KEYWORD.lower().encode()):
+            return
+        # Check for extraneous token content
+        if len(auth) != 2:
+            raise exceptions.AuthenticationFailed(
+                'Invalid authorization header: Must be in the form "Bearer <key>.<token>" or "Token <token>"'
+            )
+        # Extract the key (if v2) & token plaintext from the auth header
         try:
-            token = model.objects.prefetch_related('user').get(key=key)
-        except model.DoesNotExist:
-            raise exceptions.AuthenticationFailed("Invalid token")
+            auth_value = auth[1].decode()
+        except UnicodeError:
+            raise exceptions.AuthenticationFailed("Invalid authorization header: Token contains invalid characters")
+
+        # Infer token version from presence or absence of prefix
+        version = 2 if auth_value.startswith(TOKEN_PREFIX) else 1
+
+        if version == 1:
+            key, plaintext = None, auth_value
+        else:
+            auth_value = auth_value.removeprefix(TOKEN_PREFIX)
+            try:
+                key, plaintext = auth_value.split('.', 1)
+            except ValueError:
+                raise exceptions.AuthenticationFailed(
+                    "Invalid authorization header: Could not parse key from v2 token. Did you mean to use 'Token' "
+                    "instead of 'Bearer'?"
+                )
+
+        # Look for a matching token in the database
+        try:
+            qs = Token.objects.prefetch_related('user')
+            if version == 1:
+                # Fetch v1 token by querying plaintext value directly
+                token = qs.get(version=version, plaintext=plaintext)
+            else:
+                # Fetch v2 token by key, then validate the plaintext
+                token = qs.get(version=version, key=key)
+                if not token.validate(plaintext):
+                    # Key is valid but plaintext is not. Raise DoesNotExist to guard against key enumeration.
+                    raise Token.DoesNotExist()
+        except Token.DoesNotExist:
+            raise exceptions.AuthenticationFailed(f"Invalid v{version} token")
+
+        # 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."
+                )
+
+        # Enforce the Token's expiration time, if one has been set.
+        if token.is_expired:
+            raise exceptions.AuthenticationFailed("Token expired")
 
         # Update last used, but only once per minute at most. This reduces write load on the database
         if not token.last_used or (timezone.now() - token.last_used).total_seconds() > 60:
@@ -54,11 +97,8 @@ class TokenAuthentication(authentication.TokenAuthentication):
             else:
                 Token.objects.filter(pk=token.pk).update(last_used=timezone.now())
 
-        # Enforce the Token's expiration time, if one has been set.
-        if token.is_expired:
-            raise exceptions.AuthenticationFailed("Token expired")
-
         user = token.user
+
         # When LDAP authentication is active try to load user data from LDAP directory
         if 'netbox.authentication.LDAPBackend' in settings.REMOTE_AUTH_BACKEND:
             from netbox.authentication import LDAPBackend
@@ -132,3 +172,17 @@ class IsAuthenticatedOrLoginNotRequired(BasePermission):
         if not settings.LOGIN_REQUIRED:
             return True
         return request.user.is_authenticated
+
+
+class TokenScheme(OpenApiAuthenticationExtension):
+    target_class = 'netbox.api.authentication.TokenAuthentication'
+    name = 'tokenAuth'
+    match_subclasses = True
+
+    def get_security_definition(self, auto_schema):
+        return {
+            'type': 'apiKey',
+            'in': 'header',
+            'name': 'Authorization',
+            'description': '`Token <token>` (v1) or `Bearer <key>.<token>` (v2)',
+        }

netbox/netbox/authentication/__init__.py

@@ -184,14 +184,13 @@ class RemoteUserBackend(_RemoteUserBackend):
         else:
             user.groups.clear()
             logger.debug(f"Stripping user {user} from Groups")
+
+        # Evaluate superuser status
         user.is_superuser = self._is_superuser(user)
         logger.debug(f"User {user} is Superuser: {user.is_superuser}")
         logger.debug(
             f"User {user} should be Superuser: {self._is_superuser(user)}")
 
-        user.is_staff = self._is_staff(user)
-        logger.debug(f"User {user} is Staff: {user.is_staff}")
-        logger.debug(f"User {user} should be Staff: {self._is_staff(user)}")
         user.save()
         return user
 
@@ -251,19 +250,8 @@ class RemoteUserBackend(_RemoteUserBackend):
         return bool(result)
 
     def _is_staff(self, user):
-        logger = logging.getLogger('netbox.auth.RemoteUserBackend')
-        staff_groups = settings.REMOTE_AUTH_STAFF_GROUPS
-        logger.debug(f"Superuser Groups: {staff_groups}")
-        staff_users = settings.REMOTE_AUTH_STAFF_USERS
-        logger.debug(f"Staff Users :{staff_users}")
-        user_groups = set()
-        for g in user.groups.all():
-            user_groups.add(g.name)
-        logger.debug(f"User {user.username} is in Groups:{user_groups}")
-        result = user.username in staff_users or (
-            set(user_groups) & set(staff_groups))
-        logger.debug(f"User {user.username} in Staff Users :{result}")
-        return bool(result)
+        # Retain for pre-v4.5 compatibility
+        return user.is_superuser
 
     def configure_user(self, request, user):
         logger = logging.getLogger('netbox.auth.RemoteUserBackend')

netbox/netbox/configuration_example.py

@@ -68,6 +68,16 @@ REDIS = {
 # https://docs.djangoproject.com/en/stable/ref/settings/#std:setting-SECRET_KEY
 SECRET_KEY = ''
 
+# Define a mapping of cryptographic peppers to use when hashing API tokens. A minimum of one pepper is required to
+# enable v2 API tokens (NetBox v4.5+). Define peppers as a mapping of numeric ID to pepper value, as shown below. Each
+# pepper must be at least 50 characters in length.
+#
+#     API_TOKEN_PEPPERS = {
+#         1: "<random string>",
+#         2: "<random string>",
+#     }
+API_TOKEN_PEPPERS = {}
+
 
 #########################
 #                       #
@@ -81,9 +91,6 @@ ADMINS = [
     # ('John Doe', 'jdoe@example.com'),
 ]
 
-# Permit the retrieval of API tokens after their creation.
-ALLOW_TOKEN_RETRIEVAL = False
-
 # Enable any desired validators for local account passwords below. For a list of included validators, please see the
 # Django documentation at https://docs.djangoproject.com/en/stable/topics/auth/passwords/#password-validation.
 AUTH_PASSWORD_VALIDATORS = [

netbox/netbox/configuration_testing.py

@@ -43,7 +43,9 @@ SECRET_KEY = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'
 
 DEFAULT_PERMISSIONS = {}
 
-ALLOW_TOKEN_RETRIEVAL = True
+API_TOKEN_PEPPERS = {
+    1: 'TEST-VALUE-DO-NOT-USE-TEST-VALUE-DO-NOT-USE-TEST-VALUE-DO-NOT-USE',
+}
 
 LOGGING = {
     'version': 1,

netbox/netbox/context_processors.py

@@ -28,7 +28,6 @@ def preferences(request):
     user_preferences = request.user.config if request.user.is_authenticated else {}
     return {
         'preferences': user_preferences,
-        'htmx_navigation': user_preferences.get('ui.htmx_navigation', False) == 'true'
     }
 
 

netbox/netbox/plugins/views.py

@@ -3,12 +3,12 @@ from collections import OrderedDict
 from django.apps import apps
 from django.urls.exceptions import NoReverseMatch
 from drf_spectacular.utils import extend_schema
-from rest_framework import permissions
 from rest_framework.response import Response
 from rest_framework.reverse import reverse
 from rest_framework.views import APIView
 
 from netbox.registry import registry
+from utilities.api import IsSuperuser
 
 
 @extend_schema(exclude=True)
@@ -16,7 +16,7 @@ class InstalledPluginsAPIView(APIView):
     """
     API view for listing all installed plugins
     """
-    permission_classes = [permissions.IsAdminUser]
+    permission_classes = [IsSuperuser]
     _ignore_model_permissions = True
     schema = None
 

netbox/netbox/preferences.py

@@ -26,16 +26,6 @@ def get_csv_delimiters():
 PREFERENCES = {
 
     # User interface
-    'ui.htmx_navigation': UserPreference(
-        label=_('HTMX Navigation'),
-        choices=(
-            ('', _('Disabled')),
-            ('true', _('Enabled')),
-        ),
-        description=_('Enable dynamic UI navigation'),
-        default=False,
-        warning=_('Experimental feature')
-    ),
     'locale.language': UserPreference(
         label=_('Language'),
         choices=(

netbox/netbox/settings.py

@@ -20,6 +20,7 @@ from netbox.plugins import PluginConfig
 from netbox.registry import registry
 import storages.utils  # type: ignore
 from utilities.release import load_release_data
+from utilities.security import validate_peppers
 from utilities.string import trailing_slash
 from .monkey import get_unique_validators
 
@@ -43,9 +44,9 @@ VERSION = RELEASE.full_version  # Retained for backward compatibility
 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
 # Validate Python version
-if sys.version_info < (3, 10):
+if sys.version_info < (3, 12):
     raise RuntimeError(
-        f"NetBox requires Python 3.10 or later. (Currently installed: Python {platform.python_version()})"
+        f"NetBox requires Python 3.12 or later. (Currently installed: Python {platform.python_version()})"
     )
 
 #
@@ -75,8 +76,8 @@ elif hasattr(configuration, 'DATABASE') and hasattr(configuration, 'DATABASES'):
 
 # Set static config parameters
 ADMINS = getattr(configuration, 'ADMINS', [])
-ALLOW_TOKEN_RETRIEVAL = getattr(configuration, 'ALLOW_TOKEN_RETRIEVAL', False)
 ALLOWED_HOSTS = getattr(configuration, 'ALLOWED_HOSTS')  # Required
+API_TOKEN_PEPPERS = getattr(configuration, 'API_TOKEN_PEPPERS', {})
 AUTH_PASSWORD_VALIDATORS = getattr(configuration, 'AUTH_PASSWORD_VALIDATORS', [
     {
         "NAME": "django.contrib.auth.password_validation.MinimumLengthValidator",
@@ -174,8 +175,6 @@ REMOTE_AUTH_SUPERUSERS = getattr(configuration, 'REMOTE_AUTH_SUPERUSERS', [])
 REMOTE_AUTH_USER_EMAIL = getattr(configuration, 'REMOTE_AUTH_USER_EMAIL', 'HTTP_REMOTE_USER_EMAIL')
 REMOTE_AUTH_USER_FIRST_NAME = getattr(configuration, 'REMOTE_AUTH_USER_FIRST_NAME', 'HTTP_REMOTE_USER_FIRST_NAME')
 REMOTE_AUTH_USER_LAST_NAME = getattr(configuration, 'REMOTE_AUTH_USER_LAST_NAME', 'HTTP_REMOTE_USER_LAST_NAME')
-REMOTE_AUTH_STAFF_GROUPS = getattr(configuration, 'REMOTE_AUTH_STAFF_GROUPS', [])
-REMOTE_AUTH_STAFF_USERS = getattr(configuration, 'REMOTE_AUTH_STAFF_USERS', [])
 # Required by extras/migrations/0109_script_models.py
 REPORTS_ROOT = getattr(configuration, 'REPORTS_ROOT', os.path.join(BASE_DIR, 'reports')).rstrip('/')
 RQ_DEFAULT_TIMEOUT = getattr(configuration, 'RQ_DEFAULT_TIMEOUT', 300)
@@ -229,6 +228,12 @@ if len(SECRET_KEY) < 50:
         f"  python {BASE_DIR}/generate_secret_key.py"
     )
 
+# Validate API token peppers
+if API_TOKEN_PEPPERS:
+    validate_peppers(API_TOKEN_PEPPERS)
+else:
+    warnings.warn("API_TOKEN_PEPPERS is not defined. v2 API tokens cannot be used.")
+
 # Validate update repo URL and timeout
 if RELEASE_CHECK_URL:
     try:

netbox/netbox/tables/columns.py

@@ -270,7 +270,7 @@ class ActionsColumn(tables.Column):
         if not (self.actions or self.extra_buttons):
             return ''
         # Skip dummy records (e.g. available VLANs or IP ranges replacing individual IPs)
-        if type(record) is not model or not getattr(record, 'pk', None):
+        if not isinstance(record, model) or not getattr(record, 'pk', None):
             return ''
 
         if request := getattr(table, 'context', {}).get('request'):

netbox/netbox/tests/test_authentication.py

@@ -8,6 +8,7 @@ from rest_framework.test import APIClient
 
 from core.models import ObjectType
 from dcim.models import Rack, Site
+from users.constants import TOKEN_PREFIX
 from users.models import Group, ObjectPermission, Token, User
 from utilities.testing import TestCase
 from utilities.testing.api import APITestCase
@@ -16,67 +17,159 @@ 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')
-
+    def test_no_token(self):
         # Request without a token should return a 403
-        response = self.client.get(url)
+        response = self.client.get(reverse('dcim-api:site-list'))
         self.assertEqual(response.status_code, 403)
 
+    @override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
+    def test_v1_token_valid(self):
+        # Create a v1 token
+        token = Token.objects.create(version=1, user=self.user)
+
         # 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)
+        header = f'Token {token.token}'
+        response = self.client.get(reverse('dcim-api:site-list'), HTTP_AUTHORIZATION=header)
+        self.assertEqual(response.status_code, 200, response.data)
 
         # Check that the token's last_used time has been updated
         token.refresh_from_db()
         self.assertIsNotNone(token.last_used)
 
+    @override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
+    def test_v1_token_invalid(self):
+        # Invalid token should return a 403
+        header = 'Token XXXXXXXXXX'
+        response = self.client.get(reverse('dcim-api:site-list'), HTTP_AUTHORIZATION=header)
+        self.assertEqual(response.status_code, 403)
+        self.assertEqual(response.data['detail'], "Invalid v1 token")
+
+    @override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
+    def test_v2_token_valid(self):
+        # Create a v2 token
+        token = Token.objects.create(version=2, user=self.user)
+
+        # Valid token should return a 200
+        header = f'Bearer {TOKEN_PREFIX}{token.key}.{token.token}'
+        response = self.client.get(reverse('dcim-api:site-list'), HTTP_AUTHORIZATION=header)
+        self.assertEqual(response.status_code, 200, response.data)
+
+        # Check that the token's last_used time has been updated
+        token.refresh_from_db()
+        self.assertIsNotNone(token.last_used)
+
+    @override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
+    def test_v2_token_invalid(self):
+        # Invalid token should return a 403
+        header = f'Bearer {TOKEN_PREFIX}XXXXXX.XXXXXXXXXX'
+        response = self.client.get(reverse('dcim-api:site-list'), HTTP_AUTHORIZATION=header)
+        self.assertEqual(response.status_code, 403)
+        self.assertEqual(response.data['detail'], "Invalid v2 token")
+
     @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}')
+        # Create v1 & v2 tokens
+        future = datetime.datetime(2100, 1, 1, tzinfo=datetime.timezone.utc)
+        token1 = Token.objects.create(version=1, user=self.user, expires=future)
+        token2 = Token.objects.create(version=2, user=self.user, expires=future)
+
+        # Request with a non-expired token should succeed
+        response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token1.token}')
+        self.assertEqual(response.status_code, 200)
+        response = self.client.get(url, HTTP_AUTHORIZATION=f'Bearer {TOKEN_PREFIX}{token2.key}.{token2.token}')
         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}')
+        past = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
+        token1.expires = past
+        token1.save()
+        token2.expires = past
+        token2.save()
+        response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token1.key}')
+        self.assertEqual(response.status_code, 403)
+        response = self.client.get(url, HTTP_AUTHORIZATION=f'Bearer {TOKEN_PREFIX}{token2.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',
-        }
+        data = [
+            {
+                'name': 'Site 1',
+                'slug': 'site-1',
+            },
+            {
+                'name': 'Site 2',
+                'slug': 'site-2',
+            },
+        ]
+        self.add_permissions('dcim.view_site', 'dcim.add_site')
 
-        # 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}')
+        # Create v1 & v2 tokens
+        token1 = Token.objects.create(version=1, user=self.user, write_enabled=False)
+        token2 = Token.objects.create(version=2, user=self.user, write_enabled=False)
+
+        token1_header = f'Token {token1.token}'
+        token2_header = f'Bearer {TOKEN_PREFIX}{token2.key}.{token2.token}'
+
+        # GET request with a write-disabled token should succeed
+        response = self.client.get(url, HTTP_AUTHORIZATION=token1_header)
+        self.assertEqual(response.status_code, 200)
+        response = self.client.get(url, HTTP_AUTHORIZATION=token2_header)
+        self.assertEqual(response.status_code, 200)
+
+        # POST request with a write-disabled token should fail
+        response = self.client.post(url, data[0], format='json', HTTP_AUTHORIZATION=token1_header)
+        self.assertEqual(response.status_code, 403)
+        response = self.client.post(url, data[1], format='json', HTTP_AUTHORIZATION=token2_header)
         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)
+        # POST request with a write-enabled token should succeed
+        token1.write_enabled = True
+        token1.save()
+        token2.write_enabled = True
+        token2.save()
+        response = self.client.post(url, data[0], format='json', HTTP_AUTHORIZATION=token1_header)
+        self.assertEqual(response.status_code, 201)
+        response = self.client.post(url, data[1], format='json', HTTP_AUTHORIZATION=token2_header)
+        self.assertEqual(response.status_code, 201)
 
     @override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
     def test_token_allowed_ips(self):
         url = reverse('dcim-api:site-list')
 
+        # Create v1 & v2 tokens
+        token1 = Token.objects.create(version=1, user=self.user, allowed_ips=['192.0.2.0/24'])
+        token2 = Token.objects.create(version=2, user=self.user, allowed_ips=['192.0.2.0/24'])
+
         # 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')
+        response = self.client.get(
+            url,
+            HTTP_AUTHORIZATION=f'Token {token1.token}',
+            REMOTE_ADDR='127.0.0.1'
+        )
+        self.assertEqual(response.status_code, 403)
+        response = self.client.get(
+            url,
+            HTTP_AUTHORIZATION=f'Bearer {TOKEN_PREFIX}{token2.key}.{token2.token}',
+            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')
+        # Request from an allowed client IP should succeed
+        response = self.client.get(
+            url,
+            HTTP_AUTHORIZATION=f'Token {token1.token}',
+            REMOTE_ADDR='192.0.2.1'
+        )
+        self.assertEqual(response.status_code, 200)
+        response = self.client.get(
+            url,
+            HTTP_AUTHORIZATION=f'Bearer {TOKEN_PREFIX}{token2.key}.{token2.token}',
+            REMOTE_ADDR='192.0.2.1'
+        )
         self.assertEqual(response.status_code, 200)
 
 
@@ -427,7 +520,7 @@ class ObjectPermissionAPIViewTestCase(TestCase):
         """
         self.user = User.objects.create(username='testuser')
         self.token = Token.objects.create(user=self.user)
-        self.header = {'HTTP_AUTHORIZATION': 'Token {}'.format(self.token.key)}
+        self.header = {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{self.token.key}.{self.token.token}'}
 
     @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
     def test_get_object(self):

netbox/netbox/views/misc.py

@@ -47,9 +47,9 @@ class HomeView(ConditionalLoginRequiredMixin, View):
             ))
             dashboard = get_default_dashboard(config=DEFAULT_DASHBOARD).get_layout()
 
-        # Check whether a new release is available. (Only for staff/superusers.)
+        # Check whether a new release is available. (Only for superusers.)
         new_release = None
-        if request.user.is_staff or request.user.is_superuser:
+        if request.user.is_superuser:
             latest_release = cache.get('latest_release')
             if latest_release:
                 release_version, release_url = latest_release

netbox/templates/account/profile.html

@@ -39,10 +39,6 @@
             <th scope="row">{% trans "Superuser" %}</th>
             <td>{% checkmark request.user.is_superuser %}</td>
           </tr>
-          <tr>
-            <th scope="row">{% trans "Staff" %}</th>
-            <td>{% checkmark request.user.is_staff %}</td>
-          </tr>
         </table>
       </div>
     </div>

netbox/templates/account/token.html

@@ -1,62 +1,8 @@
-{% extends 'generic/object.html' %}
-{% load form_helpers %}
-{% load helpers %}
+{% extends 'users/token.html' %}
 {% load i18n %}
-{% load plugins %}
 
 {% block breadcrumbs %}
-  <li class="breadcrumb-item"><a href="{% url 'account:usertoken_list' %}">{% trans "My API Tokens" %}</a></li>
+  <li class="breadcrumb-item">
+    <a href="{% url 'account:usertoken_list' %}">{% trans "My API Tokens" %}</a>
+  </li>
 {% endblock breadcrumbs %}
-
-{% block title %}{% trans "Token" %} {{ object }}{% endblock %}
-
-{% block subtitle %}{% endblock %}
-
-{% block content %}
-  <div class="row">
-    <div class="col col-md-12">
-      <div class="card">
-        <h2 class="card-header">{% trans "Token" %}</h2>
-        <table class="table table-hover attr-table">
-          <tr>
-            <th scope="row">{% trans "Key" %}</th>
-            <td>
-              {% if key %}
-                <div class="float-end">
-                  {% copy_content "token_id" %}
-                </div>
-                <div id="token_id">{{ key }}</div>
-              {% else %}
-                {{ object.partial }}
-              {% endif %}
-            </td>
-          </tr>
-          <tr>
-            <th scope="row">{% trans "Description" %}</th>
-            <td>{{ object.description|placeholder }}</td>
-          </tr>
-          <tr>
-            <th scope="row">{% trans "Write enabled" %}</th>
-            <td>{% checkmark object.write_enabled %}</td>
-          </tr>
-          <tr>
-            <th scope="row">{% trans "Created" %}</th>
-            <td>{{ object.created|isodatetime }}</td>
-          </tr>
-          <tr>
-            <th scope="row">{% trans "Expires" %}</th>
-            <td>{{ object.expires|isodatetime|placeholder }}</td>
-          </tr>
-          <tr>
-            <th scope="row">{% trans "Last used" %}</th>
-            <td>{{ object.last_used|isodatetime|placeholder }}</td>
-          </tr>
-          <tr>
-            <th scope="row">{% trans "Allowed IPs" %}</th>
-            <td>{{ object.allowed_ips|join:", "|placeholder }}</td>
-          </tr>
-        </table>
-      </div>
-    </div>
-  </div>
-{% endblock %}

netbox/templates/base/layout.html

@@ -95,7 +95,7 @@ Blocks:
 
     {# Page content #}
     <div class="page-wrapper">
-      <div id="page-content" {% htmx_boost %}>
+      <div id="page-content">
 
       {# Page header #}
       {% block header %}

netbox/templates/inc/user_menu.html

@@ -27,15 +27,13 @@
         <div class="mt-1 small text-secondary">
           {% if request.user.is_superuser %}
             {% trans "Admin" %}
-          {% elif request.user.is_staff %}
-            {% trans "Staff" %}
           {% else %}
             {% trans "User" %}
           {% endif %}
         </div>
       </div>
     </a>
-    <div class="dropdown-menu dropdown-menu-end dropdown-menu-arrow" {% htmx_boost %}>
+    <div class="dropdown-menu dropdown-menu-end dropdown-menu-arrow">
       <a href="{% url 'account:profile' %}" class="dropdown-item">
         <i class="mdi mdi-account"></i> {% trans "Profile" %}
       </a>

netbox/templates/media_failure.html

@@ -37,7 +37,7 @@
                 path. Refer to <a href="{{ docs_url }}">the installation documentation</a> for further guidance.
               {% endblocktrans %}
               <ul>
-                {% if request.user.is_staff or request.user.is_superuser %}
+                {% if request.user.is_superuser %}
                   <li><code>STATIC_ROOT: <strong>{{ settings.STATIC_ROOT }}</strong></code></li>
                 {% endif %}
                 <li><code>STATIC_URL: <strong>{{ settings.STATIC_URL }}</strong></code></li>

netbox/templates/users/token.html

@@ -14,9 +14,24 @@
         <h2 class="card-header">{% trans "Token" %}</h2>
         <table class="table table-hover attr-table">
           <tr>
-            <th scope="row">{% trans "Key" %}</th>
-            <td>{% if settings.ALLOW_TOKEN_RETRIEVAL %}{{ object.key }}{% else %}{{ object.partial }}{% endif %}</td>
+            <th scope="row">{% trans "Version" %}</th>
+            <td>{{ object.version }}</td>
           </tr>
+          {% if object.version == 1 %}
+            <tr>
+              <th scope="row">{% trans "Token" %}</th>
+              <td>{{ object.partial }}</td>
+            </tr>
+          {% else %}
+            <tr>
+              <th scope="row">{% trans "Key" %}</th>
+              <td>{{ object }}</td>
+            </tr>
+            <tr>
+              <th scope="row">{% trans "Pepper ID" %}</th>
+              <td>{{ object.pepper_id }}</td>
+            </tr>
+          {% endif %}
           <tr>
             <th scope="row">{% trans "User" %}</th>
             <td>

netbox/templates/users/user.html

@@ -35,10 +35,6 @@
             <th scope="row">{% trans "Active" %}</th>
             <td>{% checkmark object.is_active %}</td>
           </tr>
-          <tr>
-            <th scope="row">{% trans "Staff" %}</th>
-            <td>{% checkmark object.is_staff %}</td>
-          </tr>
           <tr>
             <th scope="row">{% trans "Superuser" %}</th>
             <td>{% checkmark object.is_superuser %}</td>

netbox/users/api/serializers_/tokens.py

@@ -1,4 +1,3 @@
-from django.conf import settings
 from django.contrib.auth import authenticate
 from rest_framework import serializers
 from rest_framework.exceptions import AuthenticationFailed, PermissionDenied
@@ -15,14 +14,13 @@ __all__ = (
 
 
 class TokenSerializer(ValidatedModelSerializer):
-    key = serializers.CharField(
-        min_length=40,
-        max_length=40,
-        allow_blank=True,
+    token = serializers.CharField(
         required=False,
-        write_only=not settings.ALLOW_TOKEN_RETRIEVAL
+        default=Token.generate,
+    )
+    user = UserSerializer(
+        nested=True
     )
-    user = UserSerializer(nested=True)
     allowed_ips = serializers.ListField(
         child=IPNetworkSerializer(),
         required=False,
@@ -33,15 +31,20 @@ class TokenSerializer(ValidatedModelSerializer):
     class Meta:
         model = Token
         fields = (
-            'id', 'url', 'display_url', 'display', 'user', 'created', 'expires', 'last_used', 'key', 'write_enabled',
-            'description', 'allowed_ips',
+            'id', 'url', 'display_url', 'display', 'version', 'key', 'user', 'description', 'created', 'expires',
+            'last_used', 'write_enabled', 'pepper_id', 'allowed_ips', 'token',
         )
-        brief_fields = ('id', 'url', 'display', 'key', 'write_enabled', 'description')
+        read_only_fields = ('key',)
+        brief_fields = ('id', 'url', 'display', 'version', 'key', 'write_enabled', 'description')
 
-    def to_internal_value(self, data):
-        if not getattr(self.instance, 'key', None) and 'key' not in data:
-            data['key'] = Token.generate_key()
-        return super().to_internal_value(data)
+    def get_fields(self):
+        fields = super().get_fields()
+
+        # Make user field read-only if updating an existing Token.
+        if self.instance is not None:
+            fields['user'].read_only = True
+
+        return fields
 
     def validate(self, data):
 
@@ -75,8 +78,8 @@ class TokenProvisionSerializer(TokenSerializer):
     class Meta:
         model = Token
         fields = (
-            'id', 'url', 'display_url', 'display', 'user', 'created', 'expires', 'last_used', 'key', 'write_enabled',
-            'description', 'allowed_ips', 'username', 'password',
+            'id', 'url', 'display_url', 'display', 'version', 'user', 'key', 'created', 'expires', 'last_used', 'key',
+            'write_enabled', 'description', 'allowed_ips', 'username', 'password', 'token',
         )
 
     def validate(self, data):

netbox/users/api/serializers_/users.py

@@ -52,7 +52,7 @@ class UserSerializer(ValidatedModelSerializer):
         model = User
         fields = (
             'id', 'url', 'display_url', 'display', 'username', 'password', 'first_name', 'last_name', 'email',
-            'is_staff', 'is_active', 'date_joined', 'last_login', 'groups', 'permissions',
+            'is_active', 'date_joined', 'last_login', 'groups', 'permissions',
         )
         brief_fields = ('id', 'url', 'display', 'username')
         extra_kwargs = {

netbox/users/choices.py

@@ -0,0 +1,17 @@
+from django.utils.translation import gettext_lazy as _
+
+from utilities.choices import ChoiceSet
+
+__all__ = (
+    'TokenVersionChoices',
+)
+
+
+class TokenVersionChoices(ChoiceSet):
+    V1 = 1
+    V2 = 2
+
+    CHOICES = [
+        (V1, _('v1')),
+        (V2, _('v2')),
+    ]

netbox/users/constants.py

@@ -1,3 +1,5 @@
+import string
+
 from django.db.models import Q
 
 
@@ -7,3 +9,9 @@ OBJECTPERMISSION_OBJECT_TYPES = Q(
 )
 
 CONSTRAINT_TOKEN_USER = '$user'
+
+# API tokens
+TOKEN_PREFIX = 'nbt_'  # Used for v2 tokens only
+TOKEN_KEY_LENGTH = 12
+TOKEN_DEFAULT_LENGTH = 40
+TOKEN_CHARSET = string.ascii_letters + string.digits

netbox/users/filtersets.py

@@ -81,7 +81,7 @@ class UserFilterSet(BaseFilterSet):
     class Meta:
         model = User
         fields = (
-            'id', 'username', 'first_name', 'last_name', 'email', 'date_joined', 'last_login', 'is_staff', 'is_active',
+            'id', 'username', 'first_name', 'last_name', 'email', 'date_joined', 'last_login', 'is_active',
             'is_superuser',
         )
 
@@ -130,15 +130,27 @@ class TokenFilterSet(BaseFilterSet):
         field_name='expires',
         lookup_expr='lte'
     )
+    last_used = django_filters.DateTimeFilter()
+    last_used__gte = django_filters.DateTimeFilter(
+        field_name='last_used',
+        lookup_expr='gte'
+    )
+    last_used__lte = django_filters.DateTimeFilter(
+        field_name='last_used',
+        lookup_expr='lte'
+    )
 
     class Meta:
         model = Token
-        fields = ('id', 'key', 'write_enabled', 'description', 'last_used')
+        fields = (
+            'id', 'version', 'key', 'pepper_id', 'write_enabled', 'description', 'created', 'expires', 'last_used',
+        )
 
     def search(self, queryset, name, value):
         if not value.strip():
             return queryset
         return queryset.filter(
+            Q(key=value) |
             Q(user__username__icontains=value) |
             Q(description__icontains=value)
         )

netbox/users/forms/bulk_edit.py

@@ -37,11 +37,6 @@ class UserBulkEditForm(BulkEditForm):
         widget=BulkEditNullBooleanSelect,
         label=_('Active')
     )
-    is_staff = forms.NullBooleanField(
-        required=False,
-        widget=BulkEditNullBooleanSelect,
-        label=_('Staff status')
-    )
     is_superuser = forms.NullBooleanField(
         required=False,
         widget=BulkEditNullBooleanSelect,
@@ -50,7 +45,7 @@ class UserBulkEditForm(BulkEditForm):
 
     model = User
     fieldsets = (
-        FieldSet('first_name', 'last_name', 'is_active', 'is_staff', 'is_superuser'),
+        FieldSet('first_name', 'last_name', 'is_active', 'is_superuser'),
     )
     nullable_fields = ('first_name', 'last_name')
 

netbox/users/forms/bulk_import.py

@@ -1,6 +1,7 @@
 from django import forms
 from django.utils.translation import gettext as _
 from users.models import *
+from users.choices import TokenVersionChoices
 from utilities.forms import CSVModelForm
 
 
@@ -23,8 +24,7 @@ class UserImportForm(CSVModelForm):
     class Meta:
         model = User
         fields = (
-            'username', 'first_name', 'last_name', 'email', 'password', 'is_staff',
-            'is_active', 'is_superuser'
+            'username', 'first_name', 'last_name', 'email', 'password', 'is_active', 'is_superuser'
         )
 
     def save(self, *args, **kwargs):
@@ -35,12 +35,18 @@ class UserImportForm(CSVModelForm):
 
 
 class TokenImportForm(CSVModelForm):
-    key = forms.CharField(
-        label=_('Key'),
+    version = forms.ChoiceField(
+        choices=TokenVersionChoices,
+        initial=TokenVersionChoices.V2,
         required=False,
-        help_text=_("If no key is provided, one will be generated automatically.")
+        help_text=_("Specify version 1 or 2 (v2 will be used by default)")
+    )
+    token = forms.CharField(
+        label=_('Token'),
+        required=False,
+        help_text=_("If no token is provided, one will be generated automatically.")
     )
 
     class Meta:
         model = Token
-        fields = ('user', 'key', 'write_enabled', 'expires', 'description',)
+        fields = ('user', 'version', 'token', 'write_enabled', 'expires', 'description',)

netbox/users/forms/filtersets.py

@@ -3,10 +3,12 @@ from django.utils.translation import gettext_lazy as _
 
 from netbox.forms import NetBoxModelFilterSetForm
 from netbox.forms.mixins import SavedFiltersMixin
+from users.choices import TokenVersionChoices
 from users.models import Group, ObjectPermission, Token, User
 from utilities.forms import BOOLEAN_WITH_BLANK_CHOICES, FilterForm
 from utilities.forms.fields import DynamicModelMultipleChoiceField
 from utilities.forms.rendering import FieldSet
+from utilities.forms.utils import add_blank_choice
 from utilities.forms.widgets import DateTimePicker
 
 __all__ = (
@@ -29,7 +31,7 @@ class UserFilterForm(NetBoxModelFilterSetForm):
     fieldsets = (
         FieldSet('q', 'filter_id',),
         FieldSet('group_id', name=_('Group')),
-        FieldSet('is_active', 'is_staff', 'is_superuser', name=_('Status')),
+        FieldSet('is_active', 'is_superuser', name=_('Status')),
     )
     group_id = DynamicModelMultipleChoiceField(
         queryset=Group.objects.all(),
@@ -43,13 +45,6 @@ class UserFilterForm(NetBoxModelFilterSetForm):
         ),
         label=_('Is Active'),
     )
-    is_staff = forms.NullBooleanField(
-        required=False,
-        widget=forms.Select(
-            choices=BOOLEAN_WITH_BLANK_CHOICES
-        ),
-        label=_('Is Staff'),
-    )
     is_superuser = forms.NullBooleanField(
         required=False,
         widget=forms.Select(
@@ -117,7 +112,11 @@ class TokenFilterForm(SavedFiltersMixin, FilterForm):
     model = Token
     fieldsets = (
         FieldSet('q', 'filter_id',),
-        FieldSet('user_id', 'write_enabled', 'expires', 'last_used', name=_('Token')),
+        FieldSet('version', 'user_id', 'write_enabled', 'expires', 'last_used', name=_('Token')),
+    )
+    version = forms.ChoiceField(
+        choices=add_blank_choice(TokenVersionChoices),
+        required=False,
     )
     user_id = DynamicModelMultipleChoiceField(
         queryset=User.objects.all(),

netbox/users/forms/model_forms.py

@@ -1,7 +1,6 @@
 import json
 
 from django import forms
-from django.conf import settings
 from django.contrib.auth import password_validation
 from django.contrib.postgres.forms import SimpleArrayField
 from django.core.exceptions import FieldError
@@ -12,14 +11,11 @@ from core.models import ObjectType
 from ipam.formfields import IPNetworkFormField
 from ipam.validators import prefix_validator
 from netbox.preferences import PREFERENCES
+from users.choices import TokenVersionChoices
 from users.constants import *
 from users.models import *
 from utilities.data import flatten_dict
-from utilities.forms.fields import (
-    ContentTypeMultipleChoiceField,
-    DynamicModelMultipleChoiceField,
-    JSONField,
-)
+from utilities.forms.fields import ContentTypeMultipleChoiceField, DynamicModelMultipleChoiceField, JSONField
 from utilities.forms.rendering import FieldSet
 from utilities.forms.widgets import DateTimePicker, SplitMultiSelectWidget
 from utilities.permissions import qs_filter_from_constraints
@@ -64,8 +60,7 @@ class UserConfigFormMetaclass(forms.models.ModelFormMetaclass):
 class UserConfigForm(forms.ModelForm, metaclass=UserConfigFormMetaclass):
     fieldsets = (
         FieldSet(
-            'locale.language', 'pagination.per_page', 'pagination.placement', 'ui.htmx_navigation',
-            'ui.tables.striping',
+            'locale.language', 'pagination.per_page', 'pagination.placement', 'ui.tables.striping',
             name=_('User Interface')
         ),
         FieldSet('data_format', 'csv_delimiter', name=_('Miscellaneous')),
@@ -115,11 +110,11 @@ class UserConfigForm(forms.ModelForm, metaclass=UserConfigFormMetaclass):
 
 
 class UserTokenForm(forms.ModelForm):
-    key = forms.CharField(
-        label=_('Key'),
+    token = forms.CharField(
+        label=_('Token'),
         help_text=_(
-            'Keys must be at least 40 characters in length. <strong>Be sure to record your key</strong> prior to '
-            'submitting this form, as it may no longer be accessible once the token has been created.'
+            'Tokens must be at least 40 characters in length. <strong>Be sure to record your key</strong> prior to '
+            'submitting this form, as it will no longer be accessible once the token has been created.'
         ),
         widget=forms.TextInput(
             attrs={'data-clipboard': 'true'}
@@ -138,7 +133,7 @@ class UserTokenForm(forms.ModelForm):
     class Meta:
         model = Token
         fields = [
-            'key', 'write_enabled', 'expires', 'description', 'allowed_ips',
+            'version', 'token', 'write_enabled', 'expires', 'description', 'allowed_ips',
         ]
         widgets = {
             'expires': DateTimePicker(),
@@ -147,13 +142,24 @@ class UserTokenForm(forms.ModelForm):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-        # Omit the key field if token retrieval is not permitted
-        if self.instance.pk and not settings.ALLOW_TOKEN_RETRIEVAL:
-            del self.fields['key']
+        if self.instance.pk:
+            # Disable the version & user fields for existing Tokens
+            self.fields['version'].disabled = True
+            self.fields['user'].disabled = True
+
+            # Omit the key field when editing an existing Token
+            del self.fields['token']
 
         # Generate an initial random key if none has been specified
-        if not self.instance.pk and not self.initial.get('key'):
-            self.initial['key'] = Token.generate_key()
+        elif self.instance._state.adding and not self.initial.get('token'):
+            self.initial['version'] = TokenVersionChoices.V2
+            self.initial['token'] = Token.generate()
+
+    def save(self, commit=True):
+        if self.instance._state.adding and self.cleaned_data.get('token'):
+            self.instance.token = self.cleaned_data['token']
+
+        return super().save(commit=commit)
 
 
 class TokenForm(UserTokenForm):
@@ -162,14 +168,17 @@ class TokenForm(UserTokenForm):
         label=_('User')
     )
 
-    class Meta:
-        model = Token
+    class Meta(UserTokenForm.Meta):
         fields = [
-            'user', 'key', 'write_enabled', 'expires', 'description', 'allowed_ips',
+            'version', 'token', 'user', 'write_enabled', 'expires', 'description', 'allowed_ips',
         ]
-        widgets = {
-            'expires': DateTimePicker(),
-        }
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        # If not creating a new Token, disable the user field
+        if self.instance and not self.instance._state.adding:
+            self.fields['user'].disabled = True
 
 
 class UserForm(forms.ModelForm):
@@ -198,7 +207,7 @@ class UserForm(forms.ModelForm):
     fieldsets = (
         FieldSet('username', 'password', 'confirm_password', 'first_name', 'last_name', 'email', name=_('User')),
         FieldSet('groups', name=_('Groups')),
-        FieldSet('is_active', 'is_staff', 'is_superuser', name=_('Status')),
+        FieldSet('is_active', 'is_superuser', name=_('Status')),
         FieldSet('object_permissions', name=_('Permissions')),
     )
 
@@ -206,7 +215,7 @@ class UserForm(forms.ModelForm):
         model = User
         fields = [
             'username', 'first_name', 'last_name', 'email', 'groups', 'object_permissions',
-            'is_active', 'is_staff', 'is_superuser',
+            'is_active', 'is_superuser',
         ]
 
     def __init__(self, *args, **kwargs):

netbox/users/graphql/filters.py

@@ -27,7 +27,6 @@ class UserFilter(BaseObjectTypeFilterMixin):
     last_name: FilterLookup[str] | None = strawberry_django.filter_field()
     email: FilterLookup[str] | None = strawberry_django.filter_field()
     is_superuser: FilterLookup[bool] | None = strawberry_django.filter_field()
-    is_staff: FilterLookup[bool] | None = strawberry_django.filter_field()
     is_active: FilterLookup[bool] | None = strawberry_django.filter_field()
     date_joined: DatetimeFilterLookup[datetime] | None = strawberry_django.filter_field()
     last_login: DatetimeFilterLookup[datetime] | None = strawberry_django.filter_field()

netbox/users/graphql/types.py

@@ -25,7 +25,7 @@ class GroupType(BaseObjectType):
 @strawberry_django.type(
     User,
     fields=[
-        'id', 'username', 'first_name', 'last_name', 'email', 'is_staff', 'is_active', 'date_joined', 'groups',
+        'id', 'username', 'first_name', 'last_name', 'email', 'is_active', 'date_joined', 'groups',
     ],
     filters=UserFilter,
     pagination=True

netbox/users/migrations/0013_user_remove_is_staff.py

@@ -0,0 +1,15 @@
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('users', '0012_drop_django_admin_log_table'),
+    ]
+
+    operations = [
+        migrations.RemoveField(
+            model_name='user',
+            name='is_staff',
+        ),
+    ]

netbox/users/migrations/0014_users_token_v2.py

@@ -0,0 +1,100 @@
+import django.core.validators
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('users', '0013_user_remove_is_staff'),
+    ]
+
+    operations = [
+        # Rename the original key field to "plaintext"
+        migrations.RenameField(
+            model_name='token',
+            old_name='key',
+            new_name='plaintext',
+        ),
+        migrations.RunSQL(
+            sql="ALTER INDEX IF EXISTS users_token_key_820deccd_like RENAME TO users_token_plaintext_46c6f315_like",
+        ),
+        migrations.RunSQL(
+            sql="ALTER INDEX IF EXISTS users_token_key_key RENAME TO users_token_plaintext_key",
+        ),
+
+        # Make plaintext (formerly key) nullable for v2 tokens
+        migrations.AlterField(
+            model_name='token',
+            name='plaintext',
+            field=models.CharField(
+                max_length=40,
+                unique=True,
+                blank=True,
+                null=True,
+                validators=[django.core.validators.MinLengthValidator(40)]
+            ),
+        ),
+
+        # Add version field to distinguish v1 and v2 tokens
+        migrations.AddField(
+            model_name='token',
+            name='version',
+            field=models.PositiveSmallIntegerField(default=1),  # Mark all existing Tokens as v1
+            preserve_default=False,
+        ),
+
+        # Change the default version for new tokens to v2
+        migrations.AlterField(
+            model_name='token',
+            name='version',
+            field=models.PositiveSmallIntegerField(default=2),
+        ),
+
+        # Add new key, pepper, and hmac_digest fields for v2 tokens
+        migrations.AddField(
+            model_name='token',
+            name='key',
+            field=models.CharField(
+                blank=True,
+                max_length=12,
+                null=True,
+                unique=True,
+                validators=[django.core.validators.MinLengthValidator(12)]
+            ),
+        ),
+        migrations.AddField(
+            model_name='token',
+            name='pepper_id',
+            field=models.PositiveSmallIntegerField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name='token',
+            name='hmac_digest',
+            field=models.CharField(blank=True, max_length=64, null=True),
+        ),
+
+        # Add constraints to enforce v1/v2-dependent fields
+        migrations.AddConstraint(
+            model_name='token',
+            constraint=models.CheckConstraint(
+                name='enforce_version_dependent_fields',
+                condition=models.Q(
+                    models.Q(
+                        ('hmac_digest__isnull', True),
+                        ('key__isnull', True),
+                        ('pepper_id__isnull', True),
+                        ('plaintext__isnull', False),
+                        ('version', 1)
+                    ),
+                    models.Q(
+                        ('hmac_digest__isnull', False),
+                        ('key__isnull', False),
+                        ('pepper_id__isnull', False),
+                        ('plaintext__isnull', True),
+                        ('version', 2)
+                    ),
+                    _connector='OR'
+                )
+            )
+        ),
+    ]

netbox/users/models/tokens.py

@@ -1,16 +1,22 @@
-import binascii
-import os
+import hashlib
+import hmac
+import random
 
 from django.conf import settings
 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 import Q
 from django.urls import reverse
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 from netaddr import IPNetwork
 
 from ipam.fields import IPNetworkField
+from users.choices import TokenVersionChoices
+from users.constants import TOKEN_CHARSET, TOKEN_DEFAULT_LENGTH, TOKEN_KEY_LENGTH, TOKEN_PREFIX
+from users.utils import get_current_pepper
 from utilities.querysets import RestrictedQuerySet
 
 __all__ = (
@@ -23,11 +29,23 @@ class Token(models.Model):
     An API token used for user authentication. This extends the stock model to allow each user to have multiple tokens.
     It also supports setting an expiration time and toggling write ability.
     """
+    _token = None
+
+    version = models.PositiveSmallIntegerField(
+        verbose_name=_('version'),
+        choices=TokenVersionChoices,
+        default=TokenVersionChoices.V2,
+    )
     user = models.ForeignKey(
         to='users.User',
         on_delete=models.CASCADE,
         related_name='tokens'
     )
+    description = models.CharField(
+        verbose_name=_('description'),
+        max_length=200,
+        blank=True
+    )
     created = models.DateTimeField(
         verbose_name=_('created'),
         auto_now_add=True
@@ -42,21 +60,41 @@ class Token(models.Model):
         blank=True,
         null=True
     )
-    key = models.CharField(
-        verbose_name=_('key'),
-        max_length=40,
-        unique=True,
-        validators=[MinLengthValidator(40)]
-    )
     write_enabled = models.BooleanField(
         verbose_name=_('write enabled'),
         default=True,
         help_text=_('Permit create/update/delete operations using this key')
     )
-    description = models.CharField(
-        verbose_name=_('description'),
-        max_length=200,
-        blank=True
+    # For legacy v1 tokens, this field stores the plaintext 40-char token value. Not used for v2.
+    plaintext = models.CharField(
+        verbose_name=_('plaintext'),
+        max_length=40,
+        unique=True,
+        blank=True,
+        null=True,
+        validators=[MinLengthValidator(40)],
+    )
+    key = models.CharField(
+        verbose_name=_('key'),
+        max_length=TOKEN_KEY_LENGTH,
+        unique=True,
+        blank=True,
+        null=True,
+        validators=[MinLengthValidator(TOKEN_KEY_LENGTH)],
+        help_text=_('v2 token identification key'),
+    )
+    pepper_id = models.PositiveSmallIntegerField(
+        verbose_name=_('pepper ID'),
+        blank=True,
+        null=True,
+        help_text=_('ID of the cryptographic pepper used to hash the token (v2 only)'),
+    )
+    hmac_digest = models.CharField(
+        verbose_name=_('digest'),
+        max_length=64,
+        blank=True,
+        null=True,
+        help_text=_('SHA256 hash of the token and pepper (v2 only)'),
     )
     allowed_ips = ArrayField(
         base_field=IPNetworkField(),
@@ -72,29 +110,113 @@ class Token(models.Model):
     objects = RestrictedQuerySet.as_manager()
 
     class Meta:
+        ordering = ('-created',)
         verbose_name = _('token')
         verbose_name_plural = _('tokens')
-        ordering = ('-created',)
+        constraints = [
+            models.CheckConstraint(
+                name='enforce_version_dependent_fields',
+                condition=(
+                    Q(
+                        version=1,
+                        key__isnull=True,
+                        pepper_id__isnull=True,
+                        hmac_digest__isnull=True,
+                        plaintext__isnull=False
+                    ) |
+                    Q(
+                        version=2,
+                        key__isnull=False,
+                        pepper_id__isnull=False,
+                        hmac_digest__isnull=False,
+                        plaintext__isnull=True
+                    )
+                ),
+            ),
+        ]
+
+    def __init__(self, *args, token=None, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        # This stores the initial plaintext value (if given) on the creation of a new Token. If not provided, a
+        # random token value will be generated and assigned immediately prior to saving the Token instance.
+        self.token = token
 
     def __str__(self):
-        return self.key if settings.ALLOW_TOKEN_RETRIEVAL else self.partial
+        return self.key if self.v2 else self.partial
 
     def get_absolute_url(self):
         return reverse('users:token', args=[self.pk])
 
+    @property
+    def v1(self):
+        return self.version == 1
+
+    @property
+    def v2(self):
+        return self.version == 2
+
     @property
     def partial(self):
-        return f'**********************************{self.key[-6:]}' if self.key else ''
+        """
+        Return a sanitized representation of a v1 token.
+        """
+        return f'**********************************{self.plaintext[-6:]}' if self.plaintext else ''
+
+    @property
+    def token(self):
+        return self._token
+
+    @token.setter
+    def token(self, value):
+        if not self._state.adding:
+            raise ValueError("Cannot assign a new plaintext value for an existing token.")
+        self._token = value
+        if value is not None:
+            if self.v1:
+                self.plaintext = value
+            elif self.v2:
+                self.key = self.key or self.generate_key()
+                self.update_digest()
+
+    def clean(self):
+        if self._state.adding:
+            if self.pepper_id is not None and self.pepper_id not in settings.API_TOKEN_PEPPERS:
+                raise ValidationError(_(
+                    "Invalid pepper ID: {id}. Check configured API_TOKEN_PEPPERS."
+                ).format(id=self.pepper_id))
 
     def save(self, *args, **kwargs):
-        if not self.key:
-            self.key = self.generate_key()
+        # If creating a new Token and no token value has been specified, generate one
+        if self._state.adding and self.token is None:
+            self.token = self.generate()
+
         return super().save(*args, **kwargs)
 
+    @classmethod
+    def generate_key(cls):
+        """
+        Generate and return a random alphanumeric key for v2 tokens.
+        """
+        return cls.generate(length=TOKEN_KEY_LENGTH)
+
     @staticmethod
-    def generate_key():
-        # Generate a random 160-bit key expressed in hexadecimal.
-        return binascii.hexlify(os.urandom(20)).decode()
+    def generate(length=TOKEN_DEFAULT_LENGTH):
+        """
+        Generate and return a random token value of the given length.
+        """
+        return ''.join(random.choice(TOKEN_CHARSET) for _ in range(length))
+
+    def update_digest(self):
+        """
+        Recalculate and save the HMAC digest using the currently defined pepper and token values.
+        """
+        self.pepper_id, pepper = get_current_pepper()
+        self.hmac_digest = hmac.new(
+            pepper.encode('utf-8'),
+            self.token.encode('utf-8'),
+            hashlib.sha256
+        ).hexdigest()
 
     @property
     def is_expired(self):
@@ -102,6 +224,26 @@ class Token(models.Model):
             return False
         return True
 
+    def validate(self, token):
+        """
+        Validate the given plaintext against the token.
+
+        For v1 tokens, check that the given value is equal to the stored plaintext. For v2 tokens, calculate an HMAC
+        from the Token's pepper ID and the given plaintext value, and check whether the result matches the recorded
+        digest.
+        """
+        if self.v1:
+            return token == self.token
+        if self.v2:
+            token = token.removeprefix(TOKEN_PREFIX)
+            try:
+                pepper = settings.API_TOKEN_PEPPERS[self.pepper_id]
+            except KeyError:
+                # Invalid pepper ID
+                return False
+            digest = hmac.new(pepper.encode('utf-8'), token.encode('utf-8'), hashlib.sha256).hexdigest()
+            return digest == self.hmac_digest
+
     def validate_client_ip(self, client_ip):
         """
         Validate the API client IP address against the source IP restrictions (if any) set on the token.

netbox/users/models/users.py

@@ -1,12 +1,16 @@
+from django.contrib.auth.base_user import AbstractBaseUser
 from django.contrib.auth.models import (
-    AbstractUser,
     GroupManager as DjangoGroupManager,
     Permission,
-    UserManager as DjangoUserManager
+    PermissionsMixin,
+    UserManager as DjangoUserManager,
 )
+from django.contrib.auth.validators import UnicodeUsernameValidator
 from django.core.exceptions import ValidationError
+from django.core.mail import send_mail
 from django.db import models
 from django.urls import reverse
+from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 
 from utilities.querysets import RestrictedQuerySet
@@ -68,10 +72,76 @@ class Group(models.Model):
 
 
 class UserManager(DjangoUserManager.from_queryset(RestrictedQuerySet)):
-    pass
+
+    def create_user(self, username, email=None, password=None, **extra_fields):
+        extra_fields.setdefault('is_superuser', False)
+        return self._create_user(username, email, password, **extra_fields)
+
+    create_user.alters_data = True
+
+    async def acreate_user(self, username, email=None, password=None, **extra_fields):
+        extra_fields.setdefault('is_superuser', False)
+        return await self._acreate_user(username, email, password, **extra_fields)
+
+    acreate_user.alters_data = True
+
+    def create_superuser(self, username, email=None, password=None, **extra_fields):
+        extra_fields.setdefault('is_superuser', True)
+
+        if extra_fields.get('is_superuser') is not True:
+            raise ValueError('Superuser must have is_superuser=True.')
+
+        return self._create_user(username, email, password, **extra_fields)
+
+    create_superuser.alters_data = True
+
+    async def acreate_superuser(self, username, email=None, password=None, **extra_fields):
+        extra_fields.setdefault('is_superuser', True)
+
+        if extra_fields.get('is_superuser') is not True:
+            raise ValueError('Superuser must have is_superuser=True.')
+
+        return await self._acreate_user(username, email, password, **extra_fields)
+
+    acreate_superuser.alters_data = True
 
 
-class User(AbstractUser):
+class User(AbstractBaseUser, PermissionsMixin):
+    username = models.CharField(
+        _("username"),
+        max_length=150,
+        unique=True,
+        help_text=_("Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only."),
+        validators=[UnicodeUsernameValidator()],
+        error_messages={
+            "unique": _("A user with that username already exists."),
+        },
+    )
+    first_name = models.CharField(
+        _("first name"),
+        max_length=150,
+        blank=True,
+    )
+    last_name = models.CharField(
+        _("last name"),
+        max_length=150,
+        blank=True,
+    )
+    email = models.EmailField(
+        _("email address"),
+        blank=True,
+    )
+    is_active = models.BooleanField(
+        _("active"),
+        default=True,
+        help_text=_(
+            "Designates whether this user should be treated as active. Unselect this instead of deleting accounts."
+        ),
+    )
+    date_joined = models.DateTimeField(
+        _("date joined"),
+        default=timezone.now,
+    )
     groups = models.ManyToManyField(
         to='users.Group',
         verbose_name=_('groups'),
@@ -87,6 +157,11 @@ class User(AbstractUser):
 
     objects = UserManager()
 
+    # Ensure compatibility with Django's stock User model
+    EMAIL_FIELD = "email"
+    USERNAME_FIELD = "username"
+    REQUIRED_FIELDS = ["email"]
+
     class Meta:
         ordering = ('username',)
         verbose_name = _('user')
@@ -98,7 +173,25 @@ class User(AbstractUser):
     def clean(self):
         super().clean()
 
+        # Normalize email address
+        self.email = self.__class__.objects.normalize_email(self.email)
+
         # Check for any existing Users with names that differ only in case
         model = self._meta.model
         if model.objects.exclude(pk=self.pk).filter(username__iexact=self.username).exists():
             raise ValidationError(_("A user with this username already exists."))
+
+    def get_full_name(self):
+        """
+        Return the first_name plus the last_name, with a space in between.
+        """
+        full_name = "%s %s" % (self.first_name, self.last_name)
+        return full_name.strip()
+
+    def get_short_name(self):
+        """Return the short name for the user."""
+        return self.first_name
+
+    def email_user(self, subject, message, from_email=None, **kwargs):
+        """Send an email to this user."""
+        send_mail(subject, message, from_email, [self.email], **kwargs)

netbox/users/tables.py

@@ -1,7 +1,6 @@
 import django_tables2 as tables
 from django.utils.translation import gettext as _
 
-from account.tables import UserTokenTable
 from netbox.tables import NetBoxTable, columns
 from users.models import Group, ObjectPermission, Token, User
 
@@ -12,18 +11,46 @@ __all__ = (
     'UserTable',
 )
 
+TOKEN = """<samp><a href="{{ record.get_absolute_url }}">{{ record }}</a></samp>"""
 
-class TokenTable(UserTokenTable):
+
+class TokenTable(NetBoxTable):
     user = tables.Column(
         linkify=True,
         verbose_name=_('User')
     )
+    token = columns.TemplateColumn(
+        verbose_name=_('token'),
+        template_code=TOKEN,
+    )
+    write_enabled = columns.BooleanColumn(
+        verbose_name=_('Write Enabled')
+    )
+    created = columns.DateTimeColumn(
+        timespec='minutes',
+        verbose_name=_('Created'),
+    )
+    expires = columns.DateTimeColumn(
+        timespec='minutes',
+        verbose_name=_('Expires'),
+    )
+    last_used = columns.DateTimeColumn(
+        verbose_name=_('Last Used'),
+    )
+    allowed_ips = columns.ArrayColumn(
+        verbose_name=_('Allowed IPs'),
+    )
+    actions = columns.ActionsColumn(
+        actions=('edit', 'delete'),
+    )
 
     class Meta(NetBoxTable.Meta):
         model = Token
         fields = (
-            'pk', 'id', 'key', 'user', 'description', 'write_enabled', 'created', 'expires', 'last_used', 'allowed_ips',
+            'pk', 'id', 'token', 'version', 'pepper_id', 'user', 'description', 'write_enabled', 'created', 'expires',
+            'last_used', 'allowed_ips',
         )
+        default_columns = ('token', 'version', 'user', 'write_enabled', 'description', 'allowed_ips')
 
 
 class UserTable(NetBoxTable):
@@ -38,9 +65,6 @@ class UserTable(NetBoxTable):
     is_active = columns.BooleanColumn(
         verbose_name=_('Is Active'),
     )
-    is_staff = columns.BooleanColumn(
-        verbose_name=_('Is Staff'),
-    )
     is_superuser = columns.BooleanColumn(
         verbose_name=_('Is Superuser'),
     )
@@ -51,8 +75,8 @@ class UserTable(NetBoxTable):
     class Meta(NetBoxTable.Meta):
         model = User
         fields = (
-            'pk', 'id', 'username', 'first_name', 'last_name', 'email', 'groups', 'is_active', 'is_staff',
-            'is_superuser', 'last_login',
+            'pk', 'id', 'username', 'first_name', 'last_name', 'email', 'groups', 'is_active', 'is_superuser',
+            'last_login',
         )
         default_columns = ('pk', 'username', 'first_name', 'last_name', 'email', 'is_active')
 

netbox/users/tests/test_api.py

@@ -2,6 +2,7 @@ from django.test import override_settings
 from django.urls import reverse
 
 from core.models import ObjectType
+from users.constants import TOKEN_DEFAULT_LENGTH
 from users.models import Group, ObjectPermission, Token, User
 from utilities.data import deepmerge
 from utilities.testing import APIViewTestCases, APITestCase, create_test_user
@@ -197,7 +198,7 @@ class TokenTest(
     APIViewTestCases.DeleteObjectViewTestCase
 ):
     model = Token
-    brief_fields = ['description', 'display', 'id', 'key', 'url', 'write_enabled']
+    brief_fields = ['description', 'display', 'id', 'key', 'url', 'version', 'write_enabled']
     bulk_update_data = {
         'description': 'New description',
     }
@@ -211,9 +212,9 @@ class TokenTest(
     @classmethod
     def setUpTestData(cls):
         users = (
-            create_test_user('User1'),
-            create_test_user('User2'),
-            create_test_user('User3'),
+            create_test_user('User 1'),
+            create_test_user('User 2'),
+            create_test_user('User 3'),
         )
 
         tokens = (
@@ -237,6 +238,10 @@ class TokenTest(
             },
         ]
 
+        cls.update_data = {
+            'description': 'Token 1',
+        }
+
     def test_provision_token_valid(self):
         """
         Test the provisioning of a new REST API token given a valid username and password.
@@ -256,8 +261,8 @@ class TokenTest(
 
         response = self.client.post(url, data, format='json', **self.header)
         self.assertEqual(response.status_code, 201)
-        self.assertIn('key', response.data)
-        self.assertEqual(len(response.data['key']), 40)
+        self.assertIn('token', response.data)
+        self.assertEqual(len(response.data['token']), TOKEN_DEFAULT_LENGTH)
         self.assertEqual(response.data['description'], data['description'])
         self.assertEqual(response.data['expires'], data['expires'])
         token = Token.objects.get(user=user)
@@ -299,6 +304,25 @@ class TokenTest(
         response = self.client.post(url, data, format='json', **self.header)
         self.assertEqual(response.status_code, 201)
 
+    def test_reassign_token(self):
+        """
+        Check that a Token cannot be reassigned to another User.
+        """
+        user1 = User.objects.get(username='User 1')
+        user2 = User.objects.get(username='User 2')
+        token1 = Token.objects.filter(user=user1).first()
+        self.add_permissions('users.change_token')
+
+        data = {
+            'user': user2.pk,
+        }
+        url = self._get_detail_url(token1)
+        response = self.client.patch(url, data, format='json', **self.header)
+        # Response should succeed because the read-only `user` field is ignored
+        self.assertEqual(response.status_code, 200)
+        token1.refresh_from_db()
+        self.assertEqual(token1.user, user1, "Token's user should not have changed")
+
 
 class ObjectPermissionTest(
     # No GraphQL support for ObjectPermission

netbox/users/tests/test_filtersets.py

@@ -30,7 +30,6 @@ class UserTestCase(TestCase, BaseFilterSetTests):
                 first_name='Hank',
                 last_name='Hill',
                 email='hank@stricklandpropane.com',
-                is_staff=True,
                 is_superuser=True
             ),
             User(
@@ -98,10 +97,6 @@ class UserTestCase(TestCase, BaseFilterSetTests):
         params = {'is_active': True}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
 
-    def test_is_staff(self):
-        params = {'is_staff': True}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
-
     def test_is_superuser(self):
         params = {'is_superuser': True}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
@@ -271,7 +266,7 @@ class ObjectPermissionTestCase(TestCase, BaseFilterSetTests):
 class TokenTestCase(TestCase, BaseFilterSetTests):
     queryset = Token.objects.all()
     filterset = filtersets.TokenFilterSet
-    ignore_fields = ('allowed_ips',)
+    ignore_fields = ('plaintext', 'hmac_digest', 'allowed_ips')
 
     @classmethod
     def setUpTestData(cls):
@@ -287,21 +282,48 @@ class TokenTestCase(TestCase, BaseFilterSetTests):
         past_date = make_aware(datetime.datetime(2000, 1, 1))
         tokens = (
             Token(
-                user=users[0], key=Token.generate_key(), expires=future_date, write_enabled=True, description='foobar1'
+                version=1,
+                user=users[0],
+                expires=future_date,
+                write_enabled=True,
+                description='foobar1',
             ),
             Token(
-                user=users[1], key=Token.generate_key(), expires=future_date, write_enabled=True, description='foobar2'
+                version=2,
+                user=users[1],
+                expires=future_date,
+                write_enabled=True,
+                description='foobar2',
             ),
             Token(
-                user=users[2], key=Token.generate_key(), expires=past_date, write_enabled=False
+                version=2,
+                user=users[2],
+                expires=past_date,
+                write_enabled=False,
             ),
         )
-        Token.objects.bulk_create(tokens)
+        for token in tokens:
+            token.save()
 
     def test_q(self):
         params = {'q': 'foobar1'}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
 
+    def test_version(self):
+        params = {'version': 1}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
+        params = {'version': 2}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+
+    def test_key(self):
+        tokens = Token.objects.filter(version=2)
+        params = {'key': [tokens[0].key, tokens[1].key]}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+
+    def test_pepper_id(self):
+        params = {'pepper_id': [1]}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+
     def test_user(self):
         users = User.objects.order_by('id')[:2]
         params = {'user_id': [users[0].pk, users[1].pk]}
@@ -317,11 +339,6 @@ class TokenTestCase(TestCase, BaseFilterSetTests):
         params = {'expires__lte': '2021-01-01T00:00:00'}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
 
-    def test_key(self):
-        tokens = Token.objects.all()[:2]
-        params = {'key': [tokens[0].key, tokens[1].key]}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
-
     def test_write_enabled(self):
         params = {'write_enabled': True}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)

netbox/users/tests/test_views.py

@@ -215,6 +215,7 @@ class TokenTestCase(
 ):
     model = Token
     maxDiff = None
+    validation_excluded_fields = ['token', 'user']
 
     @classmethod
     def setUpTestData(cls):
@@ -223,32 +224,34 @@ class TokenTestCase(
             create_test_user('User 2'),
         )
         tokens = (
-            Token(key='123456789012345678901234567890123456789A', user=users[0]),
-            Token(key='123456789012345678901234567890123456789B', user=users[0]),
-            Token(key='123456789012345678901234567890123456789C', user=users[1]),
+            Token(user=users[0]),
+            Token(user=users[0]),
+            Token(user=users[1]),
         )
-        Token.objects.bulk_create(tokens)
+        for token in tokens:
+            token.save()
 
         cls.form_data = {
+            'version': 2,
+            'token': '4F9DAouzURLbicyoG55htImgqQ0b4UZHP5LUYgl5',
             'user': users[0].pk,
-            'key': '1234567890123456789012345678901234567890',
-            'description': 'testdescription',
+            'description': 'Test token',
         }
 
         cls.csv_data = (
-            "key,user,description",
-            f"123456789012345678901234567890123456789D,{users[0].pk},testdescriptionD",
-            f"123456789012345678901234567890123456789E,{users[1].pk},testdescriptionE",
-            f"123456789012345678901234567890123456789F,{users[1].pk},testdescriptionF",
+            "token,user,description",
+            f"zjebxBPzICiPbWz0Wtx0fTL7bCKXKGTYhNzkgC2S,{users[0].pk},Test token",
+            f"9Z5kGtQWba60Vm226dPDfEAV6BhlTr7H5hAXAfbF,{users[1].pk},Test token",
+            f"njpMnNT6r0k0MDccoUhTYYlvP9BvV3qLzYN2p6Uu,{users[1].pk},Test token",
         )
 
         cls.csv_update_data = (
             "id,description",
-            f"{tokens[0].pk},testdescriptionH",
-            f"{tokens[1].pk},testdescriptionI",
-            f"{tokens[2].pk},testdescriptionJ",
+            f"{tokens[0].pk},New description",
+            f"{tokens[1].pk},New description",
+            f"{tokens[2].pk},New description",
         )
 
         cls.bulk_edit_data = {
-            'description': 'newdescription',
+            'description': 'New description',
         }

netbox/users/utils.py

@@ -1,5 +1,11 @@
+from django.conf import settings
 from social_core.storage import NO_ASCII_REGEX, NO_SPECIAL_REGEX
 
+__all__ = (
+    'clean_username',
+    'get_current_pepper',
+)
+
 
 def clean_username(value):
     """Clean username removing any unsupported character"""
@@ -7,3 +13,13 @@ def clean_username(value):
     value = NO_SPECIAL_REGEX.sub('', value)
     value = value.replace(':', '')
     return value
+
+
+def get_current_pepper():
+    """
+    Return the ID and value of the newest (highest ID) cryptographic pepper.
+    """
+    if not settings.API_TOKEN_PEPPERS:
+        raise ValueError("API_TOKEN_PEPPERS is not defined")
+    newest_id = sorted(settings.API_TOKEN_PEPPERS.keys())[-1]
+    return newest_id, settings.API_TOKEN_PEPPERS[newest_id]

netbox/utilities/api.py

@@ -6,6 +6,7 @@ from django.db.models.fields.related import ManyToOneRel, RelatedField
 from django.urls import reverse
 from django.utils.module_loading import import_string
 from django.utils.translation import gettext_lazy as _
+from rest_framework.permissions import BasePermission
 from rest_framework.serializers import Serializer
 from rest_framework.views import get_view_name as drf_get_view_name
 
@@ -16,6 +17,7 @@ from .query import count_related, dict_to_filter_params
 from .string import title
 
 __all__ = (
+    'IsSuperuser',
     'get_annotations_for_serializer',
     'get_graphql_type_for_model',
     'get_prefetches_for_serializer',
@@ -27,6 +29,14 @@ __all__ = (
 )
 
 
+class IsSuperuser(BasePermission):
+    """
+    Allows access only to superusers.
+    """
+    def has_permission(self, request, view):
+        return bool(request.user and request.user.is_superuser)
+
+
 def get_serializer_for_model(model, prefix=''):
     """
     Return the appropriate REST API serializer for the given model.

netbox/utilities/datetime.py

@@ -20,9 +20,4 @@ def datetime_from_timestamp(value):
     """
     Convert an ISO 8601 or RFC 3339 timestamp to a datetime object.
     """
-    # Work around UTC issue for Python < 3.11; see
-    # https://docs.python.org/3/library/datetime.html#datetime.datetime.fromisoformat
-    # TODO: Remove this once Python 3.10 is no longer supported
-    if type(value) is str and value.endswith('Z'):
-        value = f'{value[:-1]}+00:00'
     return datetime.datetime.fromisoformat(value)

netbox/utilities/security.py

@@ -0,0 +1,24 @@
+from django.core.exceptions import ImproperlyConfigured
+
+__all__ = (
+    'validate_peppers',
+)
+
+
+def validate_peppers(peppers):
+    """
+    Validate the given dictionary of cryptographic peppers for type & sufficient length.
+    """
+    if type(peppers) is not dict:
+        raise ImproperlyConfigured("API_TOKEN_PEPPERS must be a dictionary.")
+    for key, pepper in peppers.items():
+        if type(key) is not int:
+            raise ImproperlyConfigured(f"Invalid API_TOKEN_PEPPERS key: {key}. All keys must be integers.")
+        if not 0 <= key <= 32767:
+            raise ImproperlyConfigured(
+                f"Invalid API_TOKEN_PEPPERS key: {key}. Key values must be between 0 and 32767, inclusive."
+            )
+        if type(pepper) is not str:
+            raise ImproperlyConfigured(f"Invalid pepper {key}: Pepper value must be a string.")
+        if len(pepper) < 50:
+            raise ImproperlyConfigured(f"Invalid pepper {key}: Pepper must be at least 50 characters in length.")

netbox/utilities/templates/navigation/menu.html

@@ -2,7 +2,7 @@
 {% load i18n %}
 {% load navigation %}
 
-<ul class="navbar-nav pt-lg-2" {% htmx_boost %}>
+<ul class="navbar-nav pt-lg-2">
   <li class="nav-item d-block d-lg-none">
     <form action="{% url 'search' %}" method="get" autocomplete="off" novalidate>
       <div class="input-group mb-1 mt-2">

netbox/utilities/templatetags/builtins/tags.py

@@ -1,9 +1,8 @@
 import logging
+from urllib.parse import urlparse, urlunparse, parse_qs, urlencode
 
 from django import template
 from django.templatetags.static import static
-from django.utils.safestring import mark_safe
-from urllib.parse import urlparse, urlunparse, parse_qs, urlencode
 
 from extras.choices import CustomFieldTypeChoices
 from utilities.querydict import dict_to_querydict
@@ -126,11 +125,9 @@ def htmx_table(context, viewname, return_url=None, **kwargs):
 @register.simple_tag(takes_context=True)
 def formaction(context):
     """
-    Replace the 'formaction' attribute on an HTML element with the appropriate HTMX attributes
-    if HTMX navigation is enabled (per the user's preferences).
+    A hook for overriding the 'formaction' attribute on an HTML element, for example to replace
+    with 'hx-push-url="true" hx-post' for HTMX navigation.
     """
-    if context.get('htmx_navigation', False):
-        return mark_safe('hx-push-url="true" hx-post')
     return 'formaction'
 
 

netbox/utilities/templatetags/buttons.py

@@ -226,7 +226,6 @@ def bulk_edit_button(context, model, action='bulk_edit', query_params=None):
     return {
         'label': _('Edit Selected'),
         'url': url,
-        'htmx_navigation': context.get('htmx_navigation'),
     }
 
 
@@ -243,5 +242,4 @@ def bulk_delete_button(context, model, action='bulk_delete', query_params=None):
     return {
         'label': _('Delete Selected'),
         'url': url,
-        'htmx_navigation': context.get('htmx_navigation'),
     }

netbox/utilities/templatetags/navigation.py

@@ -1,11 +1,9 @@
 from django import template
-from django.utils.safestring import mark_safe
 
 from netbox.navigation.menu import MENUS
 
 __all__ = (
     'nav',
-    'htmx_boost',
 )
 
 
@@ -30,7 +28,7 @@ def nav(context):
                     continue
                 if not user.has_perms(item.permissions):
                     continue
-                if item.staff_only and not any([user.is_staff, user.is_superuser]):
+                if item.staff_only and not user.is_superuser:
                     continue
                 buttons = [
                     button for button in item.buttons if user.has_perms(button.permissions)
@@ -43,30 +41,4 @@ def nav(context):
 
     return {
         'nav_items': nav_items,
-        'htmx_navigation': context['htmx_navigation']
     }
-
-
-@register.simple_tag(takes_context=True)
-def htmx_boost(context, target='#page-content', select='#page-content'):
-    """
-    Renders the HTML attributes needed to effect HTMX boosting within an element if
-    HTMX navigation is enabled for the request. The target and select parameters are
-    rendered as `hx-target` and `hx-select`, respectively. For example:
-
-        <div id="page-content" {% htmx_boost %}>
-
-    If HTMX navigation is not enabled, the tag renders no content.
-    """
-    if not context.get('htmx_navigation', False):
-        return ''
-    hx_params = {
-        'hx-boost': 'true',
-        'hx-target': target,
-        'hx-select': select,
-        'hx-swap': 'outerHTML show:window:top',
-    }
-    htmx_params = ' '.join([
-        f'{k}="{v}"' for k, v in hx_params.items()
-    ])
-    return mark_safe(htmx_params)

netbox/utilities/testing/api.py

@@ -17,6 +17,7 @@ from core.choices import ObjectChangeActionChoices
 from core.models import ObjectChange, ObjectType
 from ipam.graphql.types import IPAddressFamilyType
 from netbox.models.features import ChangeLoggingMixin
+from users.constants import TOKEN_PREFIX
 from users.models import ObjectPermission, Token, User
 from utilities.api import get_graphql_type_for_model
 from .base import ModelTestCase
@@ -50,7 +51,7 @@ class APITestCase(ModelTestCase):
         self.user = User.objects.create_user(username='testuser')
         self.add_permissions(*self.user_permissions)
         self.token = Token.objects.create(user=self.user)
-        self.header = {'HTTP_AUTHORIZATION': f'Token {self.token.key}'}
+        self.header = {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{self.token.key}.{self.token.token}'}
 
     def _get_view_namespace(self):
         return f'{self.view_namespace or self.model._meta.app_label}-api'
@@ -153,6 +154,7 @@ class APIViewTestCases:
             url = f'{self._get_list_url()}?brief=1'
             response = self.client.get(url, **self.header)
 
+            self.assertHttpStatus(response, status.HTTP_200_OK)
             self.assertEqual(len(response.data['results']), self._get_queryset().count())
             self.assertEqual(sorted(response.data['results'][0]), self.brief_fields)
 

pyproject.toml

@@ -15,9 +15,9 @@ classifiers = [
     "Natural Language :: English",
     "Programming Language :: Python",
     "Programming Language :: Python :: 3 :: Only",
-    "Programming Language :: Python :: 3.10",
-    "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
+    "Programming Language :: Python :: 3.14",
 ]
 
 [project.urls]
@@ -28,7 +28,7 @@ Issues = "https://github.com/netbox-community/netbox/issues"
 
 [tool.black]
 line-length = 120
-target_version = ['py310', 'py311', 'py312']
+target_version = ['py312', 'py313', 'py314']
 skip-string-normalization = true
 
 [tool.isort]

upgrade.sh

@@ -4,7 +4,7 @@
 
 # This script will invoke Python with the value of the PYTHON environment
 # variable (if set), or fall back to "python3". Note that NetBox v4.0+ requires
-# Python 3.10 or later.
+# Python 3.12 or later.
 
 # Parse arguments
 if [[ "$1" == "--readonly" ]]; then
@@ -22,15 +22,15 @@ VIRTUALENV="$(pwd -P)/venv"
 PYTHON="${PYTHON:-python3}"
 
 # Validate the minimum required Python version
-COMMAND="${PYTHON} -c 'import sys; exit(1 if sys.version_info < (3, 10) else 0)'"
+COMMAND="${PYTHON} -c 'import sys; exit(1 if sys.version_info < (3, 12) else 0)'"
 PYTHON_VERSION=$(eval "${PYTHON} -V")
 eval $COMMAND || {
   echo "--------------------------------------------------------------------"
   echo "ERROR: Unsupported Python version: ${PYTHON_VERSION}. NetBox requires"
-  echo "Python 3.10 or later. To specify an alternate Python executable, set"
+  echo "Python 3.12 or later. To specify an alternate Python executable, set"
   echo "the PYTHON environment variable. For example:"
   echo ""
-  echo "  sudo PYTHON=/usr/bin/python3.10 ./upgrade.sh"
+  echo "  sudo PYTHON=/usr/bin/python3.12 ./upgrade.sh"
   echo ""
   echo "To show your current Python version: ${PYTHON} -V"
   echo "--------------------------------------------------------------------"