Compare commits

..

1 Commits

161 changed files with 6813 additions and 7165 deletions

View File

@@ -15,7 +15,7 @@ body:
attributes:
label: NetBox version
description: What version of NetBox are you currently running?
placeholder: v4.4.3
placeholder: v4.4.2
validations:
required: true
- type: dropdown

View File

@@ -27,7 +27,7 @@ body:
attributes:
label: NetBox Version
description: What version of NetBox are you currently running?
placeholder: v4.4.3
placeholder: v4.4.2
validations:
required: true
- type: dropdown
@@ -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

View File

@@ -1,11 +1,3 @@
paths-ignore:
# Ignore compiled JS
- netbox/project-static/dist
query-filters:
# Exclude py/url-redirection: NetBox uses safe_for_redirect() wrapper function
# which validates all redirects via Django's url_has_allowed_host_and_scheme().
# CodeQL's taint tracking doesn't recognize wrapper functions without custom
# query configuration. See #20484.
- exclude:
id: py/url-redirection

View File

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

View File

@@ -12,7 +12,9 @@ django-cors-headers
# Runtime UI tool for debugging Django
# https://github.com/jazzband/django-debug-toolbar/blob/main/docs/changes.rst
django-debug-toolbar
# django-debug-toolbar v6.0.0 raises "Attribute Error at /: 'function' object has no attribute 'set'"
# see https://github.com/netbox-community/netbox/issues/19974
django-debug-toolbar==5.2.0
# Library for writing reusable URL query filters
# https://github.com/carltongibson/django-filter/blob/main/CHANGES.rst
@@ -69,8 +71,7 @@ django-timezone-field
# A REST API framework for Django projects
# https://www.django-rest-framework.org/community/release-notes/
# TODO: Re-evaluate the monkey-patch of get_unique_validators() before upgrading
djangorestframework==3.16.1
djangorestframework
# Sane and flexible OpenAPI 3 schema generation for Django REST framework.
# https://github.com/tfranzel/drf-spectacular/blob/master/CHANGELOG.rst

File diff suppressed because it is too large Load Diff

View File

@@ -2,7 +2,7 @@
## Local Authentication
Local user accounts and groups can be created in NetBox under the "Authentication" section in the "Admin" menu.
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.
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.

View File

@@ -127,3 +127,19 @@ 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` )

View File

@@ -23,31 +23,6 @@ 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"

View File

@@ -1,5 +1,16 @@
# 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"

View File

@@ -131,6 +131,17 @@ 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:

View File

@@ -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.12 or later
* Python 3.10 or later
### 1. Fork the Repo

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: Bearer $TOKEN" \
-H "Authorization: Token $TOKEN" \
-H "Content-Type: application/json" \
http://netbox/api/ipam/prefixes/ \
--data '{"prefix": "192.0.2.0/24", "site": {"name": "Branch 12"}}'

View File

@@ -34,6 +34,9 @@ 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 partialpage navigation for supported views. Disable this preference if unexpected behavior is observed.
### Striped table rows
Toggles alternating row backgrounds on tables.

View File

@@ -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.12 or later required"
NetBox supports only Python 3.12 or later.
!!! warning "Python 3.10 or later required"
NetBox supports Python 3.10, 3.11, and 3.12.
```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.12:
Before continuing, check that your installed Python version is at least 3.10:
```no-highlight
python3 -V
@@ -120,23 +120,6 @@ 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.
@@ -252,10 +235,10 @@ Once NetBox has been configured, we're ready to proceed with the actual installa
sudo /opt/netbox/upgrade.sh
```
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.)
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.)
```no-highlight
sudo PYTHON=/usr/bin/python3.12 /opt/netbox/upgrade.sh
sudo PYTHON=/usr/bin/python3.10 /opt/netbox/upgrade.sh
```
!!! note

View File

@@ -60,3 +60,6 @@ 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.

View File

@@ -121,6 +121,7 @@ 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"
}
@@ -133,6 +134,7 @@ 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
@@ -246,6 +248,7 @@ 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"
}

View File

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

View File

