diff --git a/netbox/templates/users/api_token.html b/netbox/templates/users/api_token.html
deleted file mode 100644
index 7fd6f064d..000000000
--- a/netbox/templates/users/api_token.html
+++ /dev/null
@@ -1,58 +0,0 @@
-{% extends 'generic/object.html' %}
-{% load form_helpers %}
-{% load helpers %}
-{% load plugins %}
-
-{% block content %}
-
-
- {% if not settings.ALLOW_TOKEN_RETRIEVAL %}
-
-
Tokens cannot be retrieved at a later time. You must
copy the token value below and store it securely.
-
- {% endif %}
-
-
-
-
-
- | Key |
-
-
- {% copy_content "token_id" %}
-
- {{ key }}
- |
-
-
- | Description |
- {{ object.description|placeholder }} |
-
-
- | User |
- {{ object.user }} |
-
-
- | Created |
- {{ object.created|annotated_date }} |
-
-
- | Expires |
-
- {% if object.expires %}
- {{ object.expires|annotated_date }}
- {% else %}
- Never
- {% endif %}
- |
-
-
-
-
-
-
-
-{% endblock %}
diff --git a/netbox/templates/users/group.html b/netbox/templates/users/group.html
new file mode 100644
index 000000000..e4eee0812
--- /dev/null
+++ b/netbox/templates/users/group.html
@@ -0,0 +1,48 @@
+{% extends 'generic/object.html' %}
+{% load i18n %}
+{% load helpers %}
+{% load render_table from django_tables2 %}
+
+{% block title %}{% trans "Group" %} {{ object.name }}{% endblock %}
+
+{% block subtitle %}{% endblock %}
+
+{% block content %}
+
+
+
+
+
+
+
+ | {% trans "Name" %} |
+ {{ object.name }} |
+
+
+
+
+
+
+
+
+
+ {% for user in object.user_set.all %}
+
{{ user }}
+ {% empty %}
+
{% trans "None" %}
+ {% endfor %}
+
+
+
+
+
+ {% for perm in object.object_permissions.all %}
+
{{ perm }}
+ {% empty %}
+
{% trans "None" %}
+ {% endfor %}
+
+
+
+
+{% endblock %}
diff --git a/netbox/templates/users/objectpermission.html b/netbox/templates/users/objectpermission.html
new file mode 100644
index 000000000..4da5a6ea5
--- /dev/null
+++ b/netbox/templates/users/objectpermission.html
@@ -0,0 +1,97 @@
+{% extends 'generic/object.html' %}
+{% load i18n %}
+{% load helpers %}
+{% load render_table from django_tables2 %}
+
+{% block title %}{% trans "Permission" %} {{ object.name }}{% endblock %}
+
+{% block subtitle %}{% endblock %}
+
+{% block content %}
+
+
+
+
+
+
+
+ | {% trans "Name" %} |
+ {{ object.name }} |
+
+
+ | {% trans "Description" %} |
+ {{ object.description|placeholder }} |
+
+
+ | {% trans "Enabled" %} |
+ {% checkmark object.enabled %} |
+
+
+
+
+
+
+
+
+
+ | {% trans "View" %} |
+ {% checkmark object.can_view %} |
+
+
+ | {% trans "Add" %} |
+ {% checkmark object.can_add %} |
+
+
+ | {% trans "Change" %} |
+ {% checkmark object.can_change %} |
+
+
+ | {% trans "Delete" %} |
+ {% checkmark object.can_delete %} |
+
+
+
+
+
+
+
+ {% if object.constraints %}
+
{{ object.constraints|json }}
+ {% else %}
+
None
+ {% endif %}
+
+
+
+
+
+
+
+ {% for user in object.object_types.all %}
+ - {{ user }}
+ {% endfor %}
+
+
+
+
+
+ {% for user in object.users.all %}
+
{{ user }}
+ {% empty %}
+
{% trans "None" %}
+ {% endfor %}
+
+
+
+
+
+ {% for group in object.groups.all %}
+
{{ group }}
+ {% empty %}
+
{% trans "None" %}
+ {% endfor %}
+
+
+
+
+{% endblock %}
diff --git a/netbox/templates/users/token.html b/netbox/templates/users/token.html
new file mode 100644
index 000000000..0fa8c572e
--- /dev/null
+++ b/netbox/templates/users/token.html
@@ -0,0 +1,56 @@
+{% extends 'generic/object.html' %}
+{% load i18n %}
+{% load helpers %}
+{% load render_table from django_tables2 %}
+
+{% block title %}{% trans "Token" %} {{ object }}{% endblock %}
+
+{% block subtitle %}{% endblock %}
+
+{% block content %}
+
+
+
+
+
+
+
+ | {% trans "Key" %} |
+ {% if settings.ALLOW_TOKEN_RETRIEVAL %}{{ object.key }}{% else %}{{ object.partial }}{% endif %} |
+
+
+ | {% trans "User" %} |
+
+ {{ object.user }}
+ |
+
+
+ | {% trans "Description" %} |
+ {{ object.description|placeholder }} |
+
+
+ | {% trans "Write enabled" %} |
+ {% checkmark object.write_enabled %} |
+
+
+ | {% trans "Created" %} |
+ {{ object.created|annotated_date }} |
+
+
+ | {% trans "Expires" %} |
+ {{ object.expires|placeholder }} |
+
+
+ | {% trans "Last used" %} |
+ {{ object.last_used|placeholder }} |
+
+
+ | {% trans "Allowed IPs" %} |
+ {{ object.allowed_ips|join:", "|placeholder }} |
+
+
+
+
+
+
+{% endblock %}
diff --git a/netbox/templates/users/user.html b/netbox/templates/users/user.html
new file mode 100644
index 000000000..fe03f41ed
--- /dev/null
+++ b/netbox/templates/users/user.html
@@ -0,0 +1,84 @@
+{% extends 'generic/object.html' %}
+{% load i18n %}
+{% load helpers %}
+{% load render_table from django_tables2 %}
+
+{% block title %}{% trans "User" %} {{ object.username }}{% endblock %}
+
+{% block subtitle %}{% endblock %}
+
+{% block content %}
+
+
+
+
+
+
+
+ | {% trans "Username" %} |
+ {{ object.username }} |
+
+
+ | {% trans "Full Name" %} |
+ {{ object.get_full_name|placeholder }} |
+
+
+ | {% trans "Email" %} |
+ {{ object.email|placeholder }} |
+
+
+ | {% trans "Account Created" %} |
+ {{ object.date_joined|annotated_date }} |
+
+
+ | {% trans "Active" %} |
+ {% checkmark object.active %} |
+
+
+ | {% trans "Staff" %} |
+ {% checkmark object.is_staff %} |
+
+
+ | {% trans "Superuser" %} |
+ {% checkmark object.is_superuser %} |
+
+
+
+
+
+
+
+
+
+ {% for group in object.groups.all %}
+
{{ group }}
+ {% empty %}
+
{% trans "None" %}
+ {% endfor %}
+
+
+
+
+
+ {% for perm in object.object_permissions.all %}
+
{{ perm }}
+ {% empty %}
+
{% trans "None" %}
+ {% endfor %}
+
+
+
+
+ {% if perms.extras.view_objectchange %}
+
+
+
+
+
+ {% render_table changelog_table 'inc/table.html' %}
+
+
+
+
+ {% endif %}
+{% endblock %}
diff --git a/netbox/users/account_urls.py b/netbox/users/account_urls.py
new file mode 100644
index 000000000..bcb031003
--- /dev/null
+++ b/netbox/users/account_urls.py
@@ -0,0 +1,18 @@
+from django.urls import include, path
+
+from utilities.urls import get_model_urls
+from . import views
+
+app_name = 'account'
+urlpatterns = [
+
+ # Account views
+ path('profile/', views.ProfileView.as_view(), name='profile'),
+ path('bookmarks/', views.BookmarkListView.as_view(), name='bookmarks'),
+ path('preferences/', views.UserConfigView.as_view(), name='preferences'),
+ path('password/', views.ChangePasswordView.as_view(), name='change_password'),
+ path('api-tokens/', views.UserTokenListView.as_view(), name='usertoken_list'),
+ path('api-tokens/add/', views.UserTokenEditView.as_view(), name='usertoken_add'),
+ path('api-tokens/
/', include(get_model_urls('users', 'usertoken'))),
+
+]
diff --git a/netbox/users/admin/__init__.py b/netbox/users/admin/__init__.py
index 2db822cfe..bc7bf7ab2 100644
--- a/netbox/users/admin/__init__.py
+++ b/netbox/users/admin/__init__.py
@@ -1,11 +1,6 @@
from django.contrib import admin
-from django.contrib.auth.admin import UserAdmin as UserAdmin_
from django.contrib.auth.models import Group, User
-from users.models import ObjectPermission, Token
-from . import filters, forms, inlines
-
-
#
# Users & groups
#
@@ -13,117 +8,3 @@ from . import filters, forms, inlines
# Unregister the built-in GroupAdmin and UserAdmin classes so that we can use our custom admin classes below
admin.site.unregister(Group)
admin.site.unregister(User)
-
-
-@admin.register(Group)
-class GroupAdmin(admin.ModelAdmin):
- form = forms.GroupAdminForm
- list_display = ('name', 'user_count')
- ordering = ('name',)
- search_fields = ('name',)
- inlines = [inlines.GroupObjectPermissionInline]
-
- @staticmethod
- def user_count(obj):
- return obj.user_set.count()
-
-
-@admin.register(User)
-class UserAdmin(UserAdmin_):
- list_display = [
- 'username', 'email', 'first_name', 'last_name', 'is_superuser', 'is_staff', 'is_active'
- ]
- fieldsets = (
- (None, {'fields': ('username', 'password', 'first_name', 'last_name', 'email')}),
- ('Groups', {'fields': ('groups',)}),
- ('Status', {
- 'fields': ('is_active', 'is_staff', 'is_superuser'),
- }),
- ('Important dates', {'fields': ('last_login', 'date_joined')}),
- )
- filter_horizontal = ('groups',)
- list_filter = ('is_active', 'is_staff', 'is_superuser', 'groups__name')
-
- def get_inlines(self, request, obj):
- if obj is not None:
- return (inlines.UserObjectPermissionInline, inlines.UserConfigInline)
- return ()
-
-
-#
-# REST API tokens
-#
-
-@admin.register(Token)
-class TokenAdmin(admin.ModelAdmin):
- form = forms.TokenAdminForm
- list_display = [
- 'key', 'user', 'created', 'expires', 'last_used', 'write_enabled', 'description', 'list_allowed_ips'
- ]
-
- def list_allowed_ips(self, obj):
- return obj.allowed_ips or 'Any'
- list_allowed_ips.short_description = "Allowed IPs"
-
-
-#
-# Permissions
-#
-
-@admin.register(ObjectPermission)
-class ObjectPermissionAdmin(admin.ModelAdmin):
- actions = ('enable', 'disable')
- fieldsets = (
- (None, {
- 'fields': ('name', 'description', 'enabled')
- }),
- ('Actions', {
- 'fields': (('can_view', 'can_add', 'can_change', 'can_delete'), 'actions')
- }),
- ('Objects', {
- 'fields': ('object_types',)
- }),
- ('Assignment', {
- 'fields': ('groups', 'users')
- }),
- ('Constraints', {
- 'fields': ('constraints',),
- 'classes': ('monospace',)
- }),
- )
- filter_horizontal = ('object_types', 'groups', 'users')
- form = forms.ObjectPermissionForm
- list_display = [
- 'name', 'enabled', 'list_models', 'list_users', 'list_groups', 'actions', 'constraints', 'description',
- ]
- list_filter = [
- 'enabled', filters.ActionListFilter, filters.ObjectTypeListFilter, 'groups', 'users'
- ]
- search_fields = ['actions', 'constraints', 'description', 'name']
-
- def get_queryset(self, request):
- return super().get_queryset(request).prefetch_related('object_types', 'users', 'groups')
-
- def list_models(self, obj):
- return ', '.join([f"{ct}" for ct in obj.object_types.all()])
- list_models.short_description = 'Models'
-
- def list_users(self, obj):
- return ', '.join([u.username for u in obj.users.all()])
- list_users.short_description = 'Users'
-
- def list_groups(self, obj):
- return ', '.join([g.name for g in obj.groups.all()])
- list_groups.short_description = 'Groups'
-
- #
- # Admin actions
- #
-
- def enable(self, request, queryset):
- updated = queryset.update(enabled=True)
- self.message_user(request, f"Enabled {updated} permissions")
-
- def disable(self, request, queryset):
- updated = queryset.update(enabled=False)
- self.message_user(request, f"Disabled {updated} permissions")
diff --git a/netbox/users/admin/forms.py b/netbox/users/admin/forms.py
deleted file mode 100644
index 986ddd0aa..000000000
--- a/netbox/users/admin/forms.py
+++ /dev/null
@@ -1,136 +0,0 @@
-from django import forms
-from django.contrib.auth.models import Group, User
-from django.contrib.admin.widgets import FilteredSelectMultiple
-from django.contrib.contenttypes.models import ContentType
-from django.core.exceptions import FieldError, ValidationError
-from django.utils.translation import gettext as _
-
-from users.constants import CONSTRAINT_TOKEN_USER, OBJECTPERMISSION_OBJECT_TYPES
-from users.models import ObjectPermission, Token
-from utilities.forms.fields import ContentTypeMultipleChoiceField
-from utilities.permissions import qs_filter_from_constraints
-
-__all__ = (
- 'GroupAdminForm',
- 'ObjectPermissionForm',
- 'TokenAdminForm',
-)
-
-
-class GroupAdminForm(forms.ModelForm):
- users = forms.ModelMultipleChoiceField(
- queryset=User.objects.all(),
- required=False,
- widget=FilteredSelectMultiple('users', False)
- )
-
- class Meta:
- model = Group
- fields = ('name', 'users')
-
- def __init__(self, *args, **kwargs):
- super(GroupAdminForm, self).__init__(*args, **kwargs)
-
- if self.instance.pk:
- self.fields['users'].initial = self.instance.user_set.all()
-
- def save_m2m(self):
- self.instance.user_set.set(self.cleaned_data['users'])
-
- def save(self, *args, **kwargs):
- instance = super(GroupAdminForm, self).save()
- self.save_m2m()
-
- return instance
-
-
-class TokenAdminForm(forms.ModelForm):
- key = forms.CharField(
- required=False,
- help_text=_("If no key is provided, one will be generated automatically.")
- )
-
- class Meta:
- fields = [
- 'user', 'key', 'write_enabled', 'expires', 'description', 'allowed_ips'
- ]
- model = Token
-
-
-class ObjectPermissionForm(forms.ModelForm):
- object_types = ContentTypeMultipleChoiceField(
- queryset=ContentType.objects.all(),
- limit_choices_to=OBJECTPERMISSION_OBJECT_TYPES
- )
- can_view = forms.BooleanField(required=False)
- can_add = forms.BooleanField(required=False)
- can_change = forms.BooleanField(required=False)
- can_delete = forms.BooleanField(required=False)
-
- class Meta:
- model = ObjectPermission
- exclude = []
- help_texts = {
- 'actions': _('Actions granted in addition to those listed above'),
- 'constraints': _('JSON expression of a queryset filter that will return only permitted objects. Leave null '
- 'to match all objects of this type. A list of multiple objects will result in a logical OR '
- 'operation.')
- }
- labels = {
- 'actions': 'Additional actions'
- }
- widgets = {
- 'constraints': forms.Textarea(attrs={'class': 'vLargeTextField'})
- }
-
- def __init__(self, *args, **kwargs):
- super().__init__(*args, **kwargs)
-
- # Make the actions field optional since the admin form uses it only for non-CRUD actions
- self.fields['actions'].required = False
-
- # Order group and user fields
- self.fields['groups'].queryset = self.fields['groups'].queryset.order_by('name')
- self.fields['users'].queryset = self.fields['users'].queryset.order_by('username')
-
- # Check the appropriate checkboxes when editing an existing ObjectPermission
- if self.instance.pk:
- for action in ['view', 'add', 'change', 'delete']:
- if action in self.instance.actions:
- self.fields[f'can_{action}'].initial = True
- self.instance.actions.remove(action)
-
- def clean(self):
- super().clean()
-
- object_types = self.cleaned_data.get('object_types')
- constraints = self.cleaned_data.get('constraints')
-
- # Append any of the selected CRUD checkboxes to the actions list
- if not self.cleaned_data.get('actions'):
- self.cleaned_data['actions'] = list()
- for action in ['view', 'add', 'change', 'delete']:
- if self.cleaned_data[f'can_{action}'] and action not in self.cleaned_data['actions']:
- self.cleaned_data['actions'].append(action)
-
- # At least one action must be specified
- if not self.cleaned_data['actions']:
- raise ValidationError("At least one action must be selected.")
-
- # Validate the specified model constraints by attempting to execute a query. We don't care whether the query
- # returns anything; we just want to make sure the specified constraints are valid.
- if object_types and constraints:
- # Normalize the constraints to a list of dicts
- if type(constraints) is not list:
- constraints = [constraints]
- for ct in object_types:
- model = ct.model_class()
- try:
- tokens = {
- CONSTRAINT_TOKEN_USER: 0, # Replace token with a null user ID
- }
- model.objects.filter(qs_filter_from_constraints(constraints, tokens)).exists()
- except FieldError as e:
- raise ValidationError({
- 'constraints': f'Invalid filter for {model}: {e}'
- })
diff --git a/netbox/users/filtersets.py b/netbox/users/filtersets.py
index 44ad98cc2..0f590e012 100644
--- a/netbox/users/filtersets.py
+++ b/netbox/users/filtersets.py
@@ -10,6 +10,7 @@ from users.models import ObjectPermission, Token
__all__ = (
'GroupFilterSet',
'ObjectPermissionFilterSet',
+ 'TokenFilterSet',
'UserFilterSet',
)
@@ -49,7 +50,7 @@ class UserFilterSet(BaseFilterSet):
class Meta:
model = get_user_model()
- fields = ['id', 'username', 'first_name', 'last_name', 'email', 'is_staff', 'is_active']
+ fields = ['id', 'username', 'first_name', 'last_name', 'email', 'is_staff', 'is_active', 'is_superuser']
def search(self, queryset, name, value):
if not value.strip():
@@ -115,6 +116,18 @@ class ObjectPermissionFilterSet(BaseFilterSet):
method='search',
label=_('Search'),
)
+ can_view = django_filters.BooleanFilter(
+ method='_check_action'
+ )
+ can_add = django_filters.BooleanFilter(
+ method='_check_action'
+ )
+ can_change = django_filters.BooleanFilter(
+ method='_check_action'
+ )
+ can_delete = django_filters.BooleanFilter(
+ method='_check_action'
+ )
user_id = django_filters.ModelMultipleChoiceFilter(
field_name='users',
queryset=get_user_model().objects.all(),
@@ -149,3 +162,10 @@ class ObjectPermissionFilterSet(BaseFilterSet):
Q(name__icontains=value) |
Q(description__icontains=value)
)
+
+ def _check_action(self, queryset, name, value):
+ action = name.split('_')[1]
+ if value:
+ return queryset.filter(actions__contains=[action])
+ else:
+ return queryset.exclude(actions__contains=[action])
diff --git a/netbox/users/forms.py b/netbox/users/forms.py
deleted file mode 100644
index 027fa5327..000000000
--- a/netbox/users/forms.py
+++ /dev/null
@@ -1,130 +0,0 @@
-from django import forms
-from django.conf import settings
-from django.contrib.auth.forms import AuthenticationForm, PasswordChangeForm as DjangoPasswordChangeForm
-from django.contrib.postgres.forms import SimpleArrayField
-from django.utils.html import mark_safe
-from django.utils.translation import gettext as _
-
-from ipam.formfields import IPNetworkFormField
-from ipam.validators import prefix_validator
-from netbox.preferences import PREFERENCES
-from utilities.forms import BootstrapMixin
-from utilities.forms.widgets import DateTimePicker
-from utilities.utils import flatten_dict
-from .models import Token, UserConfig
-
-
-class LoginForm(BootstrapMixin, AuthenticationForm):
- pass
-
-
-class PasswordChangeForm(BootstrapMixin, DjangoPasswordChangeForm):
- pass
-
-
-class UserConfigFormMetaclass(forms.models.ModelFormMetaclass):
-
- def __new__(mcs, name, bases, attrs):
-
- # Emulate a declared field for each supported user preference
- preference_fields = {}
- for field_name, preference in PREFERENCES.items():
- description = f'{preference.description}
' if preference.description else ''
- help_text = f'{description}{field_name}'
- field_kwargs = {
- 'label': preference.label,
- 'choices': preference.choices,
- 'help_text': mark_safe(help_text),
- 'coerce': preference.coerce,
- 'required': False,
- 'widget': forms.Select,
- }
- preference_fields[field_name] = forms.TypedChoiceField(**field_kwargs)
- attrs.update(preference_fields)
-
- return super().__new__(mcs, name, bases, attrs)
-
-
-class UserConfigForm(BootstrapMixin, forms.ModelForm, metaclass=UserConfigFormMetaclass):
- fieldsets = (
- ('User Interface', (
- 'pagination.per_page',
- 'pagination.placement',
- 'ui.colormode',
- )),
- ('Miscellaneous', (
- 'data_format',
- )),
- )
- # List of clearable preferences
- pk = forms.MultipleChoiceField(
- choices=[],
- required=False
- )
-
- class Meta:
- model = UserConfig
- fields = ()
-
- def __init__(self, *args, instance=None, **kwargs):
-
- # Get initial data from UserConfig instance
- initial_data = flatten_dict(instance.data)
- kwargs['initial'] = initial_data
-
- super().__init__(*args, instance=instance, **kwargs)
-
- # Compile clearable preference choices
- self.fields['pk'].choices = (
- (f'tables.{table_name}', '') for table_name in instance.data.get('tables', [])
- )
-
- def save(self, *args, **kwargs):
-
- # Set UserConfig data
- for pref_name, value in self.cleaned_data.items():
- if pref_name == 'pk':
- continue
- self.instance.set(pref_name, value, commit=False)
-
- # Clear selected preferences
- for preference in self.cleaned_data['pk']:
- self.instance.clear(preference)
-
- return super().save(*args, **kwargs)
-
- @property
- def plugin_fields(self):
- return [
- name for name in self.fields.keys() if name.startswith('plugins.')
- ]
-
-
-class TokenForm(BootstrapMixin, forms.ModelForm):
- key = forms.CharField(
- required=False,
- help_text=_("If no key is provided, one will be generated automatically.")
- )
- allowed_ips = SimpleArrayField(
- base_field=IPNetworkFormField(validators=[prefix_validator]),
- required=False,
- label=_('Allowed IPs'),
- help_text=_('Allowed IPv4/IPv6 networks from where the token can be used. Leave blank for no restrictions. '
- 'Example: 10.1.1.0/24,192.168.10.16/32,2001:db8:1::/64'),
- )
-
- class Meta:
- model = Token
- fields = [
- 'key', 'write_enabled', 'expires', 'description', 'allowed_ips',
- ]
- widgets = {
- 'expires': DateTimePicker(),
- }
-
- def __init__(self, *args, **kwargs):
- super().__init__(*args, **kwargs)
-
- # Omit the key field if token retrieval is not permitted
- if self.instance.pk and not settings.ALLOW_TOKEN_RETRIEVAL:
- del self.fields['key']
diff --git a/netbox/users/forms/__init__.py b/netbox/users/forms/__init__.py
new file mode 100644
index 000000000..a545c3add
--- /dev/null
+++ b/netbox/users/forms/__init__.py
@@ -0,0 +1,5 @@
+from .authentication import *
+from .bulk_edit import *
+from .bulk_import import *
+from .filtersets import *
+from .model_forms import *
diff --git a/netbox/users/forms/authentication.py b/netbox/users/forms/authentication.py
new file mode 100644
index 000000000..2b540b752
--- /dev/null
+++ b/netbox/users/forms/authentication.py
@@ -0,0 +1,25 @@
+from django.contrib.auth.forms import (
+ AuthenticationForm,
+ PasswordChangeForm as DjangoPasswordChangeForm,
+)
+
+from utilities.forms import BootstrapMixin
+
+__all__ = (
+ 'LoginForm',
+ 'PasswordChangeForm',
+)
+
+
+class LoginForm(BootstrapMixin, AuthenticationForm):
+ """
+ Used to authenticate a user by username and password.
+ """
+ pass
+
+
+class PasswordChangeForm(BootstrapMixin, DjangoPasswordChangeForm):
+ """
+ This form enables a user to change his or her own password.
+ """
+ pass
diff --git a/netbox/users/forms/bulk_edit.py b/netbox/users/forms/bulk_edit.py
new file mode 100644
index 000000000..0e29109a4
--- /dev/null
+++ b/netbox/users/forms/bulk_edit.py
@@ -0,0 +1,111 @@
+from django import forms
+from django.contrib.postgres.forms import SimpleArrayField
+from django.utils.translation import gettext_lazy as _
+
+from ipam.formfields import IPNetworkFormField
+from ipam.validators import prefix_validator
+from users.models import *
+from utilities.forms import BootstrapMixin, BulkEditForm
+from utilities.forms.widgets import BulkEditNullBooleanSelect, DateTimePicker
+
+__all__ = (
+ 'ObjectPermissionBulkEditForm',
+ 'UserBulkEditForm',
+ 'TokenBulkEditForm',
+)
+
+
+class UserBulkEditForm(BootstrapMixin, forms.Form):
+ pk = forms.ModelMultipleChoiceField(
+ queryset=NetBoxUser.objects.all(),
+ widget=forms.MultipleHiddenInput
+ )
+ first_name = forms.CharField(
+ label=_('First name'),
+ max_length=150,
+ required=False
+ )
+ last_name = forms.CharField(
+ label=_('Last name'),
+ max_length=150,
+ required=False
+ )
+ is_active = forms.NullBooleanField(
+ required=False,
+ widget=BulkEditNullBooleanSelect,
+ label=_('Active')
+ )
+ is_staff = forms.NullBooleanField(
+ required=False,
+ widget=BulkEditNullBooleanSelect,
+ label=_('Staff status')
+ )
+ is_superuser = forms.NullBooleanField(
+ required=False,
+ widget=BulkEditNullBooleanSelect,
+ label=_('Superuser status')
+ )
+
+ model = NetBoxUser
+ fieldsets = (
+ (None, ('first_name', 'last_name', 'is_active', 'is_staff', 'is_superuser')),
+ )
+ nullable_fields = ('first_name', 'last_name')
+
+
+class ObjectPermissionBulkEditForm(BootstrapMixin, forms.Form):
+ pk = forms.ModelMultipleChoiceField(
+ queryset=ObjectPermission.objects.all(),
+ widget=forms.MultipleHiddenInput
+ )
+ description = forms.CharField(
+ label=_('Description'),
+ max_length=200,
+ required=False
+ )
+ enabled = forms.NullBooleanField(
+ required=False,
+ widget=BulkEditNullBooleanSelect,
+ label=_('Enabled')
+ )
+
+ model = ObjectPermission
+ fieldsets = (
+ (None, ('enabled', 'description')),
+ )
+ nullable_fields = ('description',)
+
+
+class TokenBulkEditForm(BulkEditForm):
+ pk = forms.ModelMultipleChoiceField(
+ queryset=Token.objects.all(),
+ widget=forms.MultipleHiddenInput
+ )
+ write_enabled = forms.NullBooleanField(
+ required=False,
+ widget=BulkEditNullBooleanSelect,
+ label=_('Write enabled')
+ )
+ description = forms.CharField(
+ max_length=200,
+ required=False,
+ label=_('Description')
+ )
+ expires = forms.DateTimeField(
+ required=False,
+ widget=DateTimePicker(),
+ label=_('Expires')
+ )
+ allowed_ips = SimpleArrayField(
+ base_field=IPNetworkFormField(validators=[prefix_validator]),
+ required=False,
+ label=_('Allowed IPs')
+ )
+
+ model = Token
+ fieldsets = (
+ (None, ('write_enabled', 'description', 'expires', 'allowed_ips')),
+ )
+ nullable_fields = (
+ 'expires', 'description', 'allowed_ips',
+ )
diff --git a/netbox/users/forms/bulk_import.py b/netbox/users/forms/bulk_import.py
new file mode 100644
index 000000000..d1f03ff3c
--- /dev/null
+++ b/netbox/users/forms/bulk_import.py
@@ -0,0 +1,48 @@
+from django import forms
+from django.utils.translation import gettext as _
+from users.models import *
+from utilities.forms import CSVModelForm
+
+
+__all__ = (
+ 'GroupImportForm',
+ 'UserImportForm',
+ 'TokenImportForm',
+)
+
+
+class GroupImportForm(CSVModelForm):
+
+ class Meta:
+ model = NetBoxGroup
+ fields = (
+ 'name',
+ )
+
+
+class UserImportForm(CSVModelForm):
+
+ class Meta:
+ model = NetBoxUser
+ fields = (
+ 'username', 'first_name', 'last_name', 'email', 'password', 'is_staff',
+ 'is_active', 'is_superuser'
+ )
+
+ def save(self, *args, **kwargs):
+ # Set the hashed password
+ self.instance.set_password(self.cleaned_data.get('password'))
+
+ return super().save(*args, **kwargs)
+
+
+class TokenImportForm(CSVModelForm):
+ key = forms.CharField(
+ label=_('Key'),
+ required=False,
+ help_text=_("If no key is provided, one will be generated automatically.")
+ )
+
+ class Meta:
+ model = Token
+ fields = ('user', 'key', 'write_enabled', 'expires', 'description',)
diff --git a/netbox/users/forms/filtersets.py b/netbox/users/forms/filtersets.py
new file mode 100644
index 000000000..ff56cbc4c
--- /dev/null
+++ b/netbox/users/forms/filtersets.py
@@ -0,0 +1,146 @@
+from django import forms
+from extras.forms.mixins import SavedFiltersMixin
+from utilities.forms import FilterForm
+from users.models import Token
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Group
+from django.utils.translation import gettext_lazy as _
+
+from netbox.forms import NetBoxModelFilterSetForm
+from users.models import NetBoxGroup, NetBoxUser, ObjectPermission
+from utilities.forms import BOOLEAN_WITH_BLANK_CHOICES
+from utilities.forms.fields import DynamicModelMultipleChoiceField
+from utilities.forms.widgets import DateTimePicker
+
+__all__ = (
+ 'GroupFilterForm',
+ 'ObjectPermissionFilterForm',
+ 'UserFilterForm',
+ 'TokenFilterForm',
+)
+
+
+class GroupFilterForm(NetBoxModelFilterSetForm):
+ model = NetBoxGroup
+ fieldsets = (
+ (None, ('q', 'filter_id',)),
+ )
+
+
+class UserFilterForm(NetBoxModelFilterSetForm):
+ model = NetBoxUser
+ fieldsets = (
+ (None, ('q', 'filter_id',)),
+ (_('Group'), ('group_id',)),
+ (_('Status'), ('is_active', 'is_staff', 'is_superuser')),
+ )
+ group_id = DynamicModelMultipleChoiceField(
+ queryset=Group.objects.all(),
+ required=False,
+ label=_('Group')
+ )
+ is_active = forms.NullBooleanField(
+ required=False,
+ widget=forms.Select(
+ choices=BOOLEAN_WITH_BLANK_CHOICES
+ ),
+ label=_('Is Active'),
+ )
+ is_staff = forms.NullBooleanField(
+ required=False,
+ widget=forms.Select(
+ choices=BOOLEAN_WITH_BLANK_CHOICES
+ ),
+ label=_('Is Staff'),
+ )
+ is_superuser = forms.NullBooleanField(
+ required=False,
+ widget=forms.Select(
+ choices=BOOLEAN_WITH_BLANK_CHOICES
+ ),
+ label=_('Is Superuser'),
+ )
+
+
+class ObjectPermissionFilterForm(NetBoxModelFilterSetForm):
+ model = ObjectPermission
+ fieldsets = (
+ (None, ('q', 'filter_id',)),
+ (_('Permission'), ('enabled', 'group_id', 'user_id')),
+ (_('Actions'), ('can_view', 'can_add', 'can_change', 'can_delete')),
+ )
+ enabled = forms.NullBooleanField(
+ label=_('Enabled'),
+ required=False,
+ widget=forms.Select(
+ choices=BOOLEAN_WITH_BLANK_CHOICES
+ )
+ )
+ group_id = DynamicModelMultipleChoiceField(
+ queryset=Group.objects.all(),
+ required=False,
+ label=_('Group')
+ )
+ user_id = DynamicModelMultipleChoiceField(
+ queryset=get_user_model().objects.all(),
+ required=False,
+ label=_('User')
+ )
+ can_view = forms.NullBooleanField(
+ required=False,
+ widget=forms.Select(
+ choices=BOOLEAN_WITH_BLANK_CHOICES
+ ),
+ label=_('Can View'),
+ )
+ can_add = forms.NullBooleanField(
+ required=False,
+ widget=forms.Select(
+ choices=BOOLEAN_WITH_BLANK_CHOICES
+ ),
+ label=_('Can Add'),
+ )
+ can_change = forms.NullBooleanField(
+ required=False,
+ widget=forms.Select(
+ choices=BOOLEAN_WITH_BLANK_CHOICES
+ ),
+ label=_('Can Change'),
+ )
+ can_delete = forms.NullBooleanField(
+ required=False,
+ widget=forms.Select(
+ choices=BOOLEAN_WITH_BLANK_CHOICES
+ ),
+ label=_('Can Delete'),
+ )
+
+
+class TokenFilterForm(SavedFiltersMixin, FilterForm):
+ model = Token
+ fieldsets = (
+ (None, ('q', 'filter_id',)),
+ (_('Token'), ('user_id', 'write_enabled', 'expires', 'last_used')),
+ )
+ user_id = DynamicModelMultipleChoiceField(
+ queryset=get_user_model().objects.all(),
+ required=False,
+ label=_('User')
+ )
+ write_enabled = forms.NullBooleanField(
+ required=False,
+ widget=forms.Select(
+ choices=BOOLEAN_WITH_BLANK_CHOICES
+ ),
+ label=_('Write Enabled'),
+ )
+ expires = forms.DateTimeField(
+ required=False,
+ label=_('Expires'),
+ widget=DateTimePicker()
+ )
+ last_used = forms.DateTimeField(
+ required=False,
+ label=_('Last Used'),
+ widget=DateTimePicker()
+ )
diff --git a/netbox/users/forms/model_forms.py b/netbox/users/forms/model_forms.py
new file mode 100644
index 000000000..6ca050110
--- /dev/null
+++ b/netbox/users/forms/model_forms.py
@@ -0,0 +1,403 @@
+from django import forms
+from django.conf import settings
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Group
+from django.contrib.contenttypes.models import ContentType
+from django.contrib.postgres.forms import SimpleArrayField
+from django.core.exceptions import FieldError
+from django.utils.html import mark_safe
+from django.utils.translation import gettext_lazy as _
+
+from ipam.formfields import IPNetworkFormField
+from ipam.validators import prefix_validator
+from netbox.preferences import PREFERENCES
+from users.constants import *
+from users.models import *
+from utilities.forms import BootstrapMixin
+from utilities.forms.fields import ContentTypeMultipleChoiceField, DynamicModelMultipleChoiceField
+from utilities.forms.widgets import DateTimePicker
+from utilities.permissions import qs_filter_from_constraints
+from utilities.utils import flatten_dict
+
+__all__ = (
+ 'UserTokenForm',
+ 'GroupForm',
+ 'ObjectPermissionForm',
+ 'TokenForm',
+ 'UserConfigForm',
+ 'UserForm',
+ 'TokenForm',
+)
+
+
+class UserConfigFormMetaclass(forms.models.ModelFormMetaclass):
+
+ def __new__(mcs, name, bases, attrs):
+
+ # Emulate a declared field for each supported user preference
+ preference_fields = {}
+ for field_name, preference in PREFERENCES.items():
+ description = f'{preference.description}
' if preference.description else ''
+ help_text = f'{description}{field_name}'
+ field_kwargs = {
+ 'label': preference.label,
+ 'choices': preference.choices,
+ 'help_text': mark_safe(help_text),
+ 'coerce': preference.coerce,
+ 'required': False,
+ 'widget': forms.Select,
+ }
+ preference_fields[field_name] = forms.TypedChoiceField(**field_kwargs)
+ attrs.update(preference_fields)
+
+ return super().__new__(mcs, name, bases, attrs)
+
+
+class UserConfigForm(BootstrapMixin, forms.ModelForm, metaclass=UserConfigFormMetaclass):
+ fieldsets = (
+ (_('User Interface'), (
+ 'pagination.per_page',
+ 'pagination.placement',
+ 'ui.colormode',
+ )),
+ (_('Miscellaneous'), (
+ 'data_format',
+ )),
+ )
+ # List of clearable preferences
+ pk = forms.MultipleChoiceField(
+ label=_('Pk'),
+ choices=[],
+ required=False
+ )
+
+ class Meta:
+ model = UserConfig
+ fields = ()
+
+ def __init__(self, *args, instance=None, **kwargs):
+
+ # Get initial data from UserConfig instance
+ initial_data = flatten_dict(instance.data)
+ kwargs['initial'] = initial_data
+
+ super().__init__(*args, instance=instance, **kwargs)
+
+ # Compile clearable preference choices
+ self.fields['pk'].choices = (
+ (f'tables.{table_name}', '') for table_name in instance.data.get('tables', [])
+ )
+
+ def save(self, *args, **kwargs):
+
+ # Set UserConfig data
+ for pref_name, value in self.cleaned_data.items():
+ if pref_name == 'pk':
+ continue
+ self.instance.set(pref_name, value, commit=False)
+
+ # Clear selected preferences
+ for preference in self.cleaned_data['pk']:
+ self.instance.clear(preference)
+
+ return super().save(*args, **kwargs)
+
+ @property
+ def plugin_fields(self):
+ return [
+ name for name in self.fields.keys() if name.startswith('plugins.')
+ ]
+
+
+class UserTokenForm(BootstrapMixin, forms.ModelForm):
+ key = forms.CharField(
+ label=_('Key'),
+ required=False,
+ help_text=_("If no key is provided, one will be generated automatically.")
+ )
+ allowed_ips = SimpleArrayField(
+ base_field=IPNetworkFormField(validators=[prefix_validator]),
+ required=False,
+ label=_('Allowed IPs'),
+ help_text=_(
+ 'Allowed IPv4/IPv6 networks from where the token can be used. Leave blank for no restrictions. '
+ 'Example: 10.1.1.0/24,192.168.10.16/32,2001:db8:1::/64'
+ ),
+ )
+
+ class Meta:
+ model = Token
+ fields = [
+ 'key', 'write_enabled', 'expires', 'description', 'allowed_ips',
+ ]
+ widgets = {
+ 'expires': DateTimePicker(),
+ }
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+
+ # Omit the key field if token retrieval is not permitted
+ if self.instance.pk and not settings.ALLOW_TOKEN_RETRIEVAL:
+ del self.fields['key']
+
+
+class TokenForm(UserTokenForm):
+ user = forms.ModelChoiceField(
+ queryset=get_user_model().objects.order_by(
+ 'username'
+ ),
+ required=False
+ )
+
+ class Meta:
+ model = Token
+ fields = [
+ 'user', 'key', 'write_enabled', 'expires', 'description', 'allowed_ips',
+ ]
+ widgets = {
+ 'expires': DateTimePicker(),
+ }
+
+
+class UserForm(BootstrapMixin, forms.ModelForm):
+ password = forms.CharField(
+ label=_('Password'),
+ widget=forms.PasswordInput(),
+ required=True,
+ )
+ confirm_password = forms.CharField(
+ label=_('Confirm password'),
+ widget=forms.PasswordInput(),
+ required=True,
+ help_text=_("Enter the same password as before, for verification."),
+ )
+ groups = DynamicModelMultipleChoiceField(
+ label=_('Groups'),
+ required=False,
+ queryset=Group.objects.all()
+ )
+ object_permissions = DynamicModelMultipleChoiceField(
+ required=False,
+ label=_('Permissions'),
+ queryset=ObjectPermission.objects.all(),
+ to_field_name='pk',
+ )
+
+ fieldsets = (
+ (_('User'), ('username', 'password', 'confirm_password', 'first_name', 'last_name', 'email')),
+ (_('Groups'), ('groups', )),
+ (_('Status'), ('is_active', 'is_staff', 'is_superuser')),
+ (_('Permissions'), ('object_permissions',)),
+ )
+
+ class Meta:
+ model = NetBoxUser
+ fields = [
+ 'username', 'first_name', 'last_name', 'email', 'groups', 'object_permissions',
+ 'is_active', 'is_staff', 'is_superuser',
+ ]
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+
+ if self.instance.pk:
+ # Populate assigned permissions
+ self.fields['object_permissions'].initial = self.instance.object_permissions.values_list('id', flat=True)
+
+ # Password fields are optional for existing Users
+ self.fields['password'].required = False
+ self.fields['password'].widget.attrs.pop('required')
+ self.fields['confirm_password'].required = False
+ self.fields['confirm_password'].widget.attrs.pop('required')
+
+ def save(self, *args, **kwargs):
+ instance = super().save(*args, **kwargs)
+
+ # Update assigned permissions
+ instance.object_permissions.set(self.cleaned_data['object_permissions'])
+
+ # On edit, check if we have to save the password
+ if self.cleaned_data.get('password'):
+ instance.set_password(self.cleaned_data.get('password'))
+ instance.save()
+
+ return instance
+
+ def clean(self):
+
+ # Check that password confirmation matches if password is set
+ if self.cleaned_data['password'] and self.cleaned_data['password'] != self.cleaned_data['confirm_password']:
+ raise forms.ValidationError(_("Passwords do not match! Please check your input and try again."))
+
+ # TODO: Move this logic to the NetBoxUser class
+ def clean_username(self):
+ """Reject usernames that differ only in case."""
+ instance = getattr(self, 'instance', None)
+ if instance:
+ qs = self._meta.model.objects.exclude(pk=instance.pk)
+ else:
+ qs = self._meta.model.objects.all()
+
+ username = self.cleaned_data.get("username")
+ if (
+ username and qs.filter(username__iexact=username).exists()
+ ):
+ raise forms.ValidationError(
+ _("user with this username already exists")
+ )
+
+ return username
+
+
+class GroupForm(BootstrapMixin, forms.ModelForm):
+ users = DynamicModelMultipleChoiceField(
+ label=_('Users'),
+ required=False,
+ queryset=get_user_model().objects.all()
+ )
+ object_permissions = DynamicModelMultipleChoiceField(
+ required=False,
+ label=_('Permissions'),
+ queryset=ObjectPermission.objects.all(),
+ to_field_name='pk',
+ )
+
+ fieldsets = (
+ (None, ('name', )),
+ (_('Users'), ('users', )),
+ (_('Permissions'), ('object_permissions', )),
+ )
+
+ class Meta:
+ model = NetBoxGroup
+ fields = [
+ 'name', 'users', 'object_permissions',
+ ]
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+
+ # Populate assigned users and permissions
+ if self.instance.pk:
+ self.fields['users'].initial = self.instance.user_set.values_list('id', flat=True)
+ self.fields['object_permissions'].initial = self.instance.object_permissions.values_list('id', flat=True)
+
+ def save(self, *args, **kwargs):
+ instance = super().save(*args, **kwargs)
+
+ # Update assigned users and permissions
+ instance.user_set.set(self.cleaned_data['users'])
+ instance.object_permissions.set(self.cleaned_data['object_permissions'])
+
+ return instance
+
+
+class ObjectPermissionForm(BootstrapMixin, forms.ModelForm):
+ object_types = ContentTypeMultipleChoiceField(
+ label=_('Object types'),
+ queryset=ContentType.objects.all(),
+ limit_choices_to=OBJECTPERMISSION_OBJECT_TYPES,
+ widget=forms.SelectMultiple(attrs={'size': 6})
+ )
+ can_view = forms.BooleanField(
+ required=False
+ )
+ can_add = forms.BooleanField(
+ required=False
+ )
+ can_change = forms.BooleanField(
+ required=False
+ )
+ can_delete = forms.BooleanField(
+ required=False
+ )
+ actions = SimpleArrayField(
+ label=_('Additional actions'),
+ base_field=forms.CharField(),
+ required=False,
+ help_text=_('Actions granted in addition to those listed above')
+ )
+ users = DynamicModelMultipleChoiceField(
+ label=_('Users'),
+ required=False,
+ queryset=get_user_model().objects.all()
+ )
+ groups = DynamicModelMultipleChoiceField(
+ label=_('Groups'),
+ required=False,
+ queryset=Group.objects.all()
+ )
+
+ fieldsets = (
+ (None, ('name', 'description', 'enabled',)),
+ (_('Actions'), ('can_view', 'can_add', 'can_change', 'can_delete', 'actions')),
+ (_('Objects'), ('object_types', )),
+ (_('Assignment'), ('groups', 'users')),
+ (_('Constraints'), ('constraints',))
+ )
+
+ class Meta:
+ model = ObjectPermission
+ fields = [
+ 'name', 'description', 'enabled', 'object_types', 'users', 'groups', 'constraints', 'actions',
+ ]
+ help_texts = {
+ 'constraints': _(
+ 'JSON expression of a queryset filter that will return only permitted objects. Leave null '
+ 'to match all objects of this type. A list of multiple objects will result in a logical OR '
+ 'operation.'
+ )
+ }
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+
+ # Make the actions field optional since the form uses it only for non-CRUD actions
+ self.fields['actions'].required = False
+
+ # Order group and user fields
+ self.fields['groups'].queryset = self.fields['groups'].queryset.order_by('name')
+ self.fields['users'].queryset = self.fields['users'].queryset.order_by('username')
+
+ # Check the appropriate checkboxes when editing an existing ObjectPermission
+ if self.instance.pk:
+ for action in ['view', 'add', 'change', 'delete']:
+ if action in self.instance.actions:
+ self.fields[f'can_{action}'].initial = True
+ self.instance.actions.remove(action)
+
+ def clean(self):
+ super().clean()
+
+ object_types = self.cleaned_data.get('object_types')
+ constraints = self.cleaned_data.get('constraints')
+
+ # Append any of the selected CRUD checkboxes to the actions list
+ if not self.cleaned_data.get('actions'):
+ self.cleaned_data['actions'] = list()
+ for action in ['view', 'add', 'change', 'delete']:
+ if self.cleaned_data[f'can_{action}'] and action not in self.cleaned_data['actions']:
+ self.cleaned_data['actions'].append(action)
+
+ # At least one action must be specified
+ if not self.cleaned_data['actions']:
+ raise forms.ValidationError(_("At least one action must be selected."))
+
+ # Validate the specified model constraints by attempting to execute a query. We don't care whether the query
+ # returns anything; we just want to make sure the specified constraints are valid.
+ if object_types and constraints:
+ # Normalize the constraints to a list of dicts
+ if type(constraints) is not list:
+ constraints = [constraints]
+ for ct in object_types:
+ model = ct.model_class()
+ try:
+ tokens = {
+ CONSTRAINT_TOKEN_USER: 0, # Replace token with a null user ID
+ }
+ model.objects.filter(qs_filter_from_constraints(constraints, tokens)).exists()
+ except FieldError as e:
+ raise forms.ValidationError({
+ 'constraints': _('Invalid filter for {model}: {e}').format(model=model, e=e)
+ })
diff --git a/netbox/users/migrations/0004_netboxgroup_netboxuser.py b/netbox/users/migrations/0004_netboxgroup_netboxuser.py
new file mode 100644
index 000000000..59d941643
--- /dev/null
+++ b/netbox/users/migrations/0004_netboxgroup_netboxuser.py
@@ -0,0 +1,50 @@
+# Generated by Django 4.1.9 on 2023-06-06 18:15
+
+import django.contrib.auth.models
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+ dependencies = [
+ ('auth', '0012_alter_user_first_name_max_length'),
+ ('users', '0003_token_allowed_ips_last_used'),
+ ]
+
+ operations = [
+ migrations.CreateModel(
+ name='NetBoxGroup',
+ fields=[],
+ options={
+ 'verbose_name': 'Group',
+ 'proxy': True,
+ 'indexes': [],
+ 'constraints': [],
+ },
+ bases=('auth.group',),
+ managers=[
+ ('objects', django.contrib.auth.models.GroupManager()),
+ ],
+ ),
+ migrations.CreateModel(
+ name='NetBoxUser',
+ fields=[],
+ options={
+ 'verbose_name': 'User',
+ 'proxy': True,
+ 'indexes': [],
+ 'constraints': [],
+ },
+ bases=('auth.user',),
+ managers=[
+ ('objects', django.contrib.auth.models.UserManager()),
+ ],
+ ),
+ migrations.AlterModelOptions(
+ name='netboxgroup',
+ options={'ordering': ('name',), 'verbose_name': 'Group'},
+ ),
+ migrations.AlterModelOptions(
+ name='netboxuser',
+ options={'ordering': ('username',), 'verbose_name': 'User'},
+ ),
+ ]
diff --git a/netbox/users/migrations/0005_usertoken.py b/netbox/users/migrations/0005_usertoken.py
new file mode 100644
index 000000000..c6aef0590
--- /dev/null
+++ b/netbox/users/migrations/0005_usertoken.py
@@ -0,0 +1,25 @@
+# Generated by Django 4.1.10 on 2023-07-25 15:19
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('users', '0004_netboxgroup_netboxuser'),
+ ]
+
+ operations = [
+ migrations.CreateModel(
+ name='UserToken',
+ fields=[
+ ],
+ options={
+ 'proxy': True,
+ 'indexes': [],
+ 'constraints': [],
+ 'verbose_name': 'token',
+ },
+ bases=('users.token',),
+ ),
+ ]
diff --git a/netbox/users/models.py b/netbox/users/models.py
index 4e7d9ca52..0c95559ff 100644
--- a/netbox/users/models.py
+++ b/netbox/users/models.py
@@ -2,13 +2,14 @@ import binascii
import os
from django.conf import settings
-from django.contrib.auth.models import Group, User
+from django.contrib.auth.models import Group, GroupManager, User, UserManager
from django.contrib.contenttypes.models import ContentType
from django.contrib.postgres.fields import ArrayField
from django.core.validators import MinLengthValidator
from django.db import models
from django.db.models.signals import post_save
from django.dispatch import receiver
+from django.urls import reverse
from django.utils import timezone
from django.utils.translation import gettext as _
from netaddr import IPNetwork
@@ -20,9 +21,12 @@ from utilities.utils import flatten_dict
from .constants import *
__all__ = (
+ 'NetBoxGroup',
+ 'NetBoxUser',
'ObjectPermission',
'Token',
'UserConfig',
+ 'UserToken',
)
@@ -30,6 +34,7 @@ __all__ = (
# Proxy models for admin
#
+
class AdminGroup(Group):
"""
Proxy contrib.auth.models.Group for the admin UI
@@ -48,6 +53,44 @@ class AdminUser(User):
proxy = True
+class NetBoxUserManager(UserManager.from_queryset(RestrictedQuerySet)):
+ pass
+
+
+class NetBoxGroupManager(GroupManager.from_queryset(RestrictedQuerySet)):
+ pass
+
+
+class NetBoxUser(User):
+ """
+ Proxy contrib.auth.models.User for the UI
+ """
+ objects = NetBoxUserManager()
+
+ class Meta:
+ verbose_name = 'User'
+ proxy = True
+ ordering = ('username',)
+
+ def get_absolute_url(self):
+ return reverse('users:netboxuser', args=[self.pk])
+
+
+class NetBoxGroup(Group):
+ """
+ Proxy contrib.auth.models.User for the UI
+ """
+ objects = NetBoxGroupManager()
+
+ class Meta:
+ verbose_name = 'Group'
+ proxy = True
+ ordering = ('name',)
+
+ def get_absolute_url(self):
+ return reverse('users:netboxgroup', args=[self.pk])
+
+
#
# User preferences
#
@@ -231,13 +274,20 @@ class Token(models.Model):
blank=True,
null=True,
verbose_name='Allowed IPs',
- help_text=_('Allowed IPv4/IPv6 networks from where the token can be used. Leave blank for no restrictions. '
- 'Ex: "10.1.1.0/24, 192.168.10.16/32, 2001:DB8:1::/64"'),
+ help_text=_(
+ 'Allowed IPv4/IPv6 networks from where the token can be used. Leave blank for no restrictions. '
+ 'Ex: "10.1.1.0/24, 192.168.10.16/32, 2001:DB8:1::/64"'
+ ),
)
+ objects = RestrictedQuerySet.as_manager()
+
def __str__(self):
return self.key if settings.ALLOW_TOKEN_RETRIEVAL else self.partial
+ def get_absolute_url(self):
+ return reverse('users:token', args=[self.pk])
+
@property
def partial(self):
return f'**********************************{self.key[-6:]}' if self.key else ''
@@ -272,6 +322,18 @@ class Token(models.Model):
return False
+class UserToken(Token):
+ """
+ Proxy model for users to manage their own API tokens.
+ """
+ class Meta:
+ proxy = True
+ verbose_name = 'token'
+
+ def get_absolute_url(self):
+ return reverse('account:usertoken', args=[self.pk])
+
+
#
# Permissions
#
@@ -325,6 +387,22 @@ class ObjectPermission(models.Model):
def __str__(self):
return self.name
+ @property
+ def can_view(self):
+ return 'view' in self.actions
+
+ @property
+ def can_add(self):
+ return 'add' in self.actions
+
+ @property
+ def can_change(self):
+ return 'change' in self.actions
+
+ @property
+ def can_delete(self):
+ return 'delete' in self.actions
+
def list_constraints(self):
"""
Return all constraint sets as a list (even if only a single set is defined).
@@ -332,3 +410,6 @@ class ObjectPermission(models.Model):
if type(self.constraints) is not list:
return [self.constraints]
return self.constraints
+
+ def get_absolute_url(self):
+ return reverse('users:objectpermission', args=[self.pk])
diff --git a/netbox/users/tables.py b/netbox/users/tables.py
index cea50b10f..3ef885399 100644
--- a/netbox/users/tables.py
+++ b/netbox/users/tables.py
@@ -1,8 +1,14 @@
-from .models import Token
+import django_tables2 as tables
+from django.utils.translation import gettext as _
+
from netbox.tables import NetBoxTable, columns
+from users.models import NetBoxGroup, NetBoxUser, ObjectPermission, Token, UserToken
__all__ = (
+ 'GroupTable',
+ 'ObjectPermissionTable',
'TokenTable',
+ 'UserTable',
)
@@ -25,17 +31,28 @@ class TokenActionsColumn(columns.ActionsColumn):
}
-class TokenTable(NetBoxTable):
+class UserTokenTable(NetBoxTable):
+ """
+ Table for users to manager their own API tokens under account views.
+ """
key = columns.TemplateColumn(
- template_code=TOKEN
+ verbose_name=_('Key'),
+ template_code=TOKEN,
)
write_enabled = columns.BooleanColumn(
- verbose_name='Write'
+ verbose_name=_('Write Enabled')
+ )
+ created = columns.DateColumn(
+ verbose_name=_('Created'),
+ )
+ expires = columns.DateColumn(
+ verbose_name=_('Expires'),
+ )
+ last_used = columns.DateTimeColumn(
+ verbose_name=_('Last Used'),
)
- created = columns.DateColumn()
- expired = columns.DateColumn()
- last_used = columns.DateTimeColumn()
allowed_ips = columns.TemplateColumn(
+ verbose_name=_('Allowed IPs'),
template_code=ALLOWED_IPS
)
actions = TokenActionsColumn(
@@ -43,8 +60,93 @@ class TokenTable(NetBoxTable):
extra_buttons=COPY_BUTTON
)
+ class Meta(NetBoxTable.Meta):
+ model = UserToken
+ fields = (
+ 'pk', 'id', 'key', 'description', 'write_enabled', 'created', 'expires', 'last_used', 'allowed_ips',
+ )
+
+
+class TokenTable(UserTokenTable):
+ """
+ General-purpose table for API token management.
+ """
+ user = tables.Column(
+ linkify=True,
+ verbose_name=_('User')
+ )
+
class Meta(NetBoxTable.Meta):
model = Token
fields = (
- 'pk', 'description', 'key', 'write_enabled', 'created', 'expires', 'last_used', 'allowed_ips',
+ 'pk', 'id', 'key', 'user', 'description', 'write_enabled', 'created', 'expires', 'last_used', 'allowed_ips',
+ )
+
+
+class UserTable(NetBoxTable):
+ username = tables.Column(
+ linkify=True
+ )
+ groups = columns.ManyToManyColumn(
+ linkify_item=('users:netboxgroup', {'pk': tables.A('pk')})
+ )
+ is_active = columns.BooleanColumn()
+ is_staff = columns.BooleanColumn()
+ is_superuser = columns.BooleanColumn()
+ actions = columns.ActionsColumn(
+ actions=('edit', 'delete'),
+ )
+
+ class Meta(NetBoxTable.Meta):
+ model = NetBoxUser
+ fields = (
+ 'pk', 'id', 'username', 'first_name', 'last_name', 'email', 'groups', 'is_active', 'is_staff',
+ 'is_superuser',
+ )
+ default_columns = ('pk', 'username', 'first_name', 'last_name', 'email', 'is_active')
+
+
+class GroupTable(NetBoxTable):
+ name = tables.Column(linkify=True)
+ actions = columns.ActionsColumn(
+ actions=('edit', 'delete'),
+ )
+
+ class Meta(NetBoxTable.Meta):
+ model = NetBoxGroup
+ fields = (
+ 'pk', 'id', 'name', 'users_count',
+ )
+ default_columns = ('pk', 'name', 'users_count', )
+
+
+class ObjectPermissionTable(NetBoxTable):
+ name = tables.Column(linkify=True)
+ object_types = columns.ContentTypesColumn()
+ enabled = columns.BooleanColumn()
+ can_view = columns.BooleanColumn()
+ can_add = columns.BooleanColumn()
+ can_change = columns.BooleanColumn()
+ can_delete = columns.BooleanColumn()
+ custom_actions = columns.ArrayColumn(
+ accessor=tables.A('actions')
+ )
+ users = columns.ManyToManyColumn(
+ linkify_item=('users:netboxuser', {'pk': tables.A('pk')})
+ )
+ groups = columns.ManyToManyColumn(
+ linkify_item=('users:netboxgroup', {'pk': tables.A('pk')})
+ )
+ actions = columns.ActionsColumn(
+ actions=('edit', 'delete'),
+ )
+
+ class Meta(NetBoxTable.Meta):
+ model = ObjectPermission
+ fields = (
+ 'pk', 'id', 'name', 'enabled', 'object_types', 'can_view', 'can_add', 'can_change', 'can_delete',
+ 'custom_actions', 'users', 'groups', 'constraints', 'description',
+ )
+ default_columns = (
+ 'pk', 'name', 'enabled', 'object_types', 'can_view', 'can_add', 'can_change', 'can_delete', 'description',
)
diff --git a/netbox/users/tests/test_filtersets.py b/netbox/users/tests/test_filtersets.py
index d632687ef..542b40b83 100644
--- a/netbox/users/tests/test_filtersets.py
+++ b/netbox/users/tests/test_filtersets.py
@@ -10,7 +10,6 @@ from users import filtersets
from users.models import ObjectPermission, Token
from utilities.testing import BaseFilterSetTests
-
User = get_user_model()
@@ -34,7 +33,8 @@ class UserTestCase(TestCase, BaseFilterSetTests):
first_name='Hank',
last_name='Hill',
email='hank@stricklandpropane.com',
- is_staff=True
+ is_staff=True,
+ is_superuser=True
),
User(
username='User2',
@@ -83,13 +83,17 @@ class UserTestCase(TestCase, BaseFilterSetTests):
params = {'email': ['hank@stricklandpropane.com', 'dale@dalesdeadbug.com']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+ def test_is_active(self):
+ params = {'is_active': True}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
+
def test_is_staff(self):
params = {'is_staff': True}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
- def test_is_active(self):
- params = {'is_active': True}
- self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
+ def test_is_superuser(self):
+ params = {'is_superuser': True}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_group(self):
groups = Group.objects.all()[:2]
@@ -191,6 +195,22 @@ class ObjectPermissionTestCase(TestCase, BaseFilterSetTests):
params = {'description': ['foobar1', 'foobar2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+ def test_can_view(self):
+ params = {'can_view': True}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
+
+ def test_can_add(self):
+ params = {'can_add': True}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
+
+ def test_can_change(self):
+ params = {'can_change': True}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
+
+ def test_can_delete(self):
+ params = {'can_delete': True}
+ self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
+
class TokenTestCase(TestCase, BaseFilterSetTests):
queryset = Token.objects.all()
diff --git a/netbox/users/tests/test_views.py b/netbox/users/tests/test_views.py
new file mode 100644
index 000000000..2997052eb
--- /dev/null
+++ b/netbox/users/tests/test_views.py
@@ -0,0 +1,201 @@
+from django.contrib.auth.models import Group
+from django.contrib.contenttypes.models import ContentType
+
+from users.models import *
+from utilities.testing import ViewTestCases, create_test_user
+
+
+class UserTestCase(
+ ViewTestCases.GetObjectViewTestCase,
+ ViewTestCases.CreateObjectViewTestCase,
+ ViewTestCases.EditObjectViewTestCase,
+ ViewTestCases.DeleteObjectViewTestCase,
+ ViewTestCases.ListObjectsViewTestCase,
+ ViewTestCases.BulkImportObjectsViewTestCase,
+ ViewTestCases.BulkEditObjectsViewTestCase,
+ ViewTestCases.BulkDeleteObjectsViewTestCase,
+):
+ model = NetBoxUser
+ maxDiff = None
+ validation_excluded_fields = ['password']
+
+ def _get_queryset(self):
+ # Omit the user attached to the test client
+ return self.model.objects.exclude(username='testuser')
+
+ @classmethod
+ def setUpTestData(cls):
+
+ users = (
+ NetBoxUser(username='username1', first_name='first1', last_name='last1', email='user1@foo.com', password='pass1xxx'),
+ NetBoxUser(username='username2', first_name='first2', last_name='last2', email='user2@foo.com', password='pass2xxx'),
+ NetBoxUser(username='username3', first_name='first3', last_name='last3', email='user3@foo.com', password='pass3xxx'),
+ )
+ NetBoxUser.objects.bulk_create(users)
+
+ cls.form_data = {
+ 'username': 'usernamex',
+ 'first_name': 'firstx',
+ 'last_name': 'lastx',
+ 'email': 'userx@foo.com',
+ 'password': 'pass1xxx',
+ 'confirm_password': 'pass1xxx',
+ }
+
+ cls.csv_data = (
+ "username,first_name,last_name,email,password",
+ "username4,first4,last4,email4@foo.com,pass4xxx",
+ "username5,first5,last5,email5@foo.com,pass5xxx",
+ "username6,first6,last6,email6@foo.com,pass6xxx",
+ )
+
+ cls.csv_update_data = (
+ "id,first_name,last_name",
+ f"{users[0].pk},first7,last7",
+ f"{users[1].pk},first8,last8",
+ f"{users[2].pk},first9,last9",
+ )
+
+ cls.bulk_edit_data = {
+ 'last_name': 'newlastname',
+ }
+
+
+class GroupTestCase(
+ ViewTestCases.GetObjectViewTestCase,
+ ViewTestCases.CreateObjectViewTestCase,
+ ViewTestCases.EditObjectViewTestCase,
+ ViewTestCases.DeleteObjectViewTestCase,
+ ViewTestCases.ListObjectsViewTestCase,
+ ViewTestCases.BulkImportObjectsViewTestCase,
+ ViewTestCases.BulkDeleteObjectsViewTestCase,
+):
+ model = NetBoxGroup
+ maxDiff = None
+
+ @classmethod
+ def setUpTestData(cls):
+
+ groups = (
+ Group(name='group1'),
+ Group(name='group2'),
+ Group(name='group3'),
+ )
+ Group.objects.bulk_create(groups)
+
+ cls.form_data = {
+ 'name': 'groupx',
+ }
+
+ cls.csv_data = (
+ "name",
+ "group4"
+ "group5"
+ "group6"
+ )
+
+ cls.csv_update_data = (
+ "id,name",
+ f"{groups[0].pk},group7",
+ f"{groups[1].pk},group8",
+ f"{groups[2].pk},group9",
+ )
+
+
+class ObjectPermissionTestCase(
+ ViewTestCases.GetObjectViewTestCase,
+ ViewTestCases.CreateObjectViewTestCase,
+ ViewTestCases.EditObjectViewTestCase,
+ ViewTestCases.DeleteObjectViewTestCase,
+ ViewTestCases.ListObjectsViewTestCase,
+ ViewTestCases.BulkEditObjectsViewTestCase,
+ ViewTestCases.BulkDeleteObjectsViewTestCase,
+):
+ model = ObjectPermission
+ maxDiff = None
+
+ @classmethod
+ def setUpTestData(cls):
+ ct = ContentType.objects.get_by_natural_key('dcim', 'site')
+
+ permissions = (
+ ObjectPermission(name='Permission 1', actions=['view', 'add', 'delete']),
+ ObjectPermission(name='Permission 2', actions=['view', 'add', 'delete']),
+ ObjectPermission(name='Permission 3', actions=['view', 'add', 'delete']),
+ )
+ ObjectPermission.objects.bulk_create(permissions)
+
+ cls.form_data = {
+ 'name': 'Permission X',
+ 'description': 'A new permission',
+ 'object_types': [ct.pk],
+ 'actions': 'view,edit,delete',
+ }
+
+ cls.csv_data = (
+ "name",
+ "permission4"
+ "permission5"
+ "permission6"
+ )
+
+ cls.csv_update_data = (
+ "id,name,actions",
+ f"{permissions[0].pk},permission7",
+ f"{permissions[1].pk},permission8",
+ f"{permissions[2].pk},permission9",
+ )
+
+ cls.bulk_edit_data = {
+ 'description': 'New description',
+ }
+
+
+class TokenTestCase(
+ ViewTestCases.GetObjectViewTestCase,
+ ViewTestCases.CreateObjectViewTestCase,
+ ViewTestCases.EditObjectViewTestCase,
+ ViewTestCases.DeleteObjectViewTestCase,
+ ViewTestCases.ListObjectsViewTestCase,
+ ViewTestCases.BulkImportObjectsViewTestCase,
+ ViewTestCases.BulkEditObjectsViewTestCase,
+ ViewTestCases.BulkDeleteObjectsViewTestCase,
+):
+ model = Token
+ maxDiff = None
+
+ @classmethod
+ def setUpTestData(cls):
+ users = (
+ create_test_user('User 1'),
+ create_test_user('User 2'),
+ )
+ tokens = (
+ Token(key='123456790123456789012345678901234567890A', user=users[0]),
+ Token(key='123456790123456789012345678901234567890B', user=users[0]),
+ Token(key='123456790123456789012345678901234567890C', user=users[1]),
+ )
+ Token.objects.bulk_create(tokens)
+
+ cls.form_data = {
+ 'user': users[0].pk,
+ 'description': 'testdescription',
+ }
+
+ cls.csv_data = (
+ "key,user,description",
+ f"123456790123456789012345678901234567890D,{users[0].pk},testdescriptionD",
+ f"123456790123456789012345678901234567890E,{users[1].pk},testdescriptionE",
+ f"123456790123456789012345678901234567890F,{users[1].pk},testdescriptionF",
+ )
+
+ cls.csv_update_data = (
+ "id,description",
+ f"{tokens[0].pk},testdescriptionH",
+ f"{tokens[1].pk},testdescriptionI",
+ f"{tokens[2].pk},testdescriptionJ",
+ )
+
+ cls.bulk_edit_data = {
+ 'description': 'newdescription',
+ }
diff --git a/netbox/users/urls.py b/netbox/users/urls.py
index 7cb1f3435..210d8a2c7 100644
--- a/netbox/users/urls.py
+++ b/netbox/users/urls.py
@@ -6,15 +6,34 @@ from . import views
app_name = 'users'
urlpatterns = [
- # User
- path('profile/', views.ProfileView.as_view(), name='profile'),
- path('bookmarks/', views.BookmarkListView.as_view(), name='bookmarks'),
- path('preferences/', views.UserConfigView.as_view(), name='preferences'),
- path('password/', views.ChangePasswordView.as_view(), name='change_password'),
+ # Tokens
+ path('tokens/', views.TokenListView.as_view(), name='token_list'),
+ path('tokens/add/', views.TokenEditView.as_view(), name='token_add'),
+ path('tokens/import/', views.TokenBulkImportView.as_view(), name='token_import'),
+ path('tokens/edit/', views.TokenBulkEditView.as_view(), name='token_bulk_edit'),
+ path('tokens/delete/', views.TokenBulkDeleteView.as_view(), name='token_bulk_delete'),
+ path('tokens//', include(get_model_urls('users', 'token'))),
- # API tokens
- path('api-tokens/', views.TokenListView.as_view(), name='token_list'),
- path('api-tokens/add/', views.TokenEditView.as_view(), name='token_add'),
- path('api-tokens//', include(get_model_urls('users', 'token'))),
+ # Users
+ path('users/', views.UserListView.as_view(), name='netboxuser_list'),
+ path('users/add/', views.UserEditView.as_view(), name='netboxuser_add'),
+ path('users/edit/', views.UserBulkEditView.as_view(), name='netboxuser_bulk_edit'),
+ path('users/import/', views.UserBulkImportView.as_view(), name='netboxuser_import'),
+ path('users/delete/', views.UserBulkDeleteView.as_view(), name='netboxuser_bulk_delete'),
+ path('users//', include(get_model_urls('users', 'netboxuser'))),
+
+ # Groups
+ path('groups/', views.GroupListView.as_view(), name='netboxgroup_list'),
+ path('groups/add/', views.GroupEditView.as_view(), name='netboxgroup_add'),
+ path('groups/import/', views.GroupBulkImportView.as_view(), name='netboxgroup_import'),
+ path('groups/delete/', views.GroupBulkDeleteView.as_view(), name='netboxgroup_bulk_delete'),
+ path('groups//', include(get_model_urls('users', 'netboxgroup'))),
+
+ # Permissions
+ path('permissions/', views.ObjectPermissionListView.as_view(), name='objectpermission_list'),
+ path('permissions/add/', views.ObjectPermissionEditView.as_view(), name='objectpermission_add'),
+ path('permissions/edit/', views.ObjectPermissionBulkEditView.as_view(), name='objectpermission_bulk_edit'),
+ path('permissions/delete/', views.ObjectPermissionBulkDeleteView.as_view(), name='objectpermission_bulk_delete'),
+ path('permissions//', include(get_model_urls('users', 'objectpermission'))),
]
diff --git a/netbox/users/views.py b/netbox/users/views.py
index ad80fdfe5..3796d9af1 100644
--- a/netbox/users/views.py
+++ b/netbox/users/views.py
@@ -6,6 +6,7 @@ from django.contrib.auth import login as auth_login, logout as auth_logout, upda
from django.contrib.auth.mixins import LoginRequiredMixin
from django.contrib.auth.models import update_last_login
from django.contrib.auth.signals import user_logged_in
+from django.db.models import Count
from django.http import HttpResponseRedirect
from django.shortcuts import get_object_or_404, redirect, render, resolve_url
from django.urls import reverse
@@ -19,12 +20,11 @@ from extras.models import Bookmark, ObjectChange
from extras.tables import BookmarkTable, ObjectChangeTable
from netbox.authentication import get_auth_backend_display, get_saml_idps
from netbox.config import get_config
-from netbox.views.generic import ObjectListView
+from netbox.views import generic
from utilities.forms import ConfirmationForm
from utilities.views import register_model_view
-from .forms import LoginForm, PasswordChangeForm, TokenForm, UserConfigForm
-from .models import Token, UserConfig
-from .tables import TokenTable
+from . import filtersets, forms, tables
+from .models import NetBoxGroup, NetBoxUser, ObjectPermission, Token, UserConfig, UserToken
#
@@ -70,7 +70,7 @@ class LoginView(View):
return auth_backends
def get(self, request):
- form = LoginForm(request)
+ form = forms.LoginForm(request)
if request.user.is_authenticated:
logger = logging.getLogger('netbox.auth.login')
@@ -83,7 +83,7 @@ class LoginView(View):
def post(self, request):
logger = logging.getLogger('netbox.auth.login')
- form = LoginForm(request, data=request.POST)
+ form = forms.LoginForm(request, data=request.POST)
if form.is_valid():
logger.debug("Login form validation was successful")
@@ -155,7 +155,7 @@ class LogoutView(View):
#
class ProfileView(LoginRequiredMixin, View):
- template_name = 'users/profile.html'
+ template_name = 'users/account/profile.html'
def get(self, request):
@@ -174,11 +174,11 @@ class ProfileView(LoginRequiredMixin, View):
class UserConfigView(LoginRequiredMixin, View):
- template_name = 'users/preferences.html'
+ template_name = 'users/account/preferences.html'
def get(self, request):
userconfig = request.user.config
- form = UserConfigForm(instance=userconfig)
+ form = forms.UserConfigForm(instance=userconfig)
return render(request, self.template_name, {
'form': form,
@@ -187,13 +187,13 @@ class UserConfigView(LoginRequiredMixin, View):
def post(self, request):
userconfig = request.user.config
- form = UserConfigForm(request.POST, instance=userconfig)
+ form = forms.UserConfigForm(request.POST, instance=userconfig)
if form.is_valid():
form.save()
messages.success(request, "Your preferences have been updated.")
- return redirect('users:preferences')
+ return redirect('account:preferences')
return render(request, self.template_name, {
'form': form,
@@ -202,15 +202,15 @@ class UserConfigView(LoginRequiredMixin, View):
class ChangePasswordView(LoginRequiredMixin, View):
- template_name = 'users/password.html'
+ template_name = 'users/account/password.html'
def get(self, request):
# LDAP users cannot change their password here
if getattr(request.user, 'ldap_username', None):
messages.warning(request, "LDAP-authenticated user credentials cannot be changed within NetBox.")
- return redirect('users:profile')
+ return redirect('account:profile')
- form = PasswordChangeForm(user=request.user)
+ form = forms.PasswordChangeForm(user=request.user)
return render(request, self.template_name, {
'form': form,
@@ -218,12 +218,12 @@ class ChangePasswordView(LoginRequiredMixin, View):
})
def post(self, request):
- form = PasswordChangeForm(user=request.user, data=request.POST)
+ form = forms.PasswordChangeForm(user=request.user, data=request.POST)
if form.is_valid():
form.save()
update_session_auth_hash(request, form.user)
messages.success(request, "Your password has been changed successfully.")
- return redirect('users:profile')
+ return redirect('account:profile')
return render(request, self.template_name, {
'form': form,
@@ -235,9 +235,9 @@ class ChangePasswordView(LoginRequiredMixin, View):
# Bookmarks
#
-class BookmarkListView(LoginRequiredMixin, ObjectListView):
+class BookmarkListView(LoginRequiredMixin, generic.ObjectListView):
table = BookmarkTable
- template_name = 'users/bookmarks.html'
+ template_name = 'users/account/bookmarks.html'
def get_queryset(self, request):
return Bookmark.objects.filter(user=request.user)
@@ -249,53 +249,61 @@ class BookmarkListView(LoginRequiredMixin, ObjectListView):
#
-# API tokens
+# User views for token management
#
-class TokenListView(LoginRequiredMixin, View):
+class UserTokenListView(LoginRequiredMixin, View):
def get(self, request):
-
- tokens = Token.objects.filter(user=request.user)
- table = TokenTable(tokens)
+ tokens = UserToken.objects.filter(user=request.user)
+ table = tables.UserTokenTable(tokens)
table.configure(request)
- return render(request, 'users/api_tokens.html', {
+ return render(request, 'users/account/token_list.html', {
'tokens': tokens,
'active_tab': 'api-tokens',
'table': table,
})
-@register_model_view(Token, 'edit')
-class TokenEditView(LoginRequiredMixin, View):
+@register_model_view(UserToken)
+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, 'users/account/token.html', {
+ 'object': token,
+ 'key': key,
+ })
+
+
+@register_model_view(UserToken, 'edit')
+class UserTokenEditView(LoginRequiredMixin, View):
def get(self, request, pk=None):
-
if pk:
- token = get_object_or_404(Token.objects.filter(user=request.user), pk=pk)
+ token = get_object_or_404(UserToken.objects.filter(user=request.user), pk=pk)
else:
- token = Token(user=request.user)
-
- form = TokenForm(instance=token)
+ token = UserToken(user=request.user)
+ form = forms.UserTokenForm(instance=token)
return render(request, 'generic/object_edit.html', {
'object': token,
'form': form,
- 'return_url': reverse('users:token_list'),
+ 'return_url': reverse('account:usertoken_list'),
})
def post(self, request, pk=None):
-
if pk:
- token = get_object_or_404(Token.objects.filter(user=request.user), pk=pk)
- form = TokenForm(request.POST, instance=token)
+ token = get_object_or_404(UserToken.objects.filter(user=request.user), pk=pk)
+ form = forms.UserTokenForm(request.POST, instance=token)
else:
- token = Token(user=request.user)
- form = TokenForm(request.POST)
+ token = UserToken(user=request.user)
+ form = forms.UserTokenForm(request.POST)
if form.is_valid():
-
token = form.save(commit=False)
token.user = request.user
token.save()
@@ -304,7 +312,7 @@ class TokenEditView(LoginRequiredMixin, View):
messages.success(request, msg)
if not pk and not settings.ALLOW_TOKEN_RETRIEVAL:
- return render(request, 'users/api_token.html', {
+ return render(request, 'users/account/token.html', {
'object': token,
'key': token.key,
'return_url': reverse('users:token_list'),
@@ -312,44 +320,216 @@ class TokenEditView(LoginRequiredMixin, View):
elif '_addanother' in request.POST:
return redirect(request.path)
else:
- return redirect('users:token_list')
+ return redirect('account:usertoken_list')
return render(request, 'generic/object_edit.html', {
'object': token,
'form': form,
- 'return_url': reverse('users:token_list'),
+ 'return_url': reverse('account:usertoken_list'),
'disable_addanother': not settings.ALLOW_TOKEN_RETRIEVAL
})
-@register_model_view(Token, 'delete')
-class TokenDeleteView(LoginRequiredMixin, View):
+@register_model_view(UserToken, 'delete')
+class UserTokenDeleteView(LoginRequiredMixin, View):
def get(self, request, pk):
-
- token = get_object_or_404(Token.objects.filter(user=request.user), pk=pk)
- initial_data = {
- 'return_url': reverse('users:token_list'),
- }
- form = ConfirmationForm(initial=initial_data)
+ token = get_object_or_404(UserToken.objects.filter(user=request.user), pk=pk)
return render(request, 'generic/object_delete.html', {
'object': token,
- 'form': form,
- 'return_url': reverse('users:token_list'),
+ 'form': ConfirmationForm(),
+ 'return_url': reverse('account:usertoken_list'),
})
def post(self, request, pk):
-
- token = get_object_or_404(Token.objects.filter(user=request.user), pk=pk)
+ token = get_object_or_404(UserToken.objects.filter(user=request.user), pk=pk)
form = ConfirmationForm(request.POST)
+
if form.is_valid():
token.delete()
messages.success(request, "Token deleted")
- return redirect('users:token_list')
+ return redirect('account:usertoken_list')
return render(request, 'generic/object_delete.html', {
'object': token,
'form': form,
- 'return_url': reverse('users:token_list'),
+ 'return_url': reverse('account:usertoken_list'),
})
+
+
+#
+# Tokens
+#
+
+class TokenListView(generic.ObjectListView):
+ queryset = Token.objects.all()
+ filterset = filtersets.TokenFilterSet
+ filterset_form = forms.TokenFilterForm
+ table = tables.TokenTable
+
+
+@register_model_view(Token)
+class TokenView(generic.ObjectView):
+ queryset = Token.objects.all()
+
+
+@register_model_view(Token, 'edit')
+class TokenEditView(generic.ObjectEditView):
+ queryset = Token.objects.all()
+ form = forms.TokenForm
+
+
+@register_model_view(Token, 'delete')
+class TokenDeleteView(generic.ObjectDeleteView):
+ queryset = Token.objects.all()
+
+
+class TokenBulkImportView(generic.BulkImportView):
+ queryset = Token.objects.all()
+ model_form = forms.TokenImportForm
+
+
+class TokenBulkEditView(generic.BulkEditView):
+ queryset = Token.objects.all()
+ table = tables.TokenTable
+ form = forms.TokenBulkEditForm
+
+
+class TokenBulkDeleteView(generic.BulkDeleteView):
+ queryset = Token.objects.all()
+ table = tables.TokenTable
+
+
+#
+# Users
+#
+
+class UserListView(generic.ObjectListView):
+ queryset = NetBoxUser.objects.all()
+ filterset = filtersets.UserFilterSet
+ filterset_form = forms.UserFilterForm
+ table = tables.UserTable
+
+
+@register_model_view(NetBoxUser)
+class UserView(generic.ObjectView):
+ queryset = NetBoxUser.objects.all()
+ template_name = 'users/user.html'
+
+ def get_extra_context(self, request, instance):
+ changelog = ObjectChange.objects.restrict(request.user, 'view').filter(user=request.user)[:20]
+ changelog_table = ObjectChangeTable(changelog)
+
+ return {
+ 'changelog_table': changelog_table,
+ }
+
+
+@register_model_view(NetBoxUser, 'edit')
+class UserEditView(generic.ObjectEditView):
+ queryset = NetBoxUser.objects.all()
+ form = forms.UserForm
+
+
+@register_model_view(NetBoxUser, 'delete')
+class UserDeleteView(generic.ObjectDeleteView):
+ queryset = NetBoxUser.objects.all()
+
+
+class UserBulkEditView(generic.BulkEditView):
+ queryset = NetBoxUser.objects.all()
+ filterset = filtersets.UserFilterSet
+ table = tables.UserTable
+ form = forms.UserBulkEditForm
+
+
+class UserBulkImportView(generic.BulkImportView):
+ queryset = NetBoxUser.objects.all()
+ model_form = forms.UserImportForm
+
+
+class UserBulkDeleteView(generic.BulkDeleteView):
+ queryset = NetBoxUser.objects.all()
+ filterset = filtersets.UserFilterSet
+ table = tables.UserTable
+
+
+#
+# Groups
+#
+
+class GroupListView(generic.ObjectListView):
+ queryset = NetBoxGroup.objects.annotate(users_count=Count('user'))
+ filterset = filtersets.GroupFilterSet
+ filterset_form = forms.GroupFilterForm
+ table = tables.GroupTable
+
+
+@register_model_view(NetBoxGroup)
+class GroupView(generic.ObjectView):
+ queryset = NetBoxGroup.objects.all()
+ template_name = 'users/group.html'
+
+
+@register_model_view(NetBoxGroup, 'edit')
+class GroupEditView(generic.ObjectEditView):
+ queryset = NetBoxGroup.objects.all()
+ form = forms.GroupForm
+
+
+@register_model_view(NetBoxGroup, 'delete')
+class GroupDeleteView(generic.ObjectDeleteView):
+ queryset = NetBoxGroup.objects.all()
+
+
+class GroupBulkImportView(generic.BulkImportView):
+ queryset = NetBoxGroup.objects.all()
+ model_form = forms.GroupImportForm
+
+
+class GroupBulkDeleteView(generic.BulkDeleteView):
+ queryset = NetBoxGroup.objects.annotate(users_count=Count('user'))
+ filterset = filtersets.GroupFilterSet
+ table = tables.GroupTable
+
+
+#
+# ObjectPermissions
+#
+
+class ObjectPermissionListView(generic.ObjectListView):
+ queryset = ObjectPermission.objects.all()
+ filterset = filtersets.ObjectPermissionFilterSet
+ filterset_form = forms.ObjectPermissionFilterForm
+ table = tables.ObjectPermissionTable
+
+
+@register_model_view(ObjectPermission)
+class ObjectPermissionView(generic.ObjectView):
+ queryset = ObjectPermission.objects.all()
+ template_name = 'users/objectpermission.html'
+
+
+@register_model_view(ObjectPermission, 'edit')
+class ObjectPermissionEditView(generic.ObjectEditView):
+ queryset = ObjectPermission.objects.all()
+ form = forms.ObjectPermissionForm
+
+
+@register_model_view(ObjectPermission, 'delete')
+class ObjectPermissionDeleteView(generic.ObjectDeleteView):
+ queryset = ObjectPermission.objects.all()
+
+
+class ObjectPermissionBulkEditView(generic.BulkEditView):
+ queryset = ObjectPermission.objects.all()
+ filterset = filtersets.ObjectPermissionFilterSet
+ table = tables.ObjectPermissionTable
+ form = forms.ObjectPermissionBulkEditForm
+
+
+class ObjectPermissionBulkDeleteView(generic.BulkDeleteView):
+ queryset = ObjectPermission.objects.all()
+ filterset = filtersets.ObjectPermissionFilterSet
+ table = tables.ObjectPermissionTable
diff --git a/netbox/utilities/counters.py b/netbox/utilities/counters.py
new file mode 100644
index 000000000..ee6865ca2
--- /dev/null
+++ b/netbox/utilities/counters.py
@@ -0,0 +1,93 @@
+from django.apps import apps
+from django.db.models import F
+from django.db.models.signals import post_delete, post_save
+
+from netbox.registry import registry
+from .fields import CounterCacheField
+
+
+def get_counters_for_model(model):
+ """
+ Return field mappings for all counters registered to the given model.
+ """
+ return registry['counter_fields'][model].items()
+
+
+def update_counter(model, pk, counter_name, value):
+ """
+ Increment or decrement a counter field on an object identified by its model and primary key (PK). Positive values
+ will increment; negative values will decrement.
+ """
+ model.objects.filter(pk=pk).update(
+ **{counter_name: F(counter_name) + value}
+ )
+
+
+#
+# Signal handlers
+#
+
+def post_save_receiver(sender, instance, **kwargs):
+ """
+ Update counter fields on related objects when a TrackingModelMixin subclass is created or modified.
+ """
+ for field_name, counter_name in get_counters_for_model(sender):
+ parent_model = sender._meta.get_field(field_name).related_model
+ new_pk = getattr(instance, field_name, None)
+ old_pk = instance.tracker.get(field_name) if field_name in instance.tracker else None
+
+ # Update the counters on the old and/or new parents as needed
+ if old_pk is not None:
+ update_counter(parent_model, old_pk, counter_name, -1)
+ if new_pk is not None:
+ update_counter(parent_model, new_pk, counter_name, 1)
+
+
+def post_delete_receiver(sender, instance, **kwargs):
+ """
+ Update counter fields on related objects when a TrackingModelMixin subclass is deleted.
+ """
+ for field_name, counter_name in get_counters_for_model(sender):
+ parent_model = sender._meta.get_field(field_name).related_model
+ parent_pk = getattr(instance, field_name, None)
+
+ # Decrement the parent's counter by one
+ if parent_pk is not None:
+ update_counter(parent_model, parent_pk, counter_name, -1)
+
+
+#
+# Registration
+#
+
+def connect_counters(*models):
+ """
+ Register counter fields and connect post_save & post_delete signal handlers for the affected models.
+ """
+ for model in models:
+
+ # Find all CounterCacheFields on the model
+ counter_fields = [
+ field for field in model._meta.get_fields() if type(field) is CounterCacheField
+ ]
+
+ for field in counter_fields:
+ to_model = apps.get_model(field.to_model_name)
+
+ # Register the counter in the registry
+ change_tracking_fields = registry['counter_fields'][to_model]
+ change_tracking_fields[f"{field.to_field_name}_id"] = field.name
+
+ # Connect the post_save and post_delete handlers
+ post_save.connect(
+ post_save_receiver,
+ sender=to_model,
+ weak=False,
+ dispatch_uid=f'{model._meta.label}.{field.name}'
+ )
+ post_delete.connect(
+ post_delete_receiver,
+ sender=to_model,
+ weak=False,
+ dispatch_uid=f'{model._meta.label}.{field.name}'
+ )
diff --git a/netbox/utilities/fields.py b/netbox/utilities/fields.py
index 8934e4ad6..ca1342df7 100644
--- a/netbox/utilities/fields.py
+++ b/netbox/utilities/fields.py
@@ -2,6 +2,7 @@ from collections import defaultdict
from django.contrib.contenttypes.fields import GenericForeignKey
from django.db import models
+from django.utils.translation import gettext_lazy as _
from utilities.ordering import naturalize
from .forms.widgets import ColorSelect
@@ -9,6 +10,7 @@ from .validators import ColorValidator
__all__ = (
'ColorField',
+ 'CounterCacheField',
'NaturalOrderingField',
'NullableCharField',
'RestrictedGenericForeignKey',
@@ -143,3 +145,43 @@ class RestrictedGenericForeignKey(GenericForeignKey):
self.name,
False,
)
+
+
+class CounterCacheField(models.BigIntegerField):
+ """
+ Counter field to keep track of related model counts.
+ """
+ def __init__(self, to_model, to_field, *args, **kwargs):
+ if not isinstance(to_model, str):
+ raise TypeError(
+ _("%s(%r) is invalid. to_model parameter to CounterCacheField must be "
+ "a string in the format 'app.model'")
+ % (
+ self.__class__.__name__,
+ to_model,
+ )
+ )
+
+ if not isinstance(to_field, str):
+ raise TypeError(
+ _("%s(%r) is invalid. to_field parameter to CounterCacheField must be "
+ "a string in the format 'field'")
+ % (
+ self.__class__.__name__,
+ to_field,
+ )
+ )
+
+ self.to_model_name = to_model
+ self.to_field_name = to_field
+
+ kwargs['default'] = kwargs.get('default', 0)
+ kwargs['editable'] = False
+
+ super().__init__(*args, **kwargs)
+
+ def deconstruct(self):
+ name, path, args, kwargs = super().deconstruct()
+ kwargs["to_model"] = self.to_model_name
+ kwargs["to_field"] = self.to_field_name
+ return name, path, args, kwargs
diff --git a/netbox/utilities/management/__init__.py b/netbox/utilities/management/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/netbox/utilities/management/commands/__init__.py b/netbox/utilities/management/commands/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/netbox/utilities/management/commands/calculate_cached_counts.py b/netbox/utilities/management/commands/calculate_cached_counts.py
new file mode 100644
index 000000000..62354797c
--- /dev/null
+++ b/netbox/utilities/management/commands/calculate_cached_counts.py
@@ -0,0 +1,52 @@
+from collections import defaultdict
+
+from django.core.management.base import BaseCommand
+from django.db.models import Count, OuterRef, Subquery
+
+from netbox.registry import registry
+
+
+class Command(BaseCommand):
+ help = "Force a recalculation of all cached counter fields"
+
+ @staticmethod
+ def collect_models():
+ """
+ Query the registry to find all models which have one or more counter fields. Return a mapping of counter fields
+ to related query names for each model.
+ """
+ models = defaultdict(dict)
+
+ for model, field_mappings in registry['counter_fields'].items():
+ for field_name, counter_name in field_mappings.items():
+ fk_field = model._meta.get_field(field_name) # Interface.device
+ parent_model = fk_field.related_model # Device
+ related_query_name = fk_field.related_query_name() # 'interfaces'
+ models[parent_model][counter_name] = related_query_name
+
+ return models
+
+ def update_counts(self, model, field_name, related_query):
+ """
+ Perform a bulk update for the given model and counter field. For example,
+
+ update_counts(Device, '_interface_count', 'interfaces')
+
+ will effectively set
+
+ Device.objects.update(_interface_count=Count('interfaces'))
+ """
+ self.stdout.write(f'Updating {model.__name__} {field_name}...')
+ subquery = Subquery(
+ model.objects.filter(pk=OuterRef('pk')).annotate(_count=Count(related_query)).values('_count')
+ )
+ return model.objects.update(**{
+ field_name: subquery
+ })
+
+ def handle(self, *model_names, **options):
+ for model, mappings in self.collect_models().items():
+ for field_name, related_query in mappings.items():
+ self.update_counts(model, field_name, related_query)
+
+ self.stdout.write(self.style.SUCCESS('Finished.'))
diff --git a/netbox/utilities/permissions.py b/netbox/utilities/permissions.py
index b20aafce0..813a8f944 100644
--- a/netbox/utilities/permissions.py
+++ b/netbox/utilities/permissions.py
@@ -18,11 +18,10 @@ def get_permission_for_model(model, action):
:param model: A model or instance
:param action: View, add, change, or delete (string)
"""
- return '{}.{}_{}'.format(
- model._meta.app_label,
- action,
- model._meta.model_name
- )
+ # Resolve to the "concrete" model (for proxy models)
+ model = model._meta.concrete_model
+
+ return f'{model._meta.app_label}.{action}_{model._meta.model_name}'
def resolve_permission(name):
diff --git a/netbox/utilities/querysets.py b/netbox/utilities/querysets.py
index ba4b28418..50917dd0f 100644
--- a/netbox/utilities/querysets.py
+++ b/netbox/utilities/querysets.py
@@ -1,7 +1,7 @@
from django.db.models import Prefetch, QuerySet
from users.constants import CONSTRAINT_TOKEN_USER
-from utilities.permissions import permission_is_exempt, qs_filter_from_constraints
+from utilities.permissions import get_permission_for_model, permission_is_exempt, qs_filter_from_constraints
__all__ = (
'RestrictedPrefetch',
@@ -46,9 +46,7 @@ class RestrictedQuerySet(QuerySet):
:param action: The action which must be permitted (e.g. "view" for "dcim.view_site"); default is 'view'
"""
# Resolve the full name of the required permission
- app_label = self.model._meta.app_label
- model_name = self.model._meta.model_name
- permission_required = f'{app_label}.{action}_{model_name}'
+ permission_required = get_permission_for_model(self.model, action)
# Bypass restriction for superusers and exempt views
if user.is_superuser or permission_is_exempt(permission_required):
diff --git a/netbox/utilities/testing/api.py b/netbox/utilities/testing/api.py
index 8cfe1cdd7..440d662ac 100644
--- a/netbox/utilities/testing/api.py
+++ b/netbox/utilities/testing/api.py
@@ -6,7 +6,7 @@ from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
from django.urls import reverse
from django.test import override_settings
-from graphene.types import Dynamic as GQLDynamic, List as GQLList, Union as GQLUnion
+from graphene.types import Dynamic as GQLDynamic, List as GQLList, Union as GQLUnion, String as GQLString, NonNull as GQLNonNull
from rest_framework import status
from rest_framework.test import APIClient
@@ -452,6 +452,13 @@ class APIViewTestCases:
# Compile list of fields to include
fields_string = ''
for field_name, field in type_class._meta.fields.items():
+ is_string_array = False
+ if type(field.type) is GQLList:
+ if field.type.of_type is GQLString:
+ is_string_array = True
+ elif type(field.type.of_type) is GQLNonNull and field.type.of_type.of_type is GQLString:
+ is_string_array = True
+
if type(field) is GQLDynamic:
# Dynamic fields must specify a subselection
fields_string += f'{field_name} {{ id }}\n'
@@ -461,7 +468,7 @@ class APIViewTestCases:
elif type(field.type) is GQLList and inspect.isclass(field.type.of_type) and issubclass(field.type.of_type, GQLUnion):
# Union types dont' have an id or consistent values
continue
- elif type(field.type) is GQLList and field_name != 'choices':
+ elif type(field.type) is GQLList and not is_string_array:
# TODO: Come up with something more elegant
# Temporary hack to support automated testing of reverse generic relations
fields_string += f'{field_name} {{ id }}\n'
diff --git a/netbox/utilities/testing/views.py b/netbox/utilities/testing/views.py
index dc17548a2..539fe3057 100644
--- a/netbox/utilities/testing/views.py
+++ b/netbox/utilities/testing/views.py
@@ -1,5 +1,6 @@
import csv
+from django.conf import settings
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist
from django.db.models import ForeignKey
@@ -64,8 +65,15 @@ class ViewTestCases:
def test_get_object_anonymous(self):
# Make the request as an unauthenticated user
self.client.logout()
- response = self.client.get(self._get_queryset().first().get_absolute_url())
- self.assertHttpStatus(response, 200)
+ ct = ContentType.objects.get_for_model(self.model)
+ if (ct.app_label, ct.model) in settings.EXEMPT_EXCLUDE_MODELS:
+ # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users
+ with disable_warnings('django.request'):
+ response = self.client.get(self._get_queryset().first().get_absolute_url())
+ self.assertHttpStatus(response, 302)
+ else:
+ response = self.client.get(self._get_queryset().first().get_absolute_url())
+ self.assertHttpStatus(response, 200)
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_get_object_without_permission(self):
@@ -128,6 +136,7 @@ class ViewTestCases:
:form_data: Data to be used when creating a new object.
"""
form_data = {}
+ validation_excluded_fields = []
def test_create_object_without_permission(self):
@@ -146,7 +155,6 @@ class ViewTestCases:
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_create_object_with_permission(self):
- initial_count = self._get_queryset().count()
# Assign unconstrained permission
obj_perm = ObjectPermission(
@@ -161,6 +169,7 @@ class ViewTestCases:
self.assertHttpStatus(self.client.get(self._get_url('add')), 200)
# Try POST with model-level permission
+ initial_count = self._get_queryset().count()
request = {
'path': self._get_url('add'),
'data': post_data(self.form_data),
@@ -168,19 +177,19 @@ class ViewTestCases:
self.assertHttpStatus(self.client.post(**request), 302)
self.assertEqual(initial_count + 1, self._get_queryset().count())
instance = self._get_queryset().order_by('pk').last()
- self.assertInstanceEqual(instance, self.form_data)
+ self.assertInstanceEqual(instance, self.form_data, exclude=self.validation_excluded_fields)
# Verify ObjectChange creation
- objectchanges = ObjectChange.objects.filter(
- changed_object_type=ContentType.objects.get_for_model(instance),
- changed_object_id=instance.pk
- )
- self.assertEqual(len(objectchanges), 1)
- self.assertEqual(objectchanges[0].action, ObjectChangeActionChoices.ACTION_CREATE)
+ if issubclass(instance.__class__, ChangeLoggingMixin):
+ objectchanges = ObjectChange.objects.filter(
+ changed_object_type=ContentType.objects.get_for_model(instance),
+ changed_object_id=instance.pk
+ )
+ self.assertEqual(len(objectchanges), 1)
+ self.assertEqual(objectchanges[0].action, ObjectChangeActionChoices.ACTION_CREATE)
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_create_object_with_constrained_permission(self):
- initial_count = self._get_queryset().count()
# Assign constrained permission
obj_perm = ObjectPermission(
@@ -196,6 +205,7 @@ class ViewTestCases:
self.assertHttpStatus(self.client.get(self._get_url('add')), 200)
# Try to create an object (not permitted)
+ initial_count = self._get_queryset().count()
request = {
'path': self._get_url('add'),
'data': post_data(self.form_data),
@@ -214,7 +224,8 @@ class ViewTestCases:
}
self.assertHttpStatus(self.client.post(**request), 302)
self.assertEqual(initial_count + 1, self._get_queryset().count())
- self.assertInstanceEqual(self._get_queryset().order_by('pk').last(), self.form_data)
+ instance = self._get_queryset().order_by('pk').last()
+ self.assertInstanceEqual(instance, self.form_data, exclude=self.validation_excluded_fields)
class EditObjectViewTestCase(ModelViewTestCase):
"""
@@ -223,6 +234,7 @@ class ViewTestCases:
:form_data: Data to be used when updating the first existing object.
"""
form_data = {}
+ validation_excluded_fields = []
def test_edit_object_without_permission(self):
instance = self._get_queryset().first()
@@ -261,15 +273,17 @@ class ViewTestCases:
'data': post_data(self.form_data),
}
self.assertHttpStatus(self.client.post(**request), 302)
- self.assertInstanceEqual(self._get_queryset().get(pk=instance.pk), self.form_data)
+ instance = self._get_queryset().get(pk=instance.pk)
+ self.assertInstanceEqual(instance, self.form_data, exclude=self.validation_excluded_fields)
# Verify ObjectChange creation
- objectchanges = ObjectChange.objects.filter(
- changed_object_type=ContentType.objects.get_for_model(instance),
- changed_object_id=instance.pk
- )
- self.assertEqual(len(objectchanges), 1)
- self.assertEqual(objectchanges[0].action, ObjectChangeActionChoices.ACTION_UPDATE)
+ if issubclass(instance.__class__, ChangeLoggingMixin):
+ objectchanges = ObjectChange.objects.filter(
+ changed_object_type=ContentType.objects.get_for_model(instance),
+ changed_object_id=instance.pk
+ )
+ self.assertEqual(len(objectchanges), 1)
+ self.assertEqual(objectchanges[0].action, ObjectChangeActionChoices.ACTION_UPDATE)
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
def test_edit_object_with_constrained_permission(self):
@@ -297,7 +311,8 @@ class ViewTestCases:
'data': post_data(self.form_data),
}
self.assertHttpStatus(self.client.post(**request), 302)
- self.assertInstanceEqual(self._get_queryset().get(pk=instance1.pk), self.form_data)
+ instance = self._get_queryset().get(pk=instance1.pk)
+ self.assertInstanceEqual(instance, self.form_data, exclude=self.validation_excluded_fields)
# Try to edit a non-permitted object
request = {
@@ -404,8 +419,15 @@ class ViewTestCases:
def test_list_objects_anonymous(self):
# Make the request as an unauthenticated user
self.client.logout()
- response = self.client.get(self._get_url('list'))
- self.assertHttpStatus(response, 200)
+ ct = ContentType.objects.get_for_model(self.model)
+ if (ct.app_label, ct.model) in settings.EXEMPT_EXCLUDE_MODELS:
+ # Models listed in EXEMPT_EXCLUDE_MODELS should not be accessible to anonymous users
+ with disable_warnings('django.request'):
+ response = self.client.get(self._get_url('list'))
+ self.assertHttpStatus(response, 302)
+ else:
+ response = self.client.get(self._get_url('list'))
+ self.assertHttpStatus(response, 200)
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_list_objects_without_permission(self):
@@ -450,10 +472,19 @@ class ViewTestCases:
self.assertIn(instance1.get_absolute_url(), content)
self.assertNotIn(instance2.get_absolute_url(), content)
- @override_settings(EXEMPT_VIEW_PERMISSIONS=['*'])
+ @override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_export_objects(self):
url = self._get_url('list')
+ # Add model-level permission
+ obj_perm = ObjectPermission(
+ name='Test permission',
+ actions=['view']
+ )
+ obj_perm.save()
+ obj_perm.users.add(self.user)
+ obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))
+
# Test default CSV export
response = self.client.get(f'{url}?export')
self.assertHttpStatus(response, 200)
@@ -700,7 +731,7 @@ class ViewTestCases:
# Assign model-level permission
obj_perm = ObjectPermission(
name='Test permission',
- actions=['change']
+ actions=['view', 'change']
)
obj_perm.save()
obj_perm.users.add(self.user)
@@ -731,7 +762,7 @@ class ViewTestCases:
obj_perm = ObjectPermission(
name='Test permission',
constraints={attr_name: value},
- actions=['change']
+ actions=['view', 'change']
)
obj_perm.save()
obj_perm.users.add(self.user)
@@ -795,7 +826,6 @@ class ViewTestCases:
@override_settings(EXEMPT_VIEW_PERMISSIONS=[])
def test_bulk_delete_objects_with_constrained_permission(self):
- initial_count = self._get_queryset().count()
pk_list = self._get_queryset().values_list('pk', flat=True)
data = {
'pk': pk_list,
@@ -814,6 +844,7 @@ class ViewTestCases:
obj_perm.object_types.add(ContentType.objects.get_for_model(self.model))
# Attempt to bulk delete non-permitted objects
+ initial_count = self._get_queryset().count()
self.assertHttpStatus(self.client.post(self._get_url('bulk_delete'), data), 302)
self.assertEqual(self._get_queryset().count(), initial_count)
diff --git a/netbox/utilities/tests/test_counters.py b/netbox/utilities/tests/test_counters.py
new file mode 100644
index 000000000..e9561c91b
--- /dev/null
+++ b/netbox/utilities/tests/test_counters.py
@@ -0,0 +1,69 @@
+from django.test import TestCase
+
+from dcim.models import *
+from utilities.testing.utils import create_test_device
+
+
+class CountersTest(TestCase):
+ """
+ Validate the operation of dict_to_filter_params().
+ """
+ @classmethod
+ def setUpTestData(cls):
+
+ # Create devices
+ device1 = create_test_device('Device 1')
+ device2 = create_test_device('Device 2')
+
+ # Create interfaces
+ Interface.objects.create(device=device1, name='Interface 1')
+ Interface.objects.create(device=device1, name='Interface 2')
+ Interface.objects.create(device=device2, name='Interface 3')
+ Interface.objects.create(device=device2, name='Interface 4')
+
+ def test_interface_count_creation(self):
+ """
+ When a tracked object (Interface) is added the tracking counter should be updated.
+ """
+ device1, device2 = Device.objects.all()
+ self.assertEqual(device1.interface_count, 2)
+ self.assertEqual(device2.interface_count, 2)
+
+ Interface.objects.create(device=device1, name='Interface 5')
+ Interface.objects.create(device=device2, name='Interface 6')
+ device1.refresh_from_db()
+ device2.refresh_from_db()
+ self.assertEqual(device1.interface_count, 3)
+ self.assertEqual(device2.interface_count, 3)
+
+ def test_interface_count_deletion(self):
+ """
+ When a tracked object (Interface) is deleted the tracking counter should be updated.
+ """
+ device1, device2 = Device.objects.all()
+ self.assertEqual(device1.interface_count, 2)
+ self.assertEqual(device2.interface_count, 2)
+
+ Interface.objects.get(name='Interface 1').delete()
+ Interface.objects.get(name='Interface 3').delete()
+ device1.refresh_from_db()
+ device2.refresh_from_db()
+ self.assertEqual(device1.interface_count, 1)
+ self.assertEqual(device2.interface_count, 1)
+
+ def test_interface_count_move(self):
+ """
+ When a tracked object (Interface) is moved the tracking counter should be updated.
+ """
+ device1, device2 = Device.objects.all()
+ self.assertEqual(device1.interface_count, 2)
+ self.assertEqual(device2.interface_count, 2)
+
+ interface1 = Interface.objects.get(name='Interface 1')
+ interface1.device = device2
+ interface1.save()
+
+ device1.refresh_from_db()
+ device2.refresh_from_db()
+ self.assertEqual(device1.interface_count, 1)
+ self.assertEqual(device2.interface_count, 3)
diff --git a/netbox/utilities/tracking.py b/netbox/utilities/tracking.py
new file mode 100644
index 000000000..88945615b
--- /dev/null
+++ b/netbox/utilities/tracking.py
@@ -0,0 +1,78 @@
+from django.db.models.query_utils import DeferredAttribute
+
+from netbox.registry import registry
+
+
+class Tracker:
+ """
+ An ephemeral instance employed to record which tracked fields on an instance have been modified.
+ """
+ def __init__(self):
+ self._changed_fields = {}
+
+ def __contains__(self, item):
+ return item in self._changed_fields
+
+ def set(self, name, value):
+ """
+ Mark an attribute as having been changed and record its original value.
+ """
+ self._changed_fields[name] = value
+
+ def get(self, name):
+ """
+ Return the original value of a changed field. Raises KeyError if name is not found.
+ """
+ return self._changed_fields[name]
+
+ def clear(self, *names):
+ """
+ Clear any fields that were recorded as having been changed.
+ """
+ for name in names:
+ self._changed_fields.pop(name, None)
+ else:
+ self._changed_fields = {}
+
+
+class TrackingModelMixin:
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+
+ # Mark the instance as initialized, to enable our custom __setattr__()
+ self._initialized = True
+
+ @property
+ def tracker(self):
+ """
+ Return the Tracker instance for this instance, first creating it if necessary.
+ """
+ if not hasattr(self._state, "_tracker"):
+ self._state._tracker = Tracker()
+ return self._state._tracker
+
+ def save(self, *args, **kwargs):
+ super().save(*args, **kwargs)
+
+ # Clear any tracked fields now that changes have been saved
+ update_fields = kwargs.get('update_fields', [])
+ self.tracker.clear(*update_fields)
+
+ def __setattr__(self, name, value):
+ if hasattr(self, "_initialized"):
+ # Record any changes to a tracked field
+ if name in registry['counter_fields'][self.__class__]:
+ if name not in self.tracker:
+ # The attribute has been created or changed
+ if name in self.__dict__:
+ old_value = getattr(self, name)
+ if value != old_value:
+ self.tracker.set(name, old_value)
+ else:
+ self.tracker.set(name, DeferredAttribute)
+ elif value == self.tracker.get(name):
+ # A previously changed attribute has been restored
+ self.tracker.clear(name)
+
+ super().__setattr__(name, value)
diff --git a/netbox/virtualization/api/serializers.py b/netbox/virtualization/api/serializers.py
index f72215b98..693bb362f 100644
--- a/netbox/virtualization/api/serializers.py
+++ b/netbox/virtualization/api/serializers.py
@@ -80,12 +80,15 @@ class VirtualMachineSerializer(NetBoxModelSerializer):
primary_ip4 = NestedIPAddressSerializer(required=False, allow_null=True)
primary_ip6 = NestedIPAddressSerializer(required=False, allow_null=True)
+ # Counter fields
+ interface_count = serializers.IntegerField(read_only=True)
+
class Meta:
model = VirtualMachine
fields = [
'id', 'url', 'display', 'name', 'status', 'site', 'cluster', 'device', 'role', 'tenant', 'platform',
'primary_ip', 'primary_ip4', 'primary_ip6', 'vcpus', 'memory', 'disk', 'description', 'comments',
- 'local_context_data', 'tags', 'custom_fields', 'created', 'last_updated',
+ 'local_context_data', 'tags', 'custom_fields', 'created', 'last_updated', 'interface_count',
]
validators = []
@@ -98,6 +101,7 @@ class VirtualMachineWithConfigContextSerializer(VirtualMachineSerializer):
'id', 'url', 'display', 'name', 'status', 'site', 'cluster', 'device', 'role', 'tenant', 'platform',
'primary_ip', 'primary_ip4', 'primary_ip6', 'vcpus', 'memory', 'disk', 'description', 'comments',
'local_context_data', 'tags', 'custom_fields', 'config_context', 'created', 'last_updated',
+ 'interface_count',
]
@extend_schema_field(serializers.JSONField(allow_null=True))
diff --git a/netbox/virtualization/apps.py b/netbox/virtualization/apps.py
index 1b6b110df..8db943ea1 100644
--- a/netbox/virtualization/apps.py
+++ b/netbox/virtualization/apps.py
@@ -6,3 +6,8 @@ class VirtualizationConfig(AppConfig):
def ready(self):
from . import search
+ from .models import VirtualMachine
+ from utilities.counters import connect_counters
+
+ # Register counters
+ connect_counters(VirtualMachine)
diff --git a/netbox/virtualization/migrations/0035_virtualmachine_interface_count.py b/netbox/virtualization/migrations/0035_virtualmachine_interface_count.py
new file mode 100644
index 000000000..5f52d32e0
--- /dev/null
+++ b/netbox/virtualization/migrations/0035_virtualmachine_interface_count.py
@@ -0,0 +1,35 @@
+from django.db import migrations
+from django.db.models import Count
+
+import utilities.fields
+
+
+def populate_virtualmachine_counts(apps, schema_editor):
+ VirtualMachine = apps.get_model('virtualization', 'VirtualMachine')
+
+ vms = list(VirtualMachine.objects.annotate(_interface_count=Count('interfaces', distinct=True)))
+
+ for vm in vms:
+ vm.interface_count = vm._interface_count
+
+ VirtualMachine.objects.bulk_update(vms, ['interface_count'])
+
+
+class Migration(migrations.Migration):
+ dependencies = [
+ ('virtualization', '0034_standardize_description_comments'),
+ ]
+
+ operations = [
+ migrations.AddField(
+ model_name='virtualmachine',
+ name='interface_count',
+ field=utilities.fields.CounterCacheField(
+ default=0, to_field='virtual_machine', to_model='virtualization.VMInterface'
+ ),
+ ),
+ migrations.RunPython(
+ code=populate_virtualmachine_counts,
+ reverse_code=migrations.RunPython.noop
+ ),
+ ]
diff --git a/netbox/virtualization/models/virtualmachines.py b/netbox/virtualization/models/virtualmachines.py
index 6e9cc5664..dbbfe49ed 100644
--- a/netbox/virtualization/models/virtualmachines.py
+++ b/netbox/virtualization/models/virtualmachines.py
@@ -11,9 +11,10 @@ from extras.models import ConfigContextModel
from extras.querysets import ConfigContextModelQuerySet
from netbox.config import get_config
from netbox.models import NetBoxModel, PrimaryModel
-from utilities.fields import NaturalOrderingField
+from utilities.fields import CounterCacheField, NaturalOrderingField
from utilities.ordering import naturalize_interface
from utilities.query_functions import CollateAsChar
+from utilities.tracking import TrackingModelMixin
from virtualization.choices import *
__all__ = (
@@ -120,6 +121,12 @@ class VirtualMachine(PrimaryModel, ConfigContextModel):
verbose_name='Disk (GB)'
)
+ # Counter fields
+ interface_count = CounterCacheField(
+ to_model='virtualization.VMInterface',
+ to_field='virtual_machine'
+ )
+
# Generic relation
contacts = GenericRelation(
to='tenancy.ContactAssignment'
@@ -222,7 +229,7 @@ class VirtualMachine(PrimaryModel, ConfigContextModel):
return None
-class VMInterface(NetBoxModel, BaseInterface):
+class VMInterface(NetBoxModel, BaseInterface, TrackingModelMixin):
virtual_machine = models.ForeignKey(
to='virtualization.VirtualMachine',
on_delete=models.CASCADE,
diff --git a/netbox/virtualization/tables/virtualmachines.py b/netbox/virtualization/tables/virtualmachines.py
index b1d44ad02..03e3a1af6 100644
--- a/netbox/virtualization/tables/virtualmachines.py
+++ b/netbox/virtualization/tables/virtualmachines.py
@@ -1,10 +1,11 @@
import django_tables2 as tables
+from django.utils.translation import gettext as _
+
from dcim.tables.devices import BaseInterfaceTable
+from netbox.tables import NetBoxTable, columns
from tenancy.tables import ContactsColumnMixin, TenancyColumnsMixin
from virtualization.models import VirtualMachine, VMInterface
-from netbox.tables import NetBoxTable, columns
-
__all__ = (
'VirtualMachineTable',
'VirtualMachineVMInterfaceTable',
@@ -70,6 +71,9 @@ class VirtualMachineTable(TenancyColumnsMixin, ContactsColumnMixin, NetBoxTable)
tags = columns.TagColumn(
url_name='virtualization:virtualmachine_list'
)
+ interface_count = tables.Column(
+ verbose_name=_('Interfaces')
+ )
class Meta(NetBoxTable.Meta):
model = VirtualMachine
diff --git a/netbox/virtualization/views.py b/netbox/virtualization/views.py
index 75e83f9e1..c56a8ade2 100644
--- a/netbox/virtualization/views.py
+++ b/netbox/virtualization/views.py
@@ -349,7 +349,7 @@ class VirtualMachineInterfacesView(generic.ObjectChildrenView):
template_name = 'virtualization/virtualmachine/interfaces.html'
tab = ViewTab(
label=_('Interfaces'),
- badge=lambda obj: obj.interfaces.count(),
+ badge=lambda obj: obj.interface_count,
permission='virtualization.view_vminterface',
weight=500
)