Coverage for netbox_data_import/object

1# SPDX-License-Identifier: Apache-2.0

3"""Shared object-permission checks for import writes.

5NetBox grants ``user.has_perm("app.add_thing")`` when the user may act on *any* object of that

6type. An ObjectPermission's constraints are only evaluated against a saved instance, so a write

7is inside the caller's scope only once the saved object passes ``has_perm(permission, instance)``.

9Every scoped write goes through this module. Denial raises rather than returning a flag: a caller

10can ignore a False and return from an enclosing ``atomic()`` block, which commits the very write

11the denial was meant to prevent.

12"""

14from dataclasses import dataclass

15from typing import Any, Literal

17from django.core.exceptions import ValidationError

18from django.db import IntegrityError, transaction

19from utilities.permissions import get_permission_for_model

22class ObjectPermissionDenied(Exception):

23 """Reject a write outside the caller's NetBox object scope."""

26@dataclass(frozen=True)

27class PermissionScopedSaveResult:

28 """One scoped write: the saved object and whether this call created it."""

30 instance: Any

31 created: bool

34def enforce_saved_object_permission(obj, user, action):

35 """Reject a saved object whose final state is outside the user's scope.

37 ``has_perm`` rather than ``restrict()``: this decides one already-saved instance, and it holds

38 for any model rather than only those whose manager is a ``RestrictedQuerySet``.

39 """

40 if user is None:

41 return

42 permission = get_permission_for_model(type(obj), action)

43 if not user.has_perm(permission, obj):

44 raise ObjectPermissionDenied(permission)

47def reject_overlong_fields(instance, model):

48 """Reject a string the column cannot hold, before the database raises DataError.

50 Only length is checked. `full_clean` would also refuse a field that is legitimately blank.

51 """

52 for field in model._meta.concrete_fields:

53 max_length = getattr(field, "max_length", None)

54 value = getattr(instance, field.attname, None)

55 if max_length and isinstance(value, str) and len(value) > max_length:

56 raise ValidationError(f"{model._meta.verbose_name} {field.name} cannot exceed {max_length} characters.")

59def save_or_refetch(instance, model, lookup):

60 """Save *instance*, or refetch a row that won the same concurrent insert."""

61 try:

62 with transaction.atomic():

63 instance.save()

64 except IntegrityError:

65 existing = model.objects.filter(**lookup).first()

66 if existing is None:

67 raise

68 return existing, False

69 return instance, True

72def _policy_profile_id(model, lookup: dict) -> int | None:

73 """Return the profile a policy write belongs to, or None when the model carries no policy."""

74 from .models import PolicySectionModel

76 if not (isinstance(model, type) and issubclass(model, PolicySectionModel)):

77 return None

78 profile = lookup.get("profile")

79 profile_id = lookup.get("profile_id") if profile is None else profile.pk

80 if profile_id is None:

81 # A caller that names no profile cannot be serialized against an import, so it is a bug.

82 raise ValueError(f"A {model._meta.verbose_name} write must name the profile it belongs to.")

83 return profile_id

86def save_permission_scoped_object(

87 user,

88 model,

89 lookup: dict,

90 values: dict,

91 *,

92 on_existing: Literal["update", "keep", "reject"] = "update",

93) -> PermissionScopedSaveResult:

94 """Create or update one object within the user's NetBox permission scope.

96 ``on_existing`` decides what an already-present row means: ``update`` writes *values* under the

97 change permission, ``keep`` returns it untouched under the view permission, and ``reject``

98 refuses it. Raises ``ObjectPermissionDenied`` when any of those checks fails.

100 A policy write also holds its import profile, so it cannot commit inside an execution that has

101 already replanned against the old policy.

102 """

103 from .models import locked_profile_policy

104

105 profile_id = _policy_profile_id(model, lookup)

106 if profile_id is None:

107 return _scoped_write(user, model, lookup, values, on_existing=on_existing)

108 with locked_profile_policy(profile_id):

109 # atomic-exit-safe: policy-write-committed

110 return _scoped_write(user, model, lookup, values, on_existing=on_existing)

111

112

113def _scoped_write(

114 user,

115 model,

116 lookup: dict,

117 values: dict,

118 *,

119 on_existing: Literal["update", "keep", "reject"],

120) -> PermissionScopedSaveResult:

121 """Write one object and prove its saved state stays inside the user's object scope."""

122 with transaction.atomic():

123 instance = model.objects.select_for_update().filter(**lookup).first()

124 if instance is None:

125 permission = get_permission_for_model(model, "add")

126 if user is not None and not user.has_perm(permission):

127 raise ObjectPermissionDenied(permission)

128 instance = model(**lookup, **values)

129 reject_overlong_fields(instance, model)

130 instance, created = save_or_refetch(instance, model, lookup)

131 if not created:

132 instance = model.objects.select_for_update().get(pk=instance.pk)

133 else:

134 created = False

135

136 if not created:

137 if on_existing == "keep":

138 # Reusing someone else's row still exposes it, so it needs the view permission.

139 enforce_saved_object_permission(instance, user, "view")

140 # atomic-exit-safe: existing-row-kept-unwritten

141 return PermissionScopedSaveResult(instance=instance, created=False)

142 if on_existing == "reject":

143 raise ObjectPermissionDenied(get_permission_for_model(model, "add"))

144 # Before, so a row outside the user's scope cannot be taken over.

145 enforce_saved_object_permission(instance, user, "change")

146 for field_name, value in values.items():

147 setattr(instance, field_name, value)

148 reject_overlong_fields(instance, model)

149 instance.save(update_fields=list(values))

150 # After, so the new state cannot be moved outside the user's scope.

151 enforce_saved_object_permission(instance, user, "add" if created else "change")

152 # atomic-exit-safe: scoped-write-committed

153 return PermissionScopedSaveResult(instance=instance, created=created)

154

155

156def delete_permission_scoped_objects(user, queryset) -> int:

157 """Delete every row of *queryset* the user may delete, or none of them.

158

159 The rows are locked and checked one by one first, so a refusal leaves the whole set intact.

160 """

161 with transaction.atomic():

162 rows = list(queryset.select_for_update())

163 for row in rows:

164 enforce_saved_object_permission(row, user, "delete")

165 for row in rows:

166 row.delete()

167 # atomic-exit-safe: scoped-delete-committed

168 return len(rows)

Coverage for netbox_data_import/object_permissions.py: 99%

82 statements