Coverage for netbox/authentication/__init__.py: 18%
211 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import logging
2from collections import defaultdict
4from django.apps import apps
5from django.conf import settings
6from django.contrib.auth.backends import ModelBackend
7from django.contrib.auth.backends import RemoteUserBackend as _RemoteUserBackend
8from django.contrib.auth.models import AnonymousUser
9from django.core.exceptions import ImproperlyConfigured
10from django.db.models import Q
12from netbox.settings_utils import load_ldap_config
13from users.constants import CONSTRAINT_TOKEN_USER
14from users.models import Group, ObjectPermission, User
15from utilities.permissions import (
16 permission_is_exempt,
17 qs_filter_from_constraints,
18 resolve_permission,
19 resolve_permission_type,
20)
22from .misc import _mirror_groups
24AUTH_BACKEND_ATTRS = {
25 # backend name: title, MDI icon name
26 'amazon': ('Amazon AWS', 'aws'),
27 'apple': ('Apple', 'apple'),
28 'auth0': ('Auth0', None),
29 'azuread-oauth2': ('Microsoft Entra ID', 'microsoft'),
30 'azuread-b2c-oauth2': ('Microsoft Entra ID', 'microsoft'),
31 'azuread-tenant-oauth2': ('Microsoft Entra ID', 'microsoft'),
32 'azuread-v2-tenant-oauth2': ('Microsoft Entra ID', 'microsoft'),
33 'bitbucket': ('BitBucket', 'bitbucket'),
34 'bitbucket-oauth2': ('BitBucket', 'bitbucket'),
35 'digitalocean': ('DigitalOcean', 'digital-ocean'),
36 'docker': ('Docker', 'docker'),
37 'github': ('GitHub', 'github'),
38 'github-app': ('GitHub', 'github'),
39 'github-org': ('GitHub', 'github'),
40 'github-team': ('GitHub', 'github'),
41 'github-enterprise': ('GitHub Enterprise', 'github'),
42 'github-enterprise-org': ('GitHub Enterprise', 'github'),
43 'github-enterprise-team': ('GitHub Enterprise', 'github'),
44 'gitlab': ('GitLab', 'gitlab'),
45 'google-oauth2': ('Google', 'google'),
46 'google-openidconnect': ('Google', 'google'),
47 'hubspot': ('HubSpot', 'hubspot'),
48 'keycloak': ('Keycloak', None),
49 'microsoft-graph': ('Microsoft Graph', 'microsoft'),
50 'oidc': ('OpenID Connect', None),
51 'okta': ('Okta', None),
52 'okta-openidconnect': ('Okta (OIDC)', None),
53 'salesforce-oauth2': ('Salesforce', 'salesforce'),
54}
55# Override with potential user configuration
56AUTH_BACKEND_ATTRS.update(getattr(settings, 'SOCIAL_AUTH_BACKEND_ATTRS', {}))
59def get_auth_backend_display(name):
60 """
61 Return the user-friendly name and icon name for a remote authentication backend, if
62 known. Obtained from the defaults dictionary AUTH_BACKEND_ATTRS, overridden by the
63 setting `SOCIAL_AUTH_BACKEND_ATTRS`. Defaults to the raw backend name and no icon.
64 """
65 return AUTH_BACKEND_ATTRS.get(name, (name, None))
68def get_saml_idps():
69 return getattr(settings, "SOCIAL_AUTH_SAML_ENABLED_IDPS", {}).keys()
72class ObjectPermissionMixin:
74 def get_all_permissions(self, user_obj, obj=None):
75 if not user_obj.is_active or user_obj.is_anonymous:
76 return dict()
77 if not hasattr(user_obj, '_object_perm_cache'):
78 user_obj._object_perm_cache = self.get_object_permissions(user_obj)
79 return user_obj._object_perm_cache
81 def get_permission_filter(self, user_obj):
82 return Q(users=user_obj) | Q(groups__user=user_obj)
84 def get_object_permissions(self, user_obj):
85 """
86 Return all permissions granted to the user by an ObjectPermission.
87 """
88 # Initialize a dictionary mapping permission names to sets of constraints
89 perms = defaultdict(list)
91 # Collect any configured default permissions
92 for perm_name, constraints in settings.DEFAULT_PERMISSIONS.items():
93 constraints = constraints or tuple()
94 if type(constraints) not in (list, tuple):
95 raise ImproperlyConfigured(
96 f"Constraints for default permission {perm_name} must be defined as a list or tuple."
97 )
98 perms[perm_name].extend(constraints)
100 # Retrieve all assigned and enabled ObjectPermissions
101 object_permissions = ObjectPermission.objects.filter(
102 self.get_permission_filter(user_obj),
103 enabled=True
104 ).order_by('id').distinct('id').prefetch_related('object_types')
106 # Create a dictionary mapping permissions to their constraints
107 for obj_perm in object_permissions:
108 for object_type in obj_perm.object_types.all():
109 for action in obj_perm.actions:
110 perm_name = f"{object_type.app_label}.{action}_{object_type.model}"
111 perms[perm_name].extend(obj_perm.list_constraints())
113 return perms
115 def has_perm(self, user_obj, perm, obj=None):
116 app_label, __, model_name = resolve_permission(perm)
118 # Superusers implicitly have all permissions
119 if user_obj.is_active and user_obj.is_superuser: 119 ↛ 120line 119 didn't jump to line 120 because the condition on line 119 was never true
120 return True
122 # Permission is exempt from enforcement (i.e. listed in EXEMPT_VIEW_PERMISSIONS)
123 if permission_is_exempt(perm): 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true
124 return True
126 # Handle inactive/anonymous users
127 if not user_obj.is_active or user_obj.is_anonymous: 127 ↛ 130line 127 didn't jump to line 130 because the condition on line 127 was always true
128 return False
130 object_permissions = self.get_all_permissions(user_obj)
132 # If no applicable ObjectPermissions have been created for this user/permission, deny permission
133 if perm not in object_permissions:
134 return False
136 # If no object has been specified, grant permission. (The presence of a permission in this set tells
137 # us that the user has permission for *some* objects, but not necessarily a specific object.)
138 if obj is None:
139 return True
141 # Sanity check: the permission must apply to the object's model. Permissions may name proxy
142 # models, so compare concrete models and evaluate constraints via the permission model's manager.
143 try:
144 permission_model = apps.get_model(app_label, model_name)
145 except LookupError:
146 logger = logging.getLogger('netbox.auth.ObjectPermissionBackend')
147 logger.warning(f"Permission {perm} does not reference a valid model")
148 return False
149 if permission_model._meta.concrete_model is not obj._meta.concrete_model:
150 logger = logging.getLogger('netbox.auth.ObjectPermissionBackend')
151 logger.debug(f"Permission {perm} is not valid for {obj._meta.label_lower} objects")
152 return False
154 # Compile a QuerySet filter that matches all instances of the specified model
155 tokens = {
156 CONSTRAINT_TOKEN_USER: user_obj,
157 }
158 qs_filter = qs_filter_from_constraints(object_permissions[perm], tokens)
160 # Permission to perform the requested action on the object depends on whether the specified object matches
161 # the specified constraints. Note that this check is made against the *database* record representing the object,
162 # not the instance itself.
163 return permission_model.objects.filter(qs_filter, pk=obj.pk).exists()
166class ObjectPermissionBackend(ObjectPermissionMixin, ModelBackend):
167 pass
170class RemoteUserBackend(_RemoteUserBackend):
171 """
172 Custom implementation of Django's RemoteUserBackend which provides configuration hooks for basic customization.
173 """
174 @property
175 def create_unknown_user(self):
176 return settings.REMOTE_AUTH_AUTO_CREATE_USER
178 def configure_groups(self, user, remote_groups):
179 logger = logging.getLogger('netbox.auth.RemoteUserBackend')
181 # Assign default groups to the user
182 group_list = []
183 for name in remote_groups:
184 try:
185 group_list.append(Group.objects.get(name=name))
186 except Group.DoesNotExist:
187 if settings.REMOTE_AUTH_AUTO_CREATE_GROUPS:
188 group_list.append(Group.objects.create(name=name))
189 else:
190 logging.error(
191 f"Could not assign group {name} to remotely-authenticated user {user}: Group not found")
192 if group_list:
193 user.groups.set(group_list)
194 logger.debug(
195 f"Assigned groups to remotely-authenticated user {user}: {group_list}")
196 else:
197 user.groups.clear()
198 logger.debug(f"Stripping user {user} from Groups")
200 # Evaluate superuser status
201 user.is_superuser = self._is_superuser(user)
202 logger.debug(f"User {user} is Superuser: {user.is_superuser}")
203 logger.debug(
204 f"User {user} should be Superuser: {self._is_superuser(user)}")
206 user.save()
207 return user
209 def authenticate(self, request, remote_user, remote_groups=None):
210 """
211 The username passed as ``remote_user`` is considered trusted. Return
212 the ``User`` object with the given username. Create a new ``User``
213 object if ``create_unknown_user`` is ``True``.
214 Return None if ``create_unknown_user`` is ``False`` and a ``User``
215 object with the given username is not found in the database.
216 """
217 logger = logging.getLogger('netbox.auth.RemoteUserBackend')
218 logger.debug(
219 f"trying to authenticate {remote_user} with groups {remote_groups}")
220 if not remote_user:
221 return None
222 user = None
223 username = self.clean_username(remote_user)
225 # Note that this could be accomplished in one try-except clause, but
226 # instead we use get_or_create when creating unknown users since it has
227 # built-in safeguards for multiple threads.
228 if self.create_unknown_user:
229 user, created = User._default_manager.get_or_create(**{
230 User.USERNAME_FIELD: username
231 })
232 if created:
233 user = self.configure_user(request, user)
234 else:
235 try:
236 user = User._default_manager.get_by_natural_key(username)
237 except User.DoesNotExist:
238 pass
239 if self.user_can_authenticate(user):
240 if settings.REMOTE_AUTH_GROUP_SYNC_ENABLED:
241 if user is not None and not isinstance(user, AnonymousUser):
242 return self.configure_groups(user, remote_groups)
243 else:
244 return user
245 return None
247 def _is_superuser(self, user):
248 logger = logging.getLogger('netbox.auth.RemoteUserBackend')
249 superuser_groups = settings.REMOTE_AUTH_SUPERUSER_GROUPS
250 logger.debug(f"Superuser Groups: {superuser_groups}")
251 superusers = settings.REMOTE_AUTH_SUPERUSERS
252 logger.debug(f"Superuser Users: {superusers}")
253 user_groups = set()
254 for g in user.groups.all():
255 user_groups.add(g.name)
256 logger.debug(f"User {user.username} is in Groups:{user_groups}")
258 result = user.username in superusers or (
259 set(user_groups) & set(superuser_groups))
260 logger.debug(f"User {user.username} in Superuser Users :{result}")
261 return bool(result)
263 def _is_staff(self, user):
264 # Retain for pre-v4.5 compatibility
265 return user.is_superuser
267 def configure_user(self, request, user):
268 logger = logging.getLogger('netbox.auth.RemoteUserBackend')
269 if not settings.REMOTE_AUTH_GROUP_SYNC_ENABLED:
270 # Assign default groups to the user
271 group_list = []
272 for name in settings.REMOTE_AUTH_DEFAULT_GROUPS:
273 try:
274 group_list.append(Group.objects.get(name=name))
275 except Group.DoesNotExist:
276 logging.error(
277 f"Could not assign group {name} to remotely-authenticated user {user}: Group not found")
278 if group_list:
279 user.groups.add(*group_list)
280 logger.debug(
281 f"Assigned groups to remotely-authenticated user {user}: {group_list}")
283 # Assign default object permissions to the user
284 permissions_list = []
285 for permission_name, constraints in settings.REMOTE_AUTH_DEFAULT_PERMISSIONS.items():
286 try:
287 object_type, action = resolve_permission_type(permission_name)
288 # TODO: Merge multiple actions into a single ObjectPermission per object type
289 obj_perm = ObjectPermission(actions=[action], constraints=constraints)
290 obj_perm.save()
291 obj_perm.users.add(user)
292 obj_perm.object_types.add(object_type)
293 permissions_list.append(permission_name)
294 except ValueError:
295 logging.error(
296 f"Invalid permission name: '{permission_name}'. Permissions must be in the form "
297 "<app>.<action>_<model>. (Example: dcim.add_site)"
298 )
299 if permissions_list:
300 logger.debug(
301 f"Assigned permissions to remotely-authenticated user {user}: {permissions_list}")
302 else:
303 logger.debug(
304 f"Skipped initial assignment of permissions and groups to remotely-authenticated user {user} as "
305 f"Group sync is enabled"
306 )
308 return user
310 def has_perm(self, user_obj, perm, obj=None):
311 return False
314# Create a new instance of django-auth-ldap's LDAPBackend with our own ObjectPermissions
315try:
316 from django_auth_ldap.backend import LDAPBackend as LDAPBackend_
317 from django_auth_ldap.backend import _LDAPUser
319 class NBLDAPBackend(ObjectPermissionMixin, LDAPBackend_):
320 def get_permission_filter(self, user_obj):
321 permission_filter = super().get_permission_filter(user_obj)
322 if (self.settings.FIND_GROUP_PERMS and
323 hasattr(user_obj, "ldap_user") and
324 hasattr(user_obj.ldap_user, "group_names")):
325 permission_filter = permission_filter | Q(groups__name__in=user_obj.ldap_user.group_names)
326 return permission_filter
328 # Patch with our modified _mirror_groups() method to support our custom Group model
329 _LDAPUser._mirror_groups = _mirror_groups
331except ModuleNotFoundError:
332 pass
335class LDAPBackend:
337 def __new__(cls, *args, **kwargs):
338 try:
339 import ldap
340 from django_auth_ldap.backend import LDAPSettings
341 except ModuleNotFoundError as e:
342 if getattr(e, 'name') == 'django_auth_ldap':
343 raise ImproperlyConfigured(
344 "LDAP authentication has been configured, but django-auth-ldap is not installed."
345 )
346 raise e
348 ldap_config = load_ldap_config(
349 settings.CONFIGURATION_DIR,
350 allow_legacy_fallback=settings.NETBOX_INSTALL_MODE == 'checkout',
351 )
353 try:
354 getattr(ldap_config, 'AUTH_LDAP_SERVER_URI')
355 except AttributeError:
356 raise ImproperlyConfigured(
357 "Required parameter AUTH_LDAP_SERVER_URI is missing from ldap_config.py."
358 )
360 obj = NBLDAPBackend()
362 # Read LDAP configuration parameters from ldap_config.py instead of settings.py
363 ldap_settings = LDAPSettings()
364 for param in dir(ldap_config):
365 if param.startswith(ldap_settings._prefix):
366 setattr(ldap_settings, param[10:], getattr(ldap_config, param))
367 obj.settings = ldap_settings
369 # Optionally disable strict certificate checking
370 if getattr(ldap_config, 'LDAP_IGNORE_CERT_ERRORS', False):
371 ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
373 # Optionally set CA cert directory
374 if ca_cert_dir := getattr(ldap_config, 'LDAP_CA_CERT_DIR', None):
375 ldap.set_option(ldap.OPT_X_TLS_CACERTDIR, ca_cert_dir)
377 # Optionally set CA cert file
378 if ca_cert_file := getattr(ldap_config, 'LDAP_CA_CERT_FILE', None):
379 ldap.set_option(ldap.OPT_X_TLS_CACERTFILE, ca_cert_file)
381 return obj
384# Custom Social Auth Pipeline Handlers
385def user_default_groups_handler(backend, user, response, *args, **kwargs):
386 """
387 Custom pipeline handler which adds remote auth users to the default group specified in the
388 configuration file.
389 """
390 logger = logging.getLogger('netbox.auth.user_default_groups_handler')
391 if settings.REMOTE_AUTH_DEFAULT_GROUPS:
392 # Assign default groups to the user
393 group_list = []
394 for name in settings.REMOTE_AUTH_DEFAULT_GROUPS:
395 try:
396 group_list.append(Group.objects.get(name=name))
397 except Group.DoesNotExist:
398 logging.error(
399 f"Could not assign group {name} to remotely-authenticated user {user}: Group not found")
400 if group_list:
401 user.groups.add(*group_list)
402 else:
403 logger.info(f"No valid group assignments for {user} - REMOTE_AUTH_DEFAULT_GROUPS may be incorrectly set?")