@@ -19,7 +19,7 @@ NetBox requires the following dependencies:
| Dependency | Supported Versions |
|------------|--------------------|
| Python | 3.12, 3.13, 3.14 |
| Python | 3.10, 3.11, 3.12 |
| PostgreSQL | 14+ |
| Redis | 4.0+ |
@@ -27,7 +27,6 @@ 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) |
@@ -131,7 +130,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.12 ./upgrade.sh
sudo PYTHON=/usr/bin/python3.10 ./upgrade.sh
```
!!! note

View File

@@ -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 a 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 an 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,22 +653,18 @@ The NetBox REST API primarily employs token-based authentication. For convenienc
### Tokens
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.
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.
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.
#### v1 and v2 Tokens
!!! 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.
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
### 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.
@@ -685,22 +681,10 @@ It is possible to provision authentication tokens for other users via the REST A
### Authenticating to the API
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:
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:
```
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" \
$ curl -H "Authorization: Token $TOKEN" \
-H "Accept: application/json; indent=4" \
https://netbox/api/dcim/sites/
{

View File

@@ -1,6 +1,6 @@
# Filters & Filter Sets
Filter sets define the mechanisms available for filtering or searching through a set of objects in NetBox. For instance, sites can be filtered by their parent region or group, status, facility ID, and so on. The same filter set is used consistently for a model whether the request is made via the UI or REST API. (Note that the GraphQL API uses a separate filter class.) NetBox employs the [django-filter](https://django-filter.readthedocs.io/en/stable/) library to define filter sets.
Filter sets define the mechanisms available for filtering or searching through a set of objects in NetBox. For instance, sites can be filtered by their parent region or group, status, facility ID, and so on. The same filter set is used consistently for a model whether the request is made via the UI or REST API. (Note that the GraphQL API uses a separate filter class.) NetBox employs the [django-filters2](https://django-tables2.readthedocs.io/en/latest/) library to define filter sets.
## FilterSet Classes

View File

@@ -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.12.0"
requires-python = ">=3.10.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.12/site-packages/netbox.pth
echo /opt/netbox/netbox > $VENV/lib/python3.10/site-packages/netbox.pth
```
## Development Installation

View File

@@ -64,17 +64,14 @@ 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 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.
| 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 |
## Menu Buttons

View File

@@ -1,32 +1,5 @@
# NetBox v4.4
## v4.4.3 (2025-10-14)
### Enhancements
* [#20426](https://github.com/netbox-community/netbox/issues/20426) - Add a copy-to-clipboard button for custom script output
* [#20516](https://github.com/netbox-community/netbox/issues/20516) - Improve rendering of VLAN ID ranges in VLAN group tables
### Bug Fixes
* [#19302](https://github.com/netbox-community/netbox/issues/19302) - Fix uniqueness validation in REST API for nullable fields
* [#19615](https://github.com/netbox-community/netbox/issues/19615) - Fix support for static file parameters in templates when external storage is in use
* [#19818](https://github.com/netbox-community/netbox/issues/19818) - Hide primary IP assignment fields when creating a new virtual machine in the UI
* [#19825](https://github.com/netbox-community/netbox/issues/19825) - Prevent cache for config revisions from being erroneously overwritten when debugging is enabled
* [#20140](https://github.com/netbox-community/netbox/issues/20140) - Changing a site's region or group should update any associated circuit terminations
* [#20156](https://github.com/netbox-community/netbox/issues/20156) - Fix display of rack elevation labels
* [#20290](https://github.com/netbox-community/netbox/issues/20290) - Fix migration error when upgrading to NetBox v4.4 from releases earlier than v4.3
* [#20471](https://github.com/netbox-community/netbox/issues/20471) - Saving an unmodified VLAN group should not generate a change record
* [#20475](https://github.com/netbox-community/netbox/issues/20475) - Collapse singleton VLAN IDs in VLAN group display
* [#20494](https://github.com/netbox-community/netbox/issues/20494) - Correct OpenAPI schema definition for `IntegerRangeSerializer`
* [#20496](https://github.com/netbox-community/netbox/issues/20496) - REST API should always honor `MAX_PAGE_SIZE` value
* [#20497](https://github.com/netbox-community/netbox/issues/20497) - Fix filtering of VLAN groups by VLAN ID range in GraphQL API
* [#20507](https://github.com/netbox-community/netbox/issues/20507) - Fix support for fetching ASN contacts via GraphQL API
* [#20523](https://github.com/netbox-community/netbox/issues/20523) - Hide password change form for users authenticated via SSO
* [#20542](https://github.com/netbox-community/netbox/issues/20542) - Fix the creation of MAC addresses using the "quick add" form
---
## v4.4.2 (2025-09-30)
### Enhancements

57
netbox/account/tables.py Normal file
View File

@@ -0,0 +1,57 @@
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',
)

View File

@@ -26,9 +26,8 @@ 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
from users import forms, tables
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
@@ -329,8 +328,7 @@ class UserTokenListView(LoginRequiredMixin, View):
def get(self, request):
tokens = UserToken.objects.filter(user=request.user)
table = TokenTable(tokens)
table.columns.hide('user')
table = tables.UserTokenTable(tokens)
table.configure(request)
return render(request, 'account/token_list.html', {
@@ -345,9 +343,11 @@ 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,
})

View File

@@ -1,7 +1,5 @@
from django.apps import AppConfig
from netbox import denormalized
class CircuitsConfig(AppConfig):
name = "circuits"
@@ -10,16 +8,6 @@ class CircuitsConfig(AppConfig):
def ready(self):
from netbox.models.features import register_models
from . import signals, search # noqa: F401
from .models import CircuitTermination
# Register models
register_models(*self.get_models())
denormalized.register(CircuitTermination, '_site', {
'_region': 'region',
'_site_group': 'group',
})
denormalized.register(CircuitTermination, '_location', {
'_site': 'site',
})

View File

@@ -282,18 +282,18 @@ class FixSerializedPKRelatedField(OpenApiSerializerFieldExtension):
class FixIntegerRangeSerializerSchema(OpenApiSerializerExtension):
target_class = 'netbox.api.fields.IntegerRangeSerializer'
match_subclasses = True
def map_serializer(self, auto_schema: 'AutoSchema', direction: Direction) -> _SchemaType:
# One range = two integers; many=True will wrap this in an outer array
return {
'type': 'array',
'items': {
'type': 'integer',
'type': 'array',
'items': {
'type': 'integer',
},
'minItems': 2,
'maxItems': 2,
},
'minItems': 2,
'maxItems': 2,
'example': [10, 20],
}

View File

@@ -9,6 +9,7 @@ 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
@@ -23,7 +24,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
@@ -99,7 +100,7 @@ class BaseRQViewSet(viewsets.ViewSet):
"""
Base class for RQ view sets. Provides a list() method. Subclasses must implement get_data().
"""
permission_classes = [IsSuperuser]
permission_classes = [IsAdminUser]
serializer_class = None
def get_data(self):

View File

@@ -3,12 +3,12 @@ from typing import Annotated, List, TYPE_CHECKING
import strawberry
import strawberry_django
from django.contrib.contenttypes.models import ContentType
from strawberry.types import Info
from core.models import ObjectChange
if TYPE_CHECKING:
from core.graphql.types import DataFileType, DataSourceType, ObjectChangeType
from core.graphql.types import DataFileType, DataSourceType
from netbox.core.graphql.types import ObjectChangeType
__all__ = (
'ChangelogMixin',
@@ -20,7 +20,7 @@ __all__ = (
class ChangelogMixin:
@strawberry_django.field
def changelog(self, info: Info) -> List[Annotated['ObjectChangeType', strawberry.lazy('.types')]]: # noqa: F821
def changelog(self, info) -> List[Annotated["ObjectChangeType", strawberry.lazy('.types')]]: # noqa: F821
content_type = ContentType.objects.get_for_model(self)
object_changes = ObjectChange.objects.filter(
changed_object_type=content_type,
@@ -31,5 +31,5 @@ class ChangelogMixin:
@strawberry.type
class SyncedDataMixin:
data_source: Annotated['DataSourceType', strawberry.lazy('core.graphql.types')] | None
data_file: Annotated['DataFileType', strawberry.lazy('core.graphql.types')] | None
data_source: Annotated["DataSourceType", strawberry.lazy('core.graphql.types')] | None
data_file: Annotated["DataFileType", strawberry.lazy('core.graphql.types')] | None

View File

@@ -1,48 +0,0 @@
# Generated by Django 5.2.5 on 2025-09-09 16:48
from django.db import migrations, models
def get_active(apps, schema_editor):
from django.core.cache import cache
ConfigRevision = apps.get_model('core', 'ConfigRevision')
version = None
revision = None
# Try and get the latest version from cache
try:
version = cache.get('config_version')
except Exception:
pass
# If there is a version in cache, attempt to set revision to the current version from cache
# If the version in cache does not exist or there is no version, try the lastest revision in the database
if not version or (version and not (revision := ConfigRevision.objects.filter(pk=version).first())):
revision = ConfigRevision.objects.order_by('-created').first()
# If there is a revision set, set the active revision
if revision:
revision.active = True
revision.save()
class Migration(migrations.Migration):
dependencies = [
('core', '0018_concrete_objecttype'),
]
operations = [
migrations.AddField(
model_name='configrevision',
name='active',
field=models.BooleanField(default=False),
),
migrations.RunPython(code=get_active, reverse_code=migrations.RunPython.noop),
migrations.AddConstraint(
model_name='configrevision',
constraint=models.UniqueConstraint(
condition=models.Q(('active', True)), fields=('active',), name='unique_active_config_revision'
),
),
]

View File

@@ -14,9 +14,6 @@ class ConfigRevision(models.Model):
"""
An atomic revision of NetBox's configuration.
"""
active = models.BooleanField(
default=False
)
created = models.DateTimeField(
verbose_name=_('created'),
auto_now_add=True
@@ -38,13 +35,6 @@ class ConfigRevision(models.Model):
ordering = ['-created']
verbose_name = _('config revision')
verbose_name_plural = _('config revisions')
constraints = [
models.UniqueConstraint(
fields=('active',),
condition=models.Q(active=True),
name='unique_active_config_revision',
)
]
def __str__(self):
if not self.pk:
@@ -69,13 +59,8 @@ class ConfigRevision(models.Model):
"""
cache.set('config', self.data, None)
cache.set('config_version', self.pk, None)
# Set all instances of ConfigRevision to false and set this instance to true
ConfigRevision.objects.all().update(active=False)
ConfigRevision.objects.filter(pk=self.pk).update(active=True)
activate.alters_data = True
@property
def is_active(self):
return self.active
return cache.get('config_version') == self.pk

View File

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

View File

@@ -5,7 +5,7 @@ from django.contrib.contenttypes.models import ContentType
from django.contrib.postgres.fields import ArrayField
from django.contrib.postgres.indexes import GinIndex
from django.core.exceptions import ObjectDoesNotExist
from django.db import connection, models
from django.db import models
from django.db.models import Q
from django.utils.translation import gettext as _
@@ -66,14 +66,6 @@ class ObjectTypeManager(models.Manager):
"""
from netbox.models.features import get_model_features, model_is_public
# TODO: Remove this in NetBox v5.0
# If the ObjectType table has not yet been provisioned (e.g. because we're in a pre-v4.4 migration),
# fall back to ContentType.
if 'core_objecttype' not in connection.introspection.table_names():
ct = ContentType.objects.get_for_model(model, for_concrete_model=for_concrete_model)
ct.features = get_model_features(ct.model_class())
return ct
if not inspect.isclass(model):
model = model.__class__
opts = self._get_opts(model, for_concrete_model)

View File

@@ -8,7 +8,6 @@ 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
@@ -108,14 +107,14 @@ class ObjectTypeTest(APITestCase):
def test_list_objects(self):
object_type_count = ObjectType.objects.count()
response = self.client.get(reverse('core-api:objecttype-list'), **self.header)
response = self.client.get(reverse('extras-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('core-api:objecttype-detail', kwargs={'pk': object_type.pk})
url = reverse('extras-api:objecttype-detail', kwargs={'pk': object_type.pk})
self.assertHttpStatus(self.client.get(url, **self.header), status.HTTP_200_OK)
@@ -135,9 +134,12 @@ 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', is_active=True)
self.user = User.objects.create_user(username='testuser')
self.user.is_staff = True
self.user.is_active = True
self.user.save()
self.token = Token.objects.create(user=self.user)
self.header = {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{self.token.key}.{self.token.token}'}
self.header = {'HTTP_AUTHORIZATION': f'Token {self.token.key}'}
# Clear all queues prior to running each test
get_queue('default').connection.flushall()
@@ -148,11 +150,13 @@ 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_superuser = True
self.user.is_staff = True
self.user.save()
response = self.client.get(url, **self.header)
self.assertEqual(response.status_code, 200)
@@ -161,16 +165,7 @@ class BackgroundTaskTestCase(TestCase):
self.assertIn('low', str(response.content))
def test_background_queue(self):
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)
response = self.client.get(reverse('core-api:rqqueue-detail', args=['default']), **self.header)
self.assertEqual(response.status_code, 200)
self.assertIn('default', str(response.content))
self.assertIn('oldest_job_timestamp', str(response.content))
@@ -179,16 +174,8 @@ 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')
# 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)
response = self.client.get(reverse('core-api:rqtask-list'), **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))
@@ -196,16 +183,8 @@ 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])
# 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)
response = self.client.get(reverse('core-api:rqtask-detail', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
self.assertIn(str(job.id), str(response.content))
self.assertIn('origin', str(response.content))
@@ -215,65 +194,45 @@ 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])
# 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)
response = self.client.post(reverse('core-api:rqtask-delete', args=[job.id]), **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):
# Enqueue & run a job that will fail
queue = get_queue('default')
# Enqueue & run a job that will fail
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
self.user.is_superuser = True
self.user.save()
response = self.client.post(url, **self.header)
response = self.client.post(reverse('core-api:rqtask-requeue', args=[job.id]), **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):
# Enqueue some jobs that each depends on its predecessor
queue = get_queue('default')
# Enqueue some jobs that each depends on its predecessor
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
self.user.is_superuser = True
self.user.save()
response = self.client.post(url, **self.header)
response = self.client.post(reverse('core-api:rqtask-enqueue', args=[job.id]), **self.header)
self.assertEqual(response.status_code, 200)
# Check that job's status is updated correctly
@@ -283,27 +242,19 @@ 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)
# 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)
response = self.client.post(reverse('core-api:rqtask-stop', args=[job.id]), **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)
@@ -311,34 +262,19 @@ 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')
# 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)
response = self.client.get(reverse('core-api:rqworker-list'), **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])
# 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)
response = self.client.get(reverse('core-api:rqworker-detail', args=[worker1.name]), **self.header)
self.assertEqual(response.status_code, 200)
self.assertIn(str(worker1.name), str(response.content))
self.assertIn('birth_date', str(response.content))

View File

@@ -158,7 +158,7 @@ class BackgroundTaskTestCase(TestCase):
def setUp(self):
super().setUp()
self.user.is_superuser = True
self.user.is_staff = 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_superuser = False
self.user.is_staff = False
self.user.save()
response = self.client.get(url)
self.assertEqual(response.status_code, 403)
# Load view with permission
self.user.is_superuser = True
self.user.is_staff = 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_superuser = True
self.user.is_staff = True
self.user.save()
def test_system_view_default(self):

View File

@@ -372,7 +372,7 @@ class ConfigRevisionRestoreView(ContentTypePermissionRequiredMixin, View):
class BaseRQView(UserPassesTestMixin, View):
def test_func(self):
return self.request.user.is_superuser
return self.request.user.is_staff
class BackgroundQueueListView(TableMixin, BaseRQView):
@@ -555,7 +555,7 @@ class WorkerView(BaseRQView):
class SystemView(UserPassesTestMixin, View):
def test_func(self):
return self.request.user.is_superuser
return self.request.user.is_staff
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_superuser
return self.request.user.is_staff
def get_cached_plugins(self, request):
catalog_plugins = {}

View File

@@ -1,8 +1,10 @@
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
@@ -49,11 +51,9 @@ class TracedCableSerializer(BaseModelSerializer):
class CableTerminationSerializer(NetBoxModelSerializer):
termination_type = ContentTypeField(
read_only=True,
)
termination = serializers.SerializerMethodField(
read_only=True,
queryset=ContentType.objects.filter(CABLE_TERMINATION_MODELS)
)
termination = serializers.SerializerMethodField(read_only=True)
class Meta:
model = CableTermination
@@ -61,8 +61,6 @@ 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):

View File

@@ -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, NetBoxReadOnlyModelViewSet
from netbox.api.viewsets import NetBoxModelViewSet, MPTTLockedMixin
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(NetBoxReadOnlyModelViewSet):
class CableTerminationViewSet(NetBoxModelViewSet):
metadata_class = ContentTypeMetadata
queryset = CableTermination.objects.all()
serializer_class = serializers.CableTerminationSerializer

View File

@@ -1,5 +1,3 @@
from strawberry.types import Info
from circuits.graphql.types import CircuitTerminationType, ProviderNetworkType
from circuits.models import CircuitTermination, ProviderNetwork
from dcim.graphql.types import (
@@ -51,7 +49,7 @@ class InventoryItemTemplateComponentType:
)
@classmethod
def resolve_type(cls, instance, info: Info):
def resolve_type(cls, instance, info):
if type(instance) is ConsolePortTemplate:
return ConsolePortTemplateType
if type(instance) is ConsoleServerPortTemplate:
@@ -81,7 +79,7 @@ class InventoryItemComponentType:
)
@classmethod
def resolve_type(cls, instance, info: Info):
def resolve_type(cls, instance, info):
if type(instance) is ConsolePort:
return ConsolePortType
if type(instance) is ConsoleServerPort:
@@ -114,7 +112,7 @@ class ConnectedEndpointType:
)
@classmethod
def resolve_type(cls, instance, info: Info):
def resolve_type(cls, instance, info):
if type(instance) is CircuitTermination:
return CircuitTerminationType
if type(instance) is ConsolePortType:

View File

@@ -618,6 +618,12 @@ class BaseInterface(models.Model):
null=True,
verbose_name=_('primary MAC address')
)
mac_addresses = GenericRelation(
to='dcim.MACAddress',
content_type_field='assigned_object_type',
object_id_field='assigned_object_id',
related_query_name='interface',
)
class Meta:
abstract = True

View File

@@ -196,7 +196,7 @@ class DeviceTable(TenancyColumnsMixin, ContactsColumnMixin, NetBoxTable):
verbose_name=_('Type')
)
u_height = columns.TemplateColumn(
accessor=tables.A('device_type__u_height'),
accessor=tables.A('device_type.u_height'),
verbose_name=_('U Height'),
template_code='{{ value|floatformat }}'
)

View File

@@ -2376,33 +2376,6 @@ 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

View File

@@ -7,14 +7,13 @@ from django.test import override_settings, tag
from django.urls import reverse
from netaddr import EUI
from core.models import ObjectType
from dcim.choices import *
from dcim.constants import *
from dcim.models import *
from ipam.models import ASN, RIR, VLAN, VRF
from netbox.choices import CSVDelimiterChoices, ImportFormatChoices, WeightUnitChoices
from tenancy.models import Tenant
from users.models import ObjectPermission, User
from users.models import User
from utilities.testing import ViewTestCases, create_tags, create_test_device, post_data
from wireless.models import WirelessLAN
@@ -3729,29 +3728,3 @@ class MACAddressTestCase(ViewTestCases.PrimaryObjectViewTestCase):
cls.bulk_edit_data = {
'description': 'New description',
}
@tag('regression') # Issue #20542
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
def test_create_macaddress_via_quickadd(self):
"""
Test creating a MAC address via quick-add modal (e.g., from Interface form).
Regression test for issue #20542 where form prefix was missing in POST handler.
"""
obj_perm = ObjectPermission(name='Test permission', actions=['add'])
obj_perm.save()
obj_perm.users.add(self.user)
obj_perm.object_types.add(ObjectType.objects.get_for_model(self.model))
# Simulate quick-add form submission with 'quickadd-' prefix
formatted_data = post_data(self.form_data)
quickadd_data = {f'quickadd-{k}': v for k, v in formatted_data.items()}
quickadd_data['_quickadd'] = 'True'
initial_count = self._get_queryset().count()
url = f"{self._get_url('add')}?_quickadd=True&target=id_primary_mac_address"
response = self.client.post(url, data=quickadd_data)
# Should successfully create the MAC address and return the quick_add_created template
self.assertHttpStatus(response, 200)
self.assertIn(b'quick-add-object', response.content)
self.assertEqual(initial_count + 1, self._get_queryset().count())

View File

@@ -1,8 +1,10 @@
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
@@ -27,6 +29,9 @@ 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'),

View File

@@ -95,11 +95,7 @@ def process_event_rules(event_rules, object_type, event_type, data, username=Non
continue
# Compile event data
if event_rule.action_type == EventRuleActionChoices.SCRIPT:
event_data = {}
else:
event_data = event_rule.action_data or {}
event_data = event_rule.action_data or {}
event_data.update(data)
# Webhooks
@@ -135,20 +131,6 @@ 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

View File

@@ -2,7 +2,6 @@ from typing import TYPE_CHECKING, Annotated, List
import strawberry
import strawberry_django
from strawberry.types import Info
__all__ = (
'ConfigContextMixin',
@@ -38,7 +37,7 @@ class CustomFieldsMixin:
class ImageAttachmentsMixin:
@strawberry_django.field
def image_attachments(self, info: Info) -> List[Annotated['ImageAttachmentType', strawberry.lazy('.types')]]:
def image_attachments(self, info) -> List[Annotated["ImageAttachmentType", strawberry.lazy('.types')]]:
return self.images.restrict(info.context.request.user, 'view')
@@ -46,17 +45,17 @@ class ImageAttachmentsMixin:
class JournalEntriesMixin:
@strawberry_django.field
def journal_entries(self, info: Info) -> List[Annotated['JournalEntryType', strawberry.lazy('.types')]]:
def journal_entries(self, info) -> List[Annotated["JournalEntryType", strawberry.lazy('.types')]]:
return self.journal_entries.all()
@strawberry.type
class TagsMixin:
tags: List[Annotated['TagType', strawberry.lazy('.types')]]
tags: List[Annotated["TagType", strawberry.lazy('.types')]]
@strawberry.type
class ContactsMixin:
contacts: List[Annotated['ContactAssignmentType', strawberry.lazy('tenancy.graphql.types')]]
contacts: List[Annotated["ContactAssignmentType", strawberry.lazy('tenancy.graphql.types')]]

View File

@@ -1,39 +1,9 @@
from django.contrib.postgres.fields import ArrayField
from django.contrib.postgres.fields.ranges import RangeField
from django.db.models import CharField, JSONField, Lookup
from django.db.models.fields.json import KeyTextTransform
from .fields import CachedValueField
class RangeContains(Lookup):
"""
Filter ArrayField(RangeField) columns where ANY element-range contains the scalar RHS.
Usage (ORM):
Model.objects.filter(<range_array_field>__range_contains=<scalar>)
Works with int4range[], int8range[], daterange[], tstzrange[], etc.
"""
lookup_name = 'range_contains'
def as_sql(self, compiler, connection):
# Compile LHS (the array-of-ranges column/expression) and RHS (scalar)
lhs, lhs_params = self.process_lhs(compiler, connection)
rhs, rhs_params = self.process_rhs(compiler, connection)
# Guard: only allow ArrayField whose base_field is a PostgreSQL RangeField
field = getattr(self.lhs, 'output_field', None)
if not (isinstance(field, ArrayField) and isinstance(field.base_field, RangeField)):
raise TypeError('range_contains is only valid for ArrayField(RangeField) columns')
# Range-contains-element using EXISTS + UNNEST keeps the range on the LHS: r @> value
sql = f"EXISTS (SELECT 1 FROM unnest({lhs}) AS r WHERE r @> {rhs})"
params = lhs_params + rhs_params
return sql, params
class Empty(Lookup):
"""
Filter on whether a string is empty.
@@ -55,7 +25,7 @@ class JSONEmpty(Lookup):
A key is considered empty if it is "", null, or does not exist.
"""
lookup_name = 'empty'
lookup_name = "empty"
def as_sql(self, compiler, connection):
# self.lhs.lhs is the parent expression (could be a JSONField or another KeyTransform)
@@ -99,7 +69,6 @@ class NetContainsOrEquals(Lookup):
return 'CAST(%s AS INET) >>= %s' % (lhs, rhs), params
ArrayField.register_lookup(RangeContains)
CharField.register_lookup(Empty)
JSONField.register_lookup(JSONEmpty)
CachedValueField.register_lookup(NetHost)

View File

@@ -90,7 +90,7 @@ class ConfigContextModelQuerySet(RestrictedQuerySet):
ConfigContext.objects.filter(
self._get_config_context_filters()
).annotate(
_data=EmptyGroupByJSONBAgg('data', order_by=['weight', 'name'])
_data=EmptyGroupByJSONBAgg('data', ordering=['weight', 'name'])
).values("_data").order_by()
)
)

View File

@@ -1,9 +1,12 @@
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
@@ -485,7 +488,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
@@ -577,6 +580,40 @@ 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
#

View File

@@ -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['object_type'], 'dcim.site')
self.assertEqual(body['model'], 'site')
self.assertEqual(body['username'], 'testuser')
self.assertEqual(body['request_id'], str(request_id))
self.assertEqual(body['data']['name'], 'Site 1')

View File

@@ -1,3 +1,5 @@
import logging
import tempfile
from datetime import date, datetime, timezone
from decimal import Decimal
@@ -7,6 +9,7 @@ from netaddr import IPAddress, IPNetwork
from dcim.models import DeviceRole
from extras.scripts import *
from utilities.testing import disable_logging
CHOICES = (
('ff0000', 'Red'),
@@ -32,6 +35,35 @@ 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):

View File

@@ -52,6 +52,7 @@ 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,
@@ -99,7 +100,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['object_type']} {context['event']})"
f"Sending {params['method']} request to {params['url']} ({context['model']} {context['event']})"
)
logger.debug(params)
try:

View File

@@ -908,8 +908,7 @@ class VLANGroupFilterSet(OrganizationalModelFilterSet, TenancyFilterSet):
method='filter_scope'
)
contains_vid = django_filters.NumberFilter(
field_name='vid_ranges',
lookup_expr='range_contains',
method='filter_contains_vid'
)
class Meta:
@@ -932,6 +931,21 @@ class VLANGroupFilterSet(OrganizationalModelFilterSet, TenancyFilterSet):
scope_id=value
)
def filter_contains_vid(self, queryset, name, value):
"""
Return all VLANGroups which contain the given VLAN ID.
"""
table_name = VLANGroup._meta.db_table
# TODO: See if this can be optimized without compromising queryset integrity
# Expand VLAN ID ranges to query by integer
groups = VLANGroup.objects.raw(
f'SELECT id FROM {table_name}, unnest(vid_ranges) vid_range WHERE %s <@ vid_range',
params=(value,)
)
return queryset.filter(
pk__in=[g.id for g in groups]
)
class VLANFilterSet(NetBoxModelFilterSet, TenancyFilterSet):
region_id = TreeNodeMultipleChoiceFilter(

View File

@@ -19,7 +19,7 @@ from tenancy.graphql.filter_mixins import ContactFilterMixin, TenancyFilterMixin
from virtualization.models import VMInterface
if TYPE_CHECKING:
from netbox.graphql.filter_lookups import IntegerLookup, IntegerRangeArrayLookup
from netbox.graphql.filter_lookups import IntegerArrayLookup, IntegerLookup
from circuits.graphql.filters import ProviderFilter
from core.graphql.filters import ContentTypeFilter
from dcim.graphql.filters import SiteFilter
@@ -340,7 +340,7 @@ class VLANFilter(TenancyFilterMixin, PrimaryModelFilterMixin):
@strawberry_django.filter_type(models.VLANGroup, lookups=True)
class VLANGroupFilter(ScopedFilterMixin, OrganizationalModelFilterMixin):
vid_ranges: Annotated['IntegerRangeArrayLookup', strawberry.lazy('netbox.graphql.filter_lookups')] | None = (
vid_ranges: Annotated['IntegerArrayLookup', strawberry.lazy('netbox.graphql.filter_lookups')] | None = (
strawberry_django.filter_field()
)

View File

@@ -74,7 +74,7 @@ class BaseIPAddressFamilyType:
filters=ASNFilter,
pagination=True
)
class ASNType(NetBoxObjectType, ContactsMixin):
class ASNType(NetBoxObjectType):
asn: BigInt
rir: Annotated["RIRType", strawberry.lazy('ipam.graphql.types')] | None
tenant: Annotated["TenantType", strawberry.lazy('tenancy.graphql.types')] | None

View File

@@ -10,9 +10,9 @@ from django.utils.translation import gettext_lazy as _
from dcim.models import Interface, Site, SiteGroup
from ipam.choices import *
from ipam.constants import *
from ipam.querysets import VLANGroupQuerySet, VLANQuerySet
from ipam.querysets import VLANQuerySet, VLANGroupQuerySet
from netbox.models import OrganizationalModel, PrimaryModel, NetBoxModel
from utilities.data import check_ranges_overlap, ranges_to_string, ranges_to_string_list
from utilities.data import check_ranges_overlap, ranges_to_string
from virtualization.models import VMInterface
__all__ = (
@@ -164,18 +164,8 @@ class VLANGroup(OrganizationalModel):
"""
return VLAN.objects.filter(group=self).order_by('vid')
@property
def vid_ranges_items(self):
"""
Property that converts VID ranges to a list of string representations.
"""
return ranges_to_string_list(self.vid_ranges)
@property
def vid_ranges_list(self):
"""
Property that converts VID ranges into a string representation.
"""
return ranges_to_string(self.vid_ranges)

View File

@@ -41,8 +41,7 @@ class VLANGroupTable(TenancyColumnsMixin, NetBoxTable):
linkify=True,
orderable=False
)
vid_ranges_list = columns.ArrayColumn(
accessor='vid_ranges_items',
vid_ranges_list = tables.Column(
verbose_name=_('VID Ranges'),
orderable=False
)

View File

@@ -1723,10 +1723,6 @@ class VLANGroupTestCase(TestCase, ChangeLoggedFilterSetTests):
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
params = {'contains_vid': 1}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 8)
params = {'contains_vid': 12} # 11 is NOT in [1,11)
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
params = {'contains_vid': 4095}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 0)
def test_region(self):
params = {'region': Region.objects.first().pk}

View File

@@ -1,66 +0,0 @@
from django.test import TestCase
from django.db.backends.postgresql.psycopg_any import NumericRange
from ipam.models import VLANGroup
class VLANGroupRangeContainsLookupTests(TestCase):
@classmethod
def setUpTestData(cls):
# Two ranges: [1,11) and [20,31)
cls.g1 = VLANGroup.objects.create(
name='VlanGroup-A',
slug='VlanGroup-A',
vid_ranges=[NumericRange(1, 11), NumericRange(20, 31)],
)
# One range: [100,201)
cls.g2 = VLANGroup.objects.create(
name='VlanGroup-B',
slug='VlanGroup-B',
vid_ranges=[NumericRange(100, 201)],
)
cls.g_empty = VLANGroup.objects.create(
name='VlanGroup-empty',
slug='VlanGroup-empty',
vid_ranges=[],
)
def test_contains_value_in_first_range(self):
"""
Tests whether a specific value is contained within the first range in a queried
set of VLANGroup objects.
"""
names = list(
VLANGroup.objects.filter(vid_ranges__range_contains=10).values_list('name', flat=True).order_by('name')
)
self.assertEqual(names, ['VlanGroup-A'])
def test_contains_value_in_second_range(self):
"""
Tests if a value exists in the second range of VLANGroup objects and
validates the result against the expected list of names.
"""
names = list(
VLANGroup.objects.filter(vid_ranges__range_contains=25).values_list('name', flat=True).order_by('name')
)
self.assertEqual(names, ['VlanGroup-A'])
def test_upper_bound_is_exclusive(self):
"""
Tests if the upper bound of the range is exclusive in the filter method.
"""
# 11 is NOT in [1,11)
self.assertFalse(VLANGroup.objects.filter(vid_ranges__range_contains=11).exists())
def test_no_match_far_outside(self):
"""
Tests that no VLANGroup contains a VID within a specified range far outside
common VID bounds and returns `False`.
"""
self.assertFalse(VLANGroup.objects.filter(vid_ranges__range_contains=4095).exists())
def test_empty_array_never_matches(self):
"""
Tests the behavior of VLANGroup objects when an empty array is used to match a
specific condition.
"""
self.assertFalse(VLANGroup.objects.filter(pk=self.g_empty.pk, vid_ranges__range_contains=1).exists())

View File

@@ -2,90 +2,47 @@ import logging
from django.conf import settings
from django.utils import timezone
from drf_spectacular.extensions import OpenApiAuthenticationExtension
from rest_framework import exceptions
from rest_framework.authentication import BaseAuthentication, get_authorization_header
from rest_framework import authentication, exceptions
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(BaseAuthentication):
class TokenAuthentication(authentication.TokenAuthentication):
"""
A custom authentication scheme which enforces Token expiration times and source IP restrictions.
"""
model = Token
def authenticate(self, request):
# 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
result = super().authenticate(request)
if result:
token = result[1]
# Enforce source IP restrictions (if any) set on the token
if token.allowed_ips:
client_ip = get_client_ip(request)
if client_ip is None:
raise exceptions.AuthenticationFailed(
"Client IP address could not be determined for validation. Check that the HTTP server is "
"correctly configured to pass the required header(s)."
)
if not token.validate_client_ip(client_ip):
raise exceptions.AuthenticationFailed(
f"Source IP {client_ip} is not permitted to authenticate using this token."
)
return result
def authenticate_credentials(self, key):
model = self.get_model()
try:
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")
token = model.objects.prefetch_related('user').get(key=key)
except model.DoesNotExist:
raise exceptions.AuthenticationFailed("Invalid token")
# 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:
@@ -97,8 +54,11 @@ class TokenAuthentication(BaseAuthentication):
else:
Token.objects.filter(pk=token.pk).update(last_used=timezone.now())
user = token.user
# 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
@@ -172,17 +132,3 @@ 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)',
}

View File

@@ -169,7 +169,7 @@ class IntegerRangeSerializer(serializers.Serializer):
if type(data[0]) is not int or type(data[1]) is not int:
raise ValidationError(_("Range boundaries must be defined as integers."))
return NumericRange(data[0], data[1] + 1, bounds='[)')
return NumericRange(data[0], data[1], bounds='[]')
def to_representation(self, instance):
return instance.lower, instance.upper - 1

View File

@@ -44,28 +44,22 @@ class OptionalLimitOffsetPagination(LimitOffsetPagination):
return list(queryset[self.offset:])
def get_limit(self, request):
max_limit = self.default_limit
MAX_PAGE_SIZE = get_config().MAX_PAGE_SIZE
if MAX_PAGE_SIZE:
max_limit = min(max_limit, MAX_PAGE_SIZE)
if self.limit_query_param:
MAX_PAGE_SIZE = get_config().MAX_PAGE_SIZE
if MAX_PAGE_SIZE:
MAX_PAGE_SIZE = max(MAX_PAGE_SIZE, self.default_limit)
try:
limit = int(request.query_params[self.limit_query_param])
if limit < 0:
raise ValueError()
# Enforce maximum page size, if defined
if MAX_PAGE_SIZE:
if limit == 0:
max_limit = MAX_PAGE_SIZE
else:
max_limit = min(MAX_PAGE_SIZE, limit)
else:
max_limit = limit
return MAX_PAGE_SIZE if limit == 0 else min(limit, MAX_PAGE_SIZE)
return limit
except (KeyError, ValueError):
pass
return max_limit
return self.default_limit
def get_queryset_count(self, queryset):
return queryset.count()

View File

@@ -184,13 +184,14 @@ 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
@@ -250,8 +251,19 @@ class RemoteUserBackend(_RemoteUserBackend):
return bool(result)
def _is_staff(self, user):
# Retain for pre-v4.5 compatibility
return user.is_superuser
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)
def configure_user(self, request, user):
logger = logging.getLogger('netbox.auth.RemoteUserBackend')

View File

@@ -78,16 +78,11 @@ class Config:
from core.models import ConfigRevision
try:
# Enforce the creation date as the ordering parameter
revision = ConfigRevision.objects.get(active=True)
logger.debug(f"Loaded active configuration revision #{revision.pk}")
except (ConfigRevision.DoesNotExist, ConfigRevision.MultipleObjectsReturned):
logger.warning("No active configuration revision found - falling back to most recent")
revision = ConfigRevision.objects.order_by('-created').first()
revision = ConfigRevision.objects.last()
if revision is None:
logger.debug("No previous configuration found in database; proceeding with default values")
return
logger.debug(f"Using fallback configuration revision #{revision.pk}")
logger.debug("Loaded configuration data from database")
except DatabaseError:
# The database may not be available yet (e.g. when running a management command)
logger.warning("Skipping config initialization (database unavailable)")

View File

@@ -68,16 +68,6 @@ 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 = {}
#########################
# #
@@ -91,6 +81,9 @@ 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 = [

View File

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

View File

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

View File

@@ -7,7 +7,6 @@ from django.core.exceptions import FieldDoesNotExist
from django.db.models import Q, QuerySet
from django.db.models.fields.related import ForeignKey, ManyToManyField, ManyToManyRel, ManyToOneRel
from strawberry import ID
from strawberry.directive import DirectiveValue
from strawberry.types import Info
from strawberry_django import (
ComparisonFilterLookup,
@@ -25,7 +24,6 @@ __all__ = (
'FloatLookup',
'IntegerArrayLookup',
'IntegerLookup',
'IntegerRangeArrayLookup',
'JSONFilter',
'StringArrayLookup',
'TreeNodeFilter',
@@ -69,7 +67,7 @@ class IntegerLookup:
return None
@strawberry_django.filter_field
def filter(self, info: Info, queryset: QuerySet, prefix: DirectiveValue[str] = '') -> Tuple[QuerySet, Q]:
def filter(self, info: Info, queryset: QuerySet, prefix: str = '') -> Tuple[QuerySet, Q]:
filters = self.get_filter()
if not filters:
@@ -92,7 +90,7 @@ class FloatLookup:
return None
@strawberry_django.filter_field
def filter(self, info: Info, queryset: QuerySet, prefix: DirectiveValue[str] = '') -> Tuple[QuerySet, Q]:
def filter(self, info: Info, queryset: QuerySet, prefix: str = '') -> Tuple[QuerySet, Q]:
filters = self.get_filter()
if not filters:
@@ -111,7 +109,7 @@ class JSONFilter:
lookup: JSONLookup
@strawberry_django.filter_field
def filter(self, info: Info, queryset: QuerySet, prefix: DirectiveValue[str] = '') -> Tuple[QuerySet, Q]:
def filter(self, info: Info, queryset: QuerySet, prefix: str = '') -> Tuple[QuerySet, Q]:
filters = self.lookup.get_filter()
if not filters:
@@ -138,7 +136,7 @@ class TreeNodeFilter:
match_type: TreeNodeMatch
@strawberry_django.filter_field
def filter(self, info: Info, queryset: QuerySet, prefix: DirectiveValue[str] = '') -> Tuple[QuerySet, Q]:
def filter(self, info: Info, queryset: QuerySet, prefix: str = '') -> Tuple[QuerySet, Q]:
model_field_name = prefix.removesuffix('__').removesuffix('_id')
model_field = None
try:
@@ -219,30 +217,3 @@ class FloatArrayLookup(ArrayLookup[float]):
@strawberry.input(one_of=True, description='Lookup for Array fields. Only one of the lookup fields can be set.')
class StringArrayLookup(ArrayLookup[str]):
pass
@strawberry.input(one_of=True, description='Lookups for an ArrayField(RangeField). Only one may be set.')
class RangeArrayValueLookup(Generic[T]):
"""
class for Array field of Range fields lookups
"""
contains: T | None = strawberry.field(
default=strawberry.UNSET, description='Return rows where any stored range contains this value.'
)
@strawberry_django.filter_field
def filter(self, info: Info, queryset: QuerySet, prefix: str = '') -> Tuple[QuerySet, Q]:
"""
Map GraphQL: { <field>: { contains: <T> } } To Django ORM: <field>__range_contains=<T>
"""
if self.contains is strawberry.UNSET or self.contains is None:
return queryset, Q()
# Build '<prefix>range_contains' so it works for nested paths too
return queryset, Q(**{f'{prefix}range_contains': self.contains})
@strawberry.input(one_of=True, description='Lookups for an ArrayField(IntegerRangeField). Only one may be set.')
class IntegerRangeArrayLookup(RangeArrayValueLookup[int]):
pass

View File

@@ -1,6 +1,5 @@
import strawberry
import strawberry_django
from strawberry.types import Info
from django.contrib.contenttypes.models import ContentType
from core.graphql.mixins import ChangelogMixin
@@ -27,7 +26,7 @@ class BaseObjectType:
"""
@classmethod
def get_queryset(cls, queryset, info: Info, **kwargs):
def get_queryset(cls, queryset, info, **kwargs):
# Enforce object permissions on the queryset
if hasattr(queryset, 'restrict'):
return queryset.restrict(info.context.request.user, 'view')

View File

@@ -673,15 +673,10 @@ def has_feature(model_or_ct, feature):
# If an ObjectType was passed, we can use it directly
if type(model_or_ct) is ObjectType:
ot = model_or_ct
# If a ContentType was passed, resolve its model class and run the associated feature test
# If a ContentType was passed, resolve its model class
elif type(model_or_ct) is ContentType:
model = model_or_ct.model_class()
try:
test_func = registry['model_features'][feature]
except KeyError:
# Unknown feature
return False
return test_func(model)
model_class = model_or_ct.model_class()
ot = ObjectType.objects.get_for_model(model_class) if model_class else None
# For anything else, look up the ObjectType
else:
ot = ObjectType.objects.get_for_model(model_or_ct)

View File

@@ -1,39 +0,0 @@
from django.db.models import UniqueConstraint
from rest_framework.utils.field_mapping import get_unique_error_message
from rest_framework.validators import UniqueValidator
__all__ = (
'get_unique_validators',
)
def get_unique_validators(field_name, model_field):
"""
Extend Django REST Framework's get_unique_validators() function to attach a UniqueValidator to a field *only* if the
associated UniqueConstraint does NOT have a condition which references another field. See bug #19302.
"""
field_set = {field_name}
conditions = {
c.condition
for c in model_field.model._meta.constraints
if isinstance(c, UniqueConstraint) and set(c.fields) == field_set
}
# START custom logic
conditions = {
cond for cond in conditions
if cond.referenced_base_fields == field_set
}
# END custom logic
if getattr(model_field, 'unique', False):
conditions.add(None)
if not conditions:
return
unique_error_message = get_unique_error_message(model_field)
queryset = model_field.model._default_manager
for condition in conditions:
yield UniqueValidator(
queryset=queryset if condition is None else queryset.filter(condition),
message=unique_error_message
)

View File

@@ -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 = [IsSuperuser]
permission_classes = [permissions.IsAdminUser]
_ignore_model_permissions = True
schema = None

View File

@@ -26,6 +26,16 @@ 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=(

View File

@@ -11,7 +11,6 @@ from django.core.exceptions import ImproperlyConfigured, ValidationError
from django.core.validators import URLValidator
from django.utils.module_loading import import_string
from django.utils.translation import gettext_lazy as _
from rest_framework.utils import field_mapping
from core.exceptions import IncompatiblePluginError
from netbox.config import PARAMS as CONFIG_PARAMS
@@ -20,19 +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
from .monkey import get_unique_validators
#
# Monkey-patching
#
# TODO: Remove this once #20547 has been implemented
# Override DRF's get_unique_validators() function with our own (see bug #19302)
field_mapping.get_unique_validators = get_unique_validators
#
# Environment setup
@@ -44,9 +31,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, 12):
if sys.version_info < (3, 10):
raise RuntimeError(
f"NetBox requires Python 3.12 or later. (Currently installed: Python {platform.python_version()})"
f"NetBox requires Python 3.10 or later. (Currently installed: Python {platform.python_version()})"
)
#
@@ -76,8 +63,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",
@@ -175,6 +162,8 @@ 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)
@@ -228,12 +217,6 @@ 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:

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 not isinstance(record, model) or not getattr(record, 'pk', None):
if type(record) is not model or not getattr(record, 'pk', None):
return ''
if request := getattr(table, 'context', {}).get('request'):

View File

@@ -8,7 +8,6 @@ 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
@@ -17,159 +16,67 @@ from utilities.testing.api import APITestCase
class TokenAuthenticationTestCase(APITestCase):
@override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
def test_no_token(self):
def test_token_authentication(self):
url = reverse('dcim-api:site-list')
# Request without a token should return a 403
response = self.client.get(reverse('dcim-api:site-list'))
response = self.client.get(url)
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
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)
token = Token.objects.create(user=self.user)
response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token.key}')
self.assertEqual(response.status_code, 200)
# 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')
# 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}')
# Request without a non-expired token should succeed
token = Token.objects.create(user=self.user)
response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token.key}')
self.assertEqual(response.status_code, 200)
# Request with an expired token should fail
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}')
token.expires = datetime.datetime(2020, 1, 1, tzinfo=datetime.timezone.utc)
token.save()
response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token.key}')
self.assertEqual(response.status_code, 403)
@override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
def test_token_write_enabled(self):
url = reverse('dcim-api:site-list')
data = [
{
'name': 'Site 1',
'slug': 'site-1',
},
{
'name': 'Site 2',
'slug': 'site-2',
},
]
self.add_permissions('dcim.view_site', 'dcim.add_site')
data = {
'name': 'Site 1',
'slug': 'site-1',
}
# 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)
# Request with a write-disabled token should fail
token = Token.objects.create(user=self.user, write_enabled=False)
response = self.client.post(url, data, format='json', HTTP_AUTHORIZATION=f'Token {token.key}')
self.assertEqual(response.status_code, 403)
# 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)
# Request with a write-enabled token should succeed
token.write_enabled = True
token.save()
response = self.client.post(url, data, format='json', HTTP_AUTHORIZATION=f'Token {token.key}')
self.assertEqual(response.status_code, 403)
@override_settings(LOGIN_REQUIRED=True, EXEMPT_VIEW_PERMISSIONS=['*'])
def test_token_allowed_ips(self):
url = reverse('dcim-api:site-list')
# 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
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'
)
token = Token.objects.create(user=self.user, allowed_ips=['192.0.2.0/24'])
response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token.key}', REMOTE_ADDR='127.0.0.1')
self.assertEqual(response.status_code, 403)
# Request 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'
)
# Request with an expired token should fail
response = self.client.get(url, HTTP_AUTHORIZATION=f'Token {token.key}', REMOTE_ADDR='192.0.2.1')
self.assertEqual(response.status_code, 200)
@@ -520,7 +427,7 @@ class ObjectPermissionAPIViewTestCase(TestCase):
"""
self.user = User.objects.create(username='testuser')
self.token = Token.objects.create(user=self.user)
self.header = {'HTTP_AUTHORIZATION': f'Bearer {TOKEN_PREFIX}{self.token.key}.{self.token.token}'}
self.header = {'HTTP_AUTHORIZATION': 'Token {}'.format(self.token.key)}
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_get_object(self):

View File

@@ -281,8 +281,7 @@ class ObjectEditView(GetReturnURLMixin, BaseObjectView):
obj = self.alter_object(obj, request, args, kwargs)
form_prefix = 'quickadd' if request.GET.get('_quickadd') else None
form = self.form(data=request.POST, files=request.FILES, instance=obj, prefix=form_prefix)
form = self.form(data=request.POST, files=request.FILES, instance=obj)
restrict_form_fields(form, request.user)
if form.is_valid():

View File

@@ -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 superusers.)
# Check whether a new release is available. (Only for staff/superusers.)
new_release = None
if request.user.is_superuser:
if request.user.is_staff or request.user.is_superuser:
latest_release = cache.get('latest_release')
if latest_release:
release_version, release_url = latest_release

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View File

@@ -83,7 +83,7 @@ export function initRackElevation(): void {
}
for (const element of getElements<HTMLObjectElement>('.rack_elevation')) {
element.addEventListener('htmx:afterSettle', () => {
element.addEventListener('load', () => {
setRackView(initialView, element);
});
}

View File

@@ -1,3 +1,3 @@
version: "4.4.3"
version: "4.4.2"
edition: "Community"
published: "2025-10-14"
published: "2025-09-30"

View File

@@ -18,7 +18,7 @@
<li class="nav-item">
<a class="nav-link{% if active_tab == 'preferences' %} active{% endif %}" href="{% url 'account:preferences' %}">{% trans "Preferences" %}</a>
</li>
{% if request.user.has_usable_password %}
{% if not request.user.ldap_username %}
<li class="nav-item">
<a class="nav-link{% if active_tab == 'password' %} active{% endif %}" href="{% url 'account:change_password' %}">{% trans "Password" %}</a>
</li>

View File

@@ -39,6 +39,10 @@
<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>

View File

@@ -1,8 +1,62 @@
{% extends 'users/token.html' %}
{% extends 'generic/object.html' %}
{% load form_helpers %}
{% load helpers %}
{% 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 %}

View File

@@ -26,7 +26,7 @@
{# Initialize color mode #}
<script
type="text/javascript"
src="{% static_with_params 'setmode.js' v=settings.RELEASE.version %}"
src="{% static 'setmode.js' %}?v={{ settings.RELEASE.version }}"
onerror="window.location='{% url 'media_failure' %}?filename=setmode.js'">
</script>
<script type="text/javascript">
@@ -39,12 +39,12 @@
{# Static resources #}
<link
rel="stylesheet"
href="{% static_with_params 'netbox-external.css' v=settings.RELEASE.version %}"
href="{% static 'netbox-external.css'%}?v={{ settings.RELEASE.version }}"
onerror="window.location='{% url 'media_failure' %}?filename=netbox-external.css'"
/>
<link
rel="stylesheet"
href="{% static_with_params 'netbox.css' v=settings.RELEASE.version %}"
href="{% static 'netbox.css'%}?v={{ settings.RELEASE.version }}"
onerror="window.location='{% url 'media_failure' %}?filename=netbox.css'"
/>
<link rel="icon" type="image/png" href="{% static 'netbox.ico' %}" />
@@ -53,7 +53,7 @@
{# Javascript #}
<script
type="text/javascript"
src="{% static_with_params 'netbox.js' v=settings.RELEASE.version %}"
src="{% static 'netbox.js' %}?v={{ settings.RELEASE.version }}"
onerror="window.location='{% url 'media_failure' %}?filename=netbox.js'">
</script>
{% django_htmx_script %}

View File

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

View File

@@ -44,8 +44,8 @@
<div class="htmx-container table-responsive"
hx-get="{% url 'extras:script_result' job_pk=job.pk %}?embedded=True&log=True&log_threshold={{log_threshold}}"
hx-target="this"
hx-trigger="load" hx-select=".htmx-container" hx-swap="outerHTML">
</div>
hx-trigger="load" hx-select=".htmx-container" hx-swap="outerHTML"
></div>
</div>
</div>
{% endif %}
@@ -60,12 +60,11 @@
<a href="?export=output" class="btn btn-sm btn-primary" role="button">
<i class="mdi mdi-download" aria-hidden="true"></i> {% trans "Download" %}
</a>
{% copy_content "job_data_output" %}
</div>
{% endif %}
</h2>
{% if job.data.output %}
<pre class="card-body font-monospace" id="job_data_output">{{ job.data.output }}</pre>
<pre class="card-body font-monospace">{{ job.data.output }}</pre>
{% else %}
<div class="card-body text-muted">{% trans "None" %}</div>
{% endif %}

View File

@@ -27,13 +27,15 @@
<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">
<div class="dropdown-menu dropdown-menu-end dropdown-menu-arrow" {% htmx_boost %}>
<a href="{% url 'account:profile' %}" class="dropdown-item">
<i class="mdi mdi-account"></i> {% trans "Profile" %}
</a>

View File

@@ -40,7 +40,7 @@
</tr>
<tr>
<th scope="row">{% trans "VLAN IDs" %}</th>
<td>{{ object.vid_ranges_items|join:", " }}</td>
<td>{{ object.vid_ranges_list }}</td>
</tr>
<tr>
<th scope="row">Utilization</th>

View File

@@ -37,7 +37,7 @@
path. Refer to <a href="{{ docs_url }}">the installation documentation</a> for further guidance.
{% endblocktrans %}
<ul>
{% if request.user.is_superuser %}
{% if request.user.is_staff or 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>

View File

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

View File

@@ -35,6 +35,10 @@
<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>

View File

@@ -1,4 +1,3 @@
from django import forms
from django.contrib.contenttypes.models import ContentType
from django.utils.translation import gettext_lazy as _
@@ -26,7 +25,7 @@ class TenantGroupImportForm(NetBoxModelImportForm):
queryset=TenantGroup.objects.all(),
required=False,
to_field_name='name',
help_text=_('Parent group'),
help_text=_('Parent group')
)
slug = SlugField()
@@ -42,7 +41,7 @@ class TenantImportForm(NetBoxModelImportForm):
queryset=TenantGroup.objects.all(),
required=False,
to_field_name='name',
help_text=_('Assigned group'),
help_text=_('Assigned group')
)
class Meta:
@@ -60,7 +59,7 @@ class ContactGroupImportForm(NetBoxModelImportForm):
queryset=ContactGroup.objects.all(),
required=False,
to_field_name='name',
help_text=_('Parent group'),
help_text=_('Parent group')
)
slug = SlugField()
@@ -82,12 +81,7 @@ class ContactImportForm(NetBoxModelImportForm):
queryset=ContactGroup.objects.all(),
required=False,
to_field_name='name',
help_text=_('Group names separated by commas, encased with double quotes (e.g. "Group 1,Group 2")'),
)
link = forms.URLField(
label=_('Link'),
assume_scheme='https',
required=False,
help_text=_('Group names separated by commas, encased with double quotes (e.g. "Group 1,Group 2")')
)
class Meta:

View File

@@ -100,11 +100,6 @@ class ContactForm(NetBoxModelForm):
queryset=ContactGroup.objects.all(),
required=False
)
link = forms.URLField(
label=_('Link'),
assume_scheme='https',
required=False,
)
comments = CommentField()
fieldsets = (

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

Some files were not shown because too many files have changed in this diff Show More