NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-12-14 04:19:36 -06:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

Merge pull request #20477 from netbox-community/20210-new-token-auth
Some checks failed
CI / build (20.x, 3.12) (push) Has been cancelled
Details
CI / build (20.x, 3.13) (push) Has been cancelled
Details
CodeQL / Analyze (${{ matrix.language }}) (none, actions) (push) Has been cancelled
Details
CodeQL / Analyze (${{ matrix.language }}) (none, javascript-typescript) (push) Has been cancelled
Details
CodeQL / Analyze (${{ matrix.language }}) (none, python) (push) Has been cancelled
Details

Browse Source 
Closes #20210: Implement new version of API token
This commit is contained in:
bctiemann 2025-10-07 11:21:02 -04:00 committed by GitHub
commit 18a308ae3a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
32 changed files with 1041 additions and 384 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

320
contrib/openapi.json
View File

@ -166003,86 +166003,25 @@
"in": "query",
"name": "last_used",
"schema": {
"type": "array",
"items": {
"type": "string",
"format": "date-time"
}
},
"explode": true,
"style": "form"
},
{
"in": "query",
"name": "last_used__empty",
"schema": {
"type": "boolean"
"type": "string",
"format": "date-time"
}
},
{
"in": "query",
"name": "last_used__gt",
"schema": {
"type": "array",
"items": {
"type": "string",
"format": "date-time"
}
},
"explode": true,
"style": "form"
},
{
"in": "query",
"name": "last_used__gte",
"schema": {
"type": "array",
"items": {
"type": "string",
"format": "date-time"
}
},
"explode": true,
"style": "form"
},
{
"in": "query",
"name": "last_used__lt",
"schema": {
"type": "array",
"items": {
"type": "string",
"format": "date-time"
}
},
"explode": true,
"style": "form"
"type": "string",
"format": "date-time"
}
},
{
"in": "query",
"name": "last_used__lte",
"schema": {
"type": "array",
"items": {
"type": "string",
"format": "date-time"
}
},
"explode": true,
"style": "form"
},
{
"in": "query",
"name": "last_used__n",
"schema": {
"type": "array",
"items": {
"type": "string",
"format": "date-time"
}
},
"explode": true,
"style": "form"
"type": "string",
"format": "date-time"
}
},
{
"name": "limit",
@ -166111,6 +166050,91 @@
"type": "string"
}
},
{
"in": "query",
"name": "pepper_id",
"schema": {
"type": "array",
"items": {
"type": "integer",
"format": "int32"
}
},
"explode": true,
"style": "form"
},
{
"in": "query",
"name": "pepper_id__empty",
"schema": {
"type": "boolean"
}
},
{
"in": "query",
"name": "pepper_id__gt",
"schema": {
"type": "array",
"items": {
"type": "integer",
"format": "int32"
}
},
"explode": true,
"style": "form"
},
{
"in": "query",
"name": "pepper_id__gte",
"schema": {
"type": "array",
"items": {
"type": "integer",
"format": "int32"
}
},
"explode": true,
"style": "form"
},
{
"in": "query",
"name": "pepper_id__lt",
"schema": {
"type": "array",
"items": {
"type": "integer",
"format": "int32"
}
},
"explode": true,
"style": "form"
},
{
"in": "query",
"name": "pepper_id__lte",
"schema": {
"type": "array",
"items": {
"type": "integer",
"format": "int32"
}
},
"explode": true,
"style": "form"
},
{
"in": "query",
"name": "pepper_id__n",
"schema": {
"type": "array",
"items": {
"type": "integer",
"format": "int32"
}
},
"explode": true,
"style": "form"
},
{
"in": "query",
"name": "q",
@ -166171,6 +166195,19 @@
"explode": true,
"style": "form"
},
{
"in": "query",
"name": "version",
"schema": {
"type": "integer",
"x-spec-enum-id": "b5df70f0bffd12cb",
"enum": [
1,
2
]
},
"description": "* `1` - v1\n* `2` - v2"
},
{
"in": "query",
"name": "write_enabled",
@ -213892,7 +213929,7 @@
},
"mark_utilized": {
"type": "boolean",
"description": "Report space as 100% utilized"
"description": "Report space as fully utilized"
}
},
"required": [
@ -214001,7 +214038,7 @@
},
"mark_utilized": {
"type": "boolean",
"description": "Report space as 100% utilized"
"description": "Report space as fully utilized"
}
},
"required": [
@ -228068,6 +228105,17 @@
"type": "object",
"description": "Extends the built-in ModelSerializer to enforce calling full_clean() on a copy of the associated instance during\nvalidation. (DRF does not do this by default; see https://github.com/encode/django-rest-framework/issues/3144)",
"properties": {
"version": {
"enum": [
1,
2
],
"type": "integer",
"description": "* `1` - v1\n* `2` - v2",
"x-spec-enum-id": "b5df70f0bffd12cb",
"minimum": 0,
"maximum": 32767
},
"user": {
"oneOf": [
{
@ -228078,6 +228126,10 @@
}
]
},
"description": {
"type": "string",
"maxLength": 200
},
"expires": {
"type": "string",
"format": "date-time",
@ -228088,19 +228140,20 @@
"format": "date-time",
"nullable": true
},
"key": {
"type": "string",
"writeOnly": true,
"maxLength": 40,
"minLength": 40
},
"write_enabled": {
"type": "boolean",
"description": "Permit create/update/delete operations using this key"
},
"description": {
"pepper_id": {
"type": "integer",
"maximum": 32767,
"minimum": 0,
"nullable": true,
"description": "ID of the cryptographic pepper used to hash the token (v2 only)"
},
"token": {
"type": "string",
"maxLength": 200
"minLength": 1
}
}
},
@ -230979,7 +231032,7 @@
},
"mark_utilized": {
"type": "boolean",
"description": "Report space as 100% utilized"
"description": "Report space as fully utilized"
}
}
},
@ -244302,9 +244355,30 @@
"type": "string",
"readOnly": true
},
"version": {
"enum": [
1,
2
],
"type": "integer",
"description": "* `1` - v1\n* `2` - v2",
"x-spec-enum-id": "b5df70f0bffd12cb",
"minimum": 0,
"maximum": 32767
},
"key": {
"type": "string",
"readOnly": true,
"nullable": true,
"description": "v2 token identification key"
},
"user": {
"$ref": "#/components/schemas/BriefUser"
},
"description": {
"type": "string",
"maxLength": 200
},
"created": {
"type": "string",
"format": "date-time",
@ -244324,9 +244398,15 @@
"type": "boolean",
"description": "Permit create/update/delete operations using this key"
},
"description": {
"type": "string",
"maxLength": 200
"pepper_id": {
"type": "integer",
"maximum": 32767,
"minimum": 0,
"nullable": true,
"description": "ID of the cryptographic pepper used to hash the token (v2 only)"
},
"token": {
"type": "string"
}
},
"required": [
@ -244334,6 +244414,7 @@
"display",
"display_url",
"id",
"key",
"url",
"user"
]
@ -244360,6 +244441,17 @@
"type": "string",
"readOnly": true
},
"version": {
"enum": [
1,
2
],
"type": "integer",
"description": "* `1` - v1\n* `2` - v2",
"x-spec-enum-id": "b5df70f0bffd12cb",
"minimum": 0,
"maximum": 32767
},
"user": {
"allOf": [
{
@ -244368,6 +244460,10 @@
],
"readOnly": true
},
"key": {
"type": "string",
"readOnly": true
},
"created": {
"type": "string",
"format": "date-time",
@ -244383,10 +244479,6 @@
"format": "date-time",
"readOnly": true
},
"key": {
"type": "string",
"readOnly": true
},
"write_enabled": {
"type": "boolean",
"description": "Permit create/update/delete operations using this key"
@ -244394,6 +244486,9 @@
"description": {
"type": "string",
"maxLength": 200
},
"token": {
"type": "string"
}
},
"required": [
@ -244411,6 +244506,17 @@
"type": "object",
"description": "Extends the built-in ModelSerializer to enforce calling full_clean() on a copy of the associated instance during\nvalidation. (DRF does not do this by default; see https://github.com/encode/django-rest-framework/issues/3144)",
"properties": {
"version": {
"enum": [
1,
2
],
"type": "integer",
"description": "* `1` - v1\n* `2` - v2",
"x-spec-enum-id": "b5df70f0bffd12cb",
"minimum": 0,
"maximum": 32767
},
"expires": {
"type": "string",
"format": "date-time",
@ -244433,6 +244539,10 @@
"type": "string",
"writeOnly": true,
"minLength": 1
},
"token": {
"type": "string",
"minLength": 1
}
},
"required": [
@ -244444,6 +244554,17 @@
"type": "object",
"description": "Extends the built-in ModelSerializer to enforce calling full_clean() on a copy of the associated instance during\nvalidation. (DRF does not do this by default; see https://github.com/encode/django-rest-framework/issues/3144)",
"properties": {
"version": {
"enum": [
1,
2
],
"type": "integer",
"description": "* `1` - v1\n* `2` - v2",
"x-spec-enum-id": "b5df70f0bffd12cb",
"minimum": 0,
"maximum": 32767
},
"user": {
"oneOf": [
{
@ -244454,6 +244575,10 @@
}
]
},
"description": {
"type": "string",
"maxLength": 200
},
"expires": {
"type": "string",
"format": "date-time",
@ -244464,19 +244589,20 @@
"format": "date-time",
"nullable": true
},
"key": {
"type": "string",
"writeOnly": true,
"maxLength": 40,
"minLength": 40
},
"write_enabled": {
"type": "boolean",
"description": "Permit create/update/delete operations using this key"
},
"description": {
"pepper_id": {
"type": "integer",
"maximum": 32767,
"minimum": 0,
"nullable": true,
"description": "ID of the cryptographic pepper used to hash the token (v2 only)"
},
"token": {
"type": "string",
"maxLength": 200
"minLength": 1
}
},
"required": [
@ -251292,7 +251418,7 @@
},
"mark_utilized": {
"type": "boolean",
"description": "Report space as 100% utilized"
"description": "Report space as fully utilized"
}
},
"required": [
@ -256709,7 +256835,7 @@
"type": "apiKey",
"in": "header",
"name": "Authorization",
"description": "Token-based authentication with required prefix \"Token\""
"description": "`Token <token>` (v1) or `Bearer <key>.<token>` (v2)"
}
}
},

25
docs/configuration/required-parameters.md
View File

@ -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"

2
docs/features/api-integration.md
View File

@ -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"}}'

17
docs/installation/3-netbox.md
View File

@ -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.

29
docs/integrations/rest-api.md
View File

@ -653,18 +653,19 @@ 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.
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.
#### 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 +682,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/
{

57
netbox/account/tables.py
View File

@ -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',
)

8
netbox/account/views.py
View File

@ -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,
})

