Coverage for netbox_data_import/object_permissions.py: 95.38%
390 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-30 09:47 +0000
1# SPDX-License-Identifier: Apache-2.0
2# Copyright (C) 2026 Marcin Zieba <marcinpsk@gmail.com>
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"""
14import logging
15from collections.abc import Mapping
16from copy import copy
17from dataclasses import dataclass
18from typing import Any, Literal
20from django.core.exceptions import EmptyResultSet, FieldDoesNotExist, FieldError, ValidationError
21from django.db import DatabaseError, IntegrityError, connection, models, transaction
22from django.db.models.expressions import Col
23from django.db.models.lookups import IsNull
24from django.utils.functional import LazyObject, empty
25from users.constants import CONSTRAINT_TOKEN_USER
26from utilities.permissions import get_permission_for_model, qs_filter_from_constraints
29logger = logging.getLogger(__name__)
31_USER_PERMISSION_CACHE_ATTRIBUTES = (
32 "_perm_cache",
33 "_user_perm_cache",
34 "_group_perm_cache",
35 "_object_perm_cache",
36)
39# A workspace refuses a blind write to any policy row its actor cannot view, Cable or Location alike.
40POLICY_WRITE_REFUSED = "You cannot change a policy you cannot view."
43class ObjectPermissionDenied(Exception):
44 """Reject a write outside the caller's NetBox object scope."""
47def clear_user_permission_caches(user) -> None:
48 """Discard cached grants so the next permission check reads current policy."""
49 if user is None:
50 return
51 cache_owners = [user]
52 if isinstance(user, LazyObject):
53 if user._wrapped is empty:
54 user._setup()
55 cache_owners.append(user._wrapped)
56 for owner in cache_owners:
57 for attribute in _USER_PERMISSION_CACHE_ATTRIBUTES:
58 owner.__dict__.pop(attribute, None)
61@dataclass(frozen=True)
62class PermissionScopedSaveResult:
63 """One scoped write: the saved object and whether this call created it."""
65 instance: Any
66 created: bool
69@dataclass(frozen=True)
70class PermissionScopedSaveAssessment:
71 """Whether one prospective create or update stays inside the actor's object scope."""
73 allowed: bool
74 permission: str
77@dataclass(frozen=True)
78class ProspectiveRelation:
79 """One final related row, including concrete values that use generated keys."""
81 instance: models.Model
82 generated_fields: frozenset[str] = frozenset()
85@dataclass(frozen=True)
86class _PreparedCandidate:
87 """One related candidate row shared by forward relations."""
89 instance: models.Model
90 generated_primary_key: bool
91 generated_fields: frozenset[str]
94@dataclass(frozen=True)
95class _ProspectiveRelationState:
96 """One connected forward relation in the read-only database world."""
98 field: models.ForeignKey
99 instance: models.Model
100 generated_primary_key: bool
101 generated_target_value: bool
102 generated_fields: frozenset[str]
105@dataclass(frozen=True)
106class _ProspectiveWorld:
107 """Copied root and relation rows that one read-only assessment can query."""
109 root: Any
110 candidates: dict
111 relations: dict
112 generated_values: dict[type[models.Model], dict[Any, set[str]]]
115def _synthetic_primary_key(instance, occupied=None):
116 """Return a collision-free primary key without consuming the model's sequence."""
117 if instance.pk is not None:
118 return instance.pk, False
119 field = instance._meta.pk
120 if not isinstance(field, (models.AutoField, models.BigAutoField, models.SmallAutoField)):
121 raise TypeError(f"Cannot assess an unsaved {instance._meta.label} without a primary key.")
122 concrete_model = instance._meta.concrete_model
123 used = set() if occupied is None else occupied.setdefault(concrete_model, set())
124 value = -1
125 while value in used or concrete_model._base_manager.filter(pk=value).exists():
126 value -= 1
127 used.add(value)
128 return value, True
131def _generated_root_fields(world) -> set[str]:
132 """Return the root row's own field names whose values this world had to invent."""
133 return world.generated_values.get(world.root._meta.concrete_model, {}).get(world.root.pk, set())
136def _depends_on_generated_root_primary_key(constraint, instance) -> bool:
137 """Return whether the root predicate needs an automatic primary key not yet allocated."""
138 field = instance._meta.pk
139 names = {"pk", field.name, field.attname}
140 for key, value in constraint.items():
141 parts = key.split("__")
142 if parts[0] not in names:
143 continue
144 suffix = parts[1:]
145 if suffix == ["isnull"] or ((not suffix or suffix == ["exact"]) and value is None):
146 continue
147 return True
148 return False
151def _generated_value_aliases(query, generated_values) -> set[str]:
152 """Return aliases whose predicates or outward joins must not see generated values."""
153 fields_by_model = {
154 candidate_model: set().union(*rows.values()) for candidate_model, rows in generated_values.items()
155 }
156 candidate_tables = {candidate_model._meta.db_table for candidate_model in generated_values}
158 def field_is_generated(field) -> bool:
159 """Return whether any candidate carries an unknown value for this concrete field."""
160 model = getattr(field, "model", None)
161 return model is not None and getattr(field, "attname", None) in fields_by_model.get(
162 model._meta.concrete_model, set()
163 )
165 aliases = set()
166 pending = [query.where]
167 while pending:
168 node = pending.pop()
169 pending.extend(getattr(node, "children", ()))
170 if isinstance(node, IsNull):
171 continue
172 expression = getattr(node, "lhs", None)
173 while expression is not None and not isinstance(expression, Col):
174 expression = getattr(expression, "lhs", None)
175 if (
176 isinstance(expression, Col)
177 and expression.alias != query.base_table
178 and field_is_generated(expression.target)
179 ):
180 aliases.add(expression.alias)
181 for alias, join in query.alias_map.items():
182 parent_alias = getattr(join, "parent_alias", None)
183 if parent_alias is None:
184 continue
185 parent = query.alias_map[parent_alias]
186 for parent_field, related_field in getattr(join, "join_fields", ()):
187 if field_is_generated(parent_field) and join.table_name not in candidate_tables:
188 aliases.add(parent_alias)
189 if field_is_generated(related_field) and parent.table_name not in candidate_tables:
190 aliases.add(alias)
191 return aliases
194def _is_nullness_lookup(parts, value) -> bool:
195 """Return whether a lookup asks only whether a key or relation exists."""
196 return parts == ["isnull"] or ((not parts or parts == ["exact"]) and value is None)
199def _depends_on_generated_relation_value(constraint, relations) -> bool:
200 """Return whether a predicate treats a generated related value as authoritative."""
201 for key, value in constraint.items():
202 parts = key.split("__")
203 for relation_name, relation in relations.items():
204 field = relation.field
205 if parts[0] not in {relation_name, field.attname}:
206 continue
207 suffix = parts[1:]
208 if relation.generated_target_value:
209 if _is_nullness_lookup(suffix, value):
210 continue
211 if parts[0] == field.attname or not suffix or suffix[0] in field.get_lookups():
212 return True
213 if relation.generated_primary_key and suffix:
214 related_pk = relation.instance._meta.pk
215 if suffix[0] in {"pk", related_pk.name, related_pk.attname} and not _is_nullness_lookup(
216 suffix[1:], value
217 ):
218 return True
219 if suffix and suffix[0] in relation.generated_fields and not _is_nullness_lookup(suffix[1:], value):
220 return True
221 return False
224def _uses_root_reverse_relation(constraint, model) -> bool:
225 """Return whether a predicate needs root-owned rows this assessment does not simulate."""
226 for key in constraint:
227 current_model = model
228 for part in key.split("__"):
229 try:
230 field = current_model._meta.get_field(part)
231 except FieldDoesNotExist:
232 break
233 reverse = field.is_relation and (field.one_to_many or (field.auto_created and not field.concrete))
234 if current_model._meta.concrete_model is model._meta.concrete_model and reverse:
235 return True
236 if not field.is_relation or field.related_model is None:
237 break
238 current_model = field.related_model
239 return False
242def _normalized_generated_fields(relation, instance) -> frozenset[str]:
243 """Return generated concrete field attribute names, or reject an unknown field."""
244 concrete_fields = {
245 name: field.attname for field in instance._meta.concrete_fields for name in (field.name, field.attname)
246 }
247 unknown = relation.generated_fields - concrete_fields.keys()
248 if unknown:
249 names = ", ".join(sorted(unknown))
250 raise TypeError(f"The prospective {instance._meta.label} has unknown generated fields: {names}.")
251 return frozenset(concrete_fields[name] for name in relation.generated_fields)
254def _concrete_row_values(instance):
255 """Return the complete concrete state used to detect conflicting candidate rows."""
256 return tuple(field.value_from_object(instance) for field in instance._meta.concrete_fields)
259def _prospective_relation_is_unsaved(instance, expected_model) -> bool:
260 """Validate the model save state and return whether it needs a synthetic key."""
261 unsaved = instance.pk is None and instance._state.adding
262 saved = instance.pk is not None and not instance._state.adding
263 if not unsaved and not saved:
264 raise ValueError(f"The prospective {instance._meta.label} has an invalid saved state.")
265 if saved and not expected_model._base_manager.filter(pk=instance.pk).exists():
266 raise ValueError(f"The prospective {instance._meta.label} row does not exist.")
267 return unsaved
270def _prepare_prospective_world(instance, prospective_relations):
271 """Copy and connect the root row and each final forward relation."""
272 root = copy(instance)
273 occupied = {}
274 root.pk, root_generated_key = _synthetic_primary_key(root, occupied)
275 candidates = {root._meta.concrete_model: [root]}
276 generated_values = {root._meta.concrete_model: {root.pk: {root._meta.pk.attname}}} if root_generated_key else {}
277 relations = {}
278 prepared_by_key = {}
279 prepared_unsaved = {}
280 for relation_name, relation_value in (prospective_relations or {}).items():
281 try:
282 field = root._meta.get_field(relation_name)
283 except FieldDoesNotExist as exc:
284 raise TypeError(f"{root._meta.label}.{relation_name} is not a model field.") from exc
285 if not isinstance(field, models.ForeignKey) or field.auto_created or not field.concrete:
286 raise TypeError(f"{root._meta.label}.{relation_name} is not a concrete forward relation.")
287 relation = (
288 relation_value if isinstance(relation_value, ProspectiveRelation) else ProspectiveRelation(relation_value)
289 )
290 related_instance = relation.instance
291 if not isinstance(related_instance, models.Model):
292 raise TypeError(f"The prospective {root._meta.label}.{relation_name} value is not a model row.")
293 expected_model = field.remote_field.model._meta.concrete_model
294 if related_instance._meta.concrete_model is not expected_model:
295 raise TypeError(f"{root._meta.label}.{relation_name} does not reference {related_instance._meta.label}.")
296 generated_fields = _normalized_generated_fields(relation, related_instance)
297 unsaved = _prospective_relation_is_unsaved(related_instance, expected_model)
299 source_key = (expected_model, id(related_instance))
300 prepared = prepared_unsaved.get(source_key) if unsaved else None
301 if prepared is None:
302 related = copy(related_instance)
303 related.pk, generated_key = _synthetic_primary_key(related, occupied)
304 candidate_key = (expected_model, related.pk)
305 row_values = _concrete_row_values(related)
306 existing = prepared_by_key.get(candidate_key)
307 if existing is not None:
308 prepared, existing_values = existing
309 if existing_values != row_values or prepared.generated_fields != generated_fields:
310 raise ValueError(f"Prospective {related._meta.label} rows with key {related.pk} conflict.")
311 else:
312 prepared = _PreparedCandidate(related, generated_key, generated_fields)
313 prepared_by_key[candidate_key] = (prepared, row_values)
314 candidates.setdefault(expected_model, []).append(related)
315 if unsaved:
316 prepared_unsaved[source_key] = prepared
317 elif prepared.generated_fields != generated_fields:
318 raise ValueError(f"The prospective {related_instance._meta.label} generated fields conflict.")
320 related = prepared.instance
321 related_generated_fields = set(prepared.generated_fields)
322 if prepared.generated_primary_key:
323 related_generated_fields.add(related._meta.pk.attname)
324 if related_generated_fields:
325 generated_values.setdefault(expected_model, {}).setdefault(related.pk, set()).update(
326 related_generated_fields
327 )
328 relation_key = field.target_field.value_from_object(related)
329 generated_target_value = field.target_field.attname in related_generated_fields
330 current_key = getattr(root, field.attname)
331 if current_key is not None and current_key != relation_key:
332 raise ValueError(f"The prospective {root._meta.label} already names a different {relation_name}.")
333 setattr(root, field.attname, relation_key)
334 if generated_target_value:
335 generated_values.setdefault(root._meta.concrete_model, {}).setdefault(root.pk, set()).add(field.attname)
336 relations[relation_name] = _ProspectiveRelationState(
337 field,
338 related,
339 prepared.generated_primary_key,
340 generated_target_value,
341 prepared.generated_fields,
342 )
343 return _ProspectiveWorld(root, candidates, relations, generated_values)
346def _prospective_ctes(candidates, generated_values):
347 """Return typed candidate and database-world CTEs for each participating model."""
348 quote = connection.ops.quote_name
349 clauses = []
350 params = []
351 aliases = {}
352 for index, (candidate_model, rows) in enumerate(candidates.items()):
353 fields = candidate_model._meta.concrete_fields
354 field_types = [field.db_type(connection) for field in fields]
355 if any(field_type is None for field_type in field_types):
356 raise ValueError(f"Cannot assess every concrete field of {candidate_model._meta.label}.")
357 candidate_cte = f"ndi_permission_candidates_{index}"
358 world_cte = f"ndi_permission_world_{index}"
359 known_value_world_cte = f"ndi_permission_known_value_world_{index}"
360 if candidate_model._meta.db_table in (candidate_cte, world_cte, known_value_world_cte):
361 raise ValueError("A model table conflicts with an internal permission query name.")
362 columns = ", ".join(quote(field.column) for field in fields)
363 typed_row = "(" + ", ".join(f"CAST(%s AS {field_type})" for field_type in field_types) + ")"
364 row_values = ", ".join(typed_row for _row in rows)
365 quoted_candidate = quote(candidate_cte)
366 quoted_world = quote(world_cte)
367 quoted_known_value_world = quote(known_value_world_cte)
368 quoted_table = quote(candidate_model._meta.db_table)
369 quoted_pk = quote(candidate_model._meta.pk.column)
370 key_placeholders = ", ".join("%s" for _row in rows)
371 candidate_clause = f"{quoted_candidate} ({columns}) AS (VALUES {row_values})"
372 world_clause = (
373 f"{quoted_world} ({columns}) AS (" # noqa: S608 - Every identifier is quoted model metadata; values are parameters.
374 f"SELECT {columns} FROM {quoted_table} WHERE {quoted_pk} NOT IN ({key_placeholders}) "
375 f"UNION ALL SELECT {columns} FROM {quoted_candidate})"
376 )
377 clauses.extend(
378 (
379 candidate_clause,
380 world_clause,
381 )
382 )
383 params.extend(
384 field.get_db_prep_save(field.value_from_object(row), connection) for row in rows for field in fields
385 )
386 params.extend(row.pk for row in rows)
387 model_generated_keys = set(generated_values.get(candidate_model, {}))
388 if model_generated_keys:
389 generated_placeholders = ", ".join("%s" for _key in model_generated_keys)
390 known_value_world_clause = (
391 f"{quoted_known_value_world} ({columns}) AS (" # noqa: S608 - Identifiers are quoted metadata.
392 f"SELECT {columns} FROM {quoted_table} WHERE {quoted_pk} NOT IN ({key_placeholders}) "
393 f"UNION ALL SELECT {columns} FROM {quoted_candidate} "
394 f"WHERE {quoted_pk} NOT IN ({generated_placeholders}))"
395 )
396 clauses.append(known_value_world_clause)
397 params.extend(row.pk for row in rows)
398 params.extend(model_generated_keys)
399 known_value_world = known_value_world_cte
400 else:
401 known_value_world = world_cte
402 aliases[candidate_model._meta.db_table] = (candidate_cte, world_cte, known_value_world)
403 return clauses, params, aliases
406def _prospective_row_matches(user, model, constraint, world) -> bool:
407 """Evaluate one NetBox constraint arm against a read-only prospective database world."""
408 if (
409 _uses_root_reverse_relation(constraint, model)
410 or (
411 # A planned relation marks the root's foreign key generated, never its primary key.
412 world.root._meta.pk.attname in _generated_root_fields(world)
413 and _depends_on_generated_root_primary_key(constraint, world.root)
414 )
415 or _depends_on_generated_relation_value(constraint, world.relations)
416 ):
417 return False
418 permission_filter = qs_filter_from_constraints([constraint], {CONSTRAINT_TOKEN_USER: user})
419 queryset = model.objects.filter(permission_filter, pk=world.root.pk).values_list("pk", flat=True).order_by()
420 query = queryset.query.clone()
421 known_value_aliases = _generated_value_aliases(query, world.generated_values)
422 if query.base_table in known_value_aliases:
423 return False
424 clauses, cte_params, aliases = _prospective_ctes(world.candidates, world.generated_values)
425 root_candidate_cte = aliases[model._meta.db_table][0]
426 for alias, join in tuple(query.alias_map.items()):
427 cte_names = aliases.get(join.table_name)
428 if cte_names is None:
429 continue
430 replacement = copy(join)
431 if alias == query.base_table:
432 replacement.table_name = root_candidate_cte
433 else:
434 replacement.table_name = cte_names[2] if alias in known_value_aliases else cte_names[1]
435 query.alias_map[alias] = replacement
436 sql, params = query.sql_with_params()
437 # Model metadata supplies every identifier and type. Values remain query parameters.
438 statement = f"WITH {', '.join(clauses)} {sql}"
439 with connection.cursor() as cursor:
440 cursor.execute(statement, [*cte_params, *params])
441 return cursor.fetchone() is not None
444def _assess_permission_scoped_save(
445 user,
446 model,
447 lookup: dict,
448 values: dict,
449 *,
450 on_existing: Literal["update", "keep", "reject"],
451 current,
452 prospective_relations: Mapping[str, models.Model | ProspectiveRelation] | None = None,
453 unknown_fields: frozenset[str] = frozenset(),
454) -> PermissionScopedSaveAssessment:
455 """Assess one known current row without locking or writing it."""
456 if current is not None and on_existing == "reject":
457 return PermissionScopedSaveAssessment(False, get_permission_for_model(model, "add"))
458 action = "add" if current is None else "view" if on_existing == "keep" else "change"
459 permission = get_permission_for_model(model, action)
460 if user is None or user.is_superuser:
461 return PermissionScopedSaveAssessment(True, permission)
462 try:
463 if not user.has_perm(permission):
464 return PermissionScopedSaveAssessment(False, permission)
465 if current is not None and not user.has_perm(permission, current):
466 return PermissionScopedSaveAssessment(False, permission)
467 if on_existing == "keep" and current is not None:
468 return PermissionScopedSaveAssessment(True, permission)
469 prospective = model(**lookup, **values) if current is None else copy(current)
470 if current is not None:
471 for field_name, value in values.items():
472 setattr(prospective, field_name, value)
473 constraints = getattr(user, "_object_perm_cache", {}).get(permission, ())
474 if unknown_fields:
475 fields = {name: field for field in model._meta.concrete_fields for name in (field.name, field.attname)}
476 try:
477 unknown_roots = {root for name in unknown_fields for root in (fields[name].name, fields[name].attname)}
478 except KeyError as exc:
479 raise TypeError(f"{model._meta.label} has no concrete field named {exc.args[0]}.") from exc
480 # A null or empty constraint is an unconstrained grant, which has no field to strip.
481 constraints = tuple(
482 {key: value for key, value in constraint.items() if key.split("__", maxsplit=1)[0] not in unknown_roots}
483 if constraint
484 else constraint
485 for constraint in constraints
486 )
487 # An unconstrained grant admits every row, so only relations to validate need the prospective world.
488 if prospective_relations is None and any(not constraint for constraint in constraints):
489 return PermissionScopedSaveAssessment(True, permission)
490 world = _prepare_prospective_world(prospective, prospective_relations)
491 allowed = any(
492 not constraint or _prospective_row_matches(user, model, constraint, world) for constraint in constraints
493 )
494 except (DatabaseError, EmptyResultSet, FieldError, TypeError, ValidationError, ValueError) as exc:
495 logger.warning("Prospective %s permission assessment failed closed: %s", model._meta.label, exc)
496 allowed = False
497 return PermissionScopedSaveAssessment(allowed, permission)
500def assess_permission_scoped_save(
501 user,
502 model,
503 lookup: dict,
504 values: dict,
505 *,
506 on_existing: Literal["update", "keep", "reject"] = "update",
507 prospective_relations: Mapping[str, models.Model | ProspectiveRelation] | None = None,
508) -> PermissionScopedSaveAssessment:
509 """Assess a save and its final forward relations without writing or locking."""
510 current = model.objects.filter(**lookup).first()
511 return _assess_permission_scoped_save(
512 user,
513 model,
514 lookup,
515 values,
516 on_existing=on_existing,
517 current=current,
518 prospective_relations=prospective_relations,
519 )
522def assess_permission_scoped_save_option(
523 user,
524 model,
525 lookup: dict,
526 values: dict,
527 *,
528 unknown_fields: set[str] | frozenset[str],
529 on_existing: Literal["update", "keep", "reject"] = "update",
530 prospective_relations: Mapping[str, models.Model | ProspectiveRelation] | None = None,
531) -> PermissionScopedSaveAssessment:
532 """Assess whether known values permit a UI option that supplies the remaining values.
534 This check ignores permission predicates rooted at ``unknown_fields``. It controls whether the
535 UI offers a choice. It does not authorize the later save, which must assess every final value.
536 """
537 return assess_loaded_save_option(
538 user,
539 model,
540 lookup,
541 values,
542 current=model.objects.filter(**lookup).first(),
543 unknown_fields=unknown_fields,
544 on_existing=on_existing,
545 prospective_relations=prospective_relations,
546 )
549def assess_loaded_save_option(
550 user,
551 model,
552 lookup: dict,
553 values: dict,
554 *,
555 current,
556 unknown_fields: set[str] | frozenset[str],
557 on_existing: Literal["update", "keep", "reject"] = "update",
558 prospective_relations: Mapping[str, models.Model | ProspectiveRelation] | None = None,
559) -> PermissionScopedSaveAssessment:
560 """Assess a UI option like `assess_permission_scoped_save_option`, against a row the caller already read.
562 *current* is the row *lookup* matches, or None when no row matches, so a bulk reader skips the row read.
563 A loaded row still costs the one query of NetBox's object permission check.
564 """
565 return _assess_permission_scoped_save(
566 user,
567 model,
568 lookup,
569 values,
570 on_existing=on_existing,
571 current=current,
572 prospective_relations=prospective_relations,
573 unknown_fields=frozenset(unknown_fields),
574 )
577def enforce_saved_object_permission(obj, user, action):
578 """Reject a saved object whose final state is outside the user's scope.
580 ``has_perm`` rather than ``restrict()``: this decides one already-saved instance, and it holds
581 for any model rather than only those whose manager is a ``RestrictedQuerySet``.
582 """
583 if user is None:
584 return
585 permission = get_permission_for_model(type(obj), action)
586 if not user.has_perm(permission, obj):
587 raise ObjectPermissionDenied(permission)
590def reject_overlong_fields(instance, model):
591 """Reject a string the column cannot hold, before the database raises DataError.
593 Only length is checked. `full_clean` would also refuse a field that is legitimately blank.
594 """
595 for field in model._meta.concrete_fields:
596 max_length = getattr(field, "max_length", None)
597 value = getattr(instance, field.attname, None)
598 if max_length and isinstance(value, str) and len(value) > max_length:
599 raise ValidationError(f"{model._meta.verbose_name} {field.name} cannot exceed {max_length} characters.")
602def save_or_refetch(instance, model, lookup):
603 """Save *instance*, or refetch a row that won the same concurrent insert."""
604 try:
605 with transaction.atomic():
606 instance.save()
607 except IntegrityError:
608 existing = model.objects.filter(**lookup).first()
609 if existing is None:
610 raise
611 return existing, False
612 return instance, True
615def _policy_profile_id(model, lookup: dict) -> int | None:
616 """Return the profile a policy write belongs to, or None when the model carries no policy."""
617 from .models import PolicySectionModel
619 if not (isinstance(model, type) and issubclass(model, PolicySectionModel)):
620 return None
621 profile = lookup.get("profile")
622 profile_id = lookup.get("profile_id") if profile is None else profile.pk
623 if profile_id is None:
624 # A caller that names no profile cannot be serialized against an import, so it is a bug.
625 raise ValueError(f"A {model._meta.verbose_name} write must name the profile it belongs to.")
626 return profile_id
629def save_permission_scoped_object(
630 user,
631 model,
632 lookup: dict,
633 values: dict,
634 *,
635 on_existing: Literal["update", "keep", "reject"] = "update",
636) -> PermissionScopedSaveResult:
637 """Create or update one object within the user's NetBox permission scope.
639 ``on_existing`` decides what an already-present row means: ``update`` writes *values* under the
640 change permission, ``keep`` returns it untouched under the view permission, and ``reject``
641 refuses it. Raises ``ObjectPermissionDenied`` when any of those checks fails.
643 A policy write also holds its import profile, so it cannot commit inside an execution that has
644 already replanned against the old policy.
645 """
646 from .models import locked_profile_policy
648 profile_id = _policy_profile_id(model, lookup)
649 if profile_id is None:
650 return _scoped_write(user, model, lookup, values, on_existing=on_existing)
651 with locked_profile_policy(profile_id):
652 # atomic-exit-safe: policy-write-committed
653 return _scoped_write(user, model, lookup, values, on_existing=on_existing)
656def _scoped_write(
657 user,
658 model,
659 lookup: dict,
660 values: dict,
661 *,
662 on_existing: Literal["update", "keep", "reject"],
663) -> PermissionScopedSaveResult:
664 """Write one object and prove its saved state stays inside the user's object scope."""
665 with transaction.atomic():
666 instance = model.objects.select_for_update().filter(**lookup).first()
667 if instance is None:
668 permission = get_permission_for_model(model, "add")
669 if user is not None and not user.has_perm(permission):
670 raise ObjectPermissionDenied(permission)
671 instance = model(**lookup, **values)
672 reject_overlong_fields(instance, model)
673 instance, created = save_or_refetch(instance, model, lookup)
674 if not created:
675 instance = model.objects.select_for_update().get(pk=instance.pk)
676 else:
677 created = False
679 if not created:
680 if on_existing == "keep":
681 # Reusing someone else's row still exposes it, so it needs the view permission.
682 enforce_saved_object_permission(instance, user, "view")
683 # atomic-exit-safe: existing-row-kept-unwritten
684 return PermissionScopedSaveResult(instance=instance, created=False)
685 if on_existing == "reject":
686 raise ObjectPermissionDenied(get_permission_for_model(model, "add"))
687 # Before, so a row outside the user's scope cannot be taken over.
688 enforce_saved_object_permission(instance, user, "change")
689 if hasattr(instance, "snapshot"):
690 instance.snapshot()
691 for field_name, value in values.items():
692 setattr(instance, field_name, value)
693 reject_overlong_fields(instance, model)
694 instance.save(update_fields=list(values))
695 # After, so the new state cannot be moved outside the user's scope.
696 enforce_saved_object_permission(instance, user, "add" if created else "change")
697 # atomic-exit-safe: scoped-write-committed
698 return PermissionScopedSaveResult(instance=instance, created=created)
701def delete_permission_scoped_objects(user, queryset) -> int:
702 """Delete every row of *queryset* the user may delete, or none of them.
704 The rows are locked and checked one by one first, so a refusal leaves the whole set intact.
705 """
706 with transaction.atomic():
707 rows = list(queryset.select_for_update())
708 for row in rows:
709 enforce_saved_object_permission(row, user, "delete")
710 for row in rows:
711 row.delete()
712 # atomic-exit-safe: scoped-delete-committed
713 return len(rows)