3
netbox/core/tests/test_api.py
View File

@ -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
@ -136,7 +137,7 @@ class BackgroundTaskTestCase(TestCase):
# Create the test user and assign permissions
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()

116
netbox/netbox/api/authentication.py
View File

@ -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)',
}

10
netbox/netbox/configuration_example.py
View File

@ -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 = {}
#########################
# #

4
netbox/netbox/configuration_testing.py
View File

@ -45,6 +45,10 @@ 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,
'disable_existing_loggers': True

8
netbox/netbox/settings.py
View File

@ -19,6 +19,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
#
@ -65,6 +66,7 @@ elif hasattr(configuration, 'DATABASE') and hasattr(configuration, 'DATABASES'):
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",
@ -215,6 +217,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:

2
netbox/netbox/tables/columns.py
View File

@ -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'):

153
netbox/netbox/tests/test_authentication.py
View File

@ -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):

62
netbox/templates/account/token.html
View File

@ -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 %}

26
netbox/templates/users/token.html
View File

@ -14,9 +14,31 @@
<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>
{% if settings.ALLOW_TOKEN_RETRIEVAL %}
<span id="secret" class="font-monospace" data-secret="{{ object.plaintext }}">{{ object.plaintext }}</span>
<button type="button" class="btn btn-primary toggle-secret float-end" data-bs-toggle="button">{% trans "Show Secret" %}</button>
{% else %}
{{ object.partial }}
{% endif %}
</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>

28
netbox/users/api/serializers_/tokens.py
View File

@ -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,11 @@ 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')
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)
read_only_fields = ('key',)
brief_fields = ('id', 'url', 'display', 'version', 'key', 'write_enabled', 'description')
def validate(self, data):
@ -75,8 +69,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):

17
netbox/users/choices.py Normal file
View File

@ -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')),
]

8
netbox/users/constants.py
View File

@ -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

14
netbox/users/filtersets.py
View File

@ -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)
)

15
netbox/users/forms/bulk_import.py
View File

@ -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
@ -34,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',)

8
netbox/users/forms/filtersets.py
View File

@ -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__ = (
@ -110,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(),

47
netbox/users/forms/model_forms.py
View File

@ -12,14 +12,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
@ -115,10 +112,10 @@ 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 '
'Tokens 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.'
),
widget=forms.TextInput(
@ -138,7 +135,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 +144,27 @@ 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 if token retrieval is not permitted
if self.instance.v1 and settings.ALLOW_TOKEN_RETRIEVAL:
self.initial['token'] = self.instance.plaintext
else:
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 +173,10 @@ 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(),
}
class UserForm(forms.ModelForm):

100
netbox/users/migrations/0014_users_token_v2.py Normal file
View File

@ -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'
)
)
),
]

182
netbox/users/models/tokens.py
View File

@ -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.

40
netbox/users/tables.py
View File

@ -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,53 @@ __all__ = (
'UserTable',
)
TOKEN = """<samp><a href="{{ record.get_absolute_url }}" id="token_{{ record.pk }}">{{ record }}</a></samp>"""
class TokenTable(UserTokenTable):
COPY_BUTTON = """
{% if settings.ALLOW_TOKEN_RETRIEVAL %}
{% copy_content record.pk prefix="token_" color="success" %}
{% endif %}
"""
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'),
extra_buttons=COPY_BUTTON
)
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):

7
netbox/users/tests/test_api.py
View File

@ -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',
}
@ -256,8 +257,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)

42
netbox/users/tests/test_filtersets.py
View File

@ -266,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):
@ -282,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]}
@ -312,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)

31
netbox/users/tests/test_views.py
View File

@ -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',
}

16
netbox/users/utils.py
View File

@ -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]

24
netbox/utilities/security.py Normal file
View File

@ -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.")

4
netbox/utilities/testing/api.py
View File

@ -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)