Coverage for documents/permissions.py: 76%

192 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-10-10 09:07 +0000

1from typing import Any 

2from typing import TypeVar 

3 

4from django.contrib.auth.models import Group 

5from django.contrib.auth.models import Permission 

6from django.contrib.auth.models import User 

7from django.contrib.contenttypes.models import ContentType 

8from django.db.models import Case 

9from django.db.models import Count 

10from django.db.models import IntegerField 

11from django.db.models import Model 

12from django.db.models import Q 

13from django.db.models import QuerySet 

14from django.db.models import Value 

15from django.db.models import When 

16from django.db.models.functions import Cast 

17from guardian.core import ObjectPermissionChecker 

18from guardian.models import GroupObjectPermission 

19from guardian.models import UserObjectPermission 

20from guardian.shortcuts import assign_perm 

21from guardian.shortcuts import get_objects_for_user 

22from guardian.shortcuts import get_users_with_perms 

23from guardian.shortcuts import remove_perm 

24from rest_framework.permissions import BasePermission 

25from rest_framework.permissions import DjangoObjectPermissions 

26 

27from documents.models import Document 

28 

29 

30class PaperlessObjectPermissions(DjangoObjectPermissions): 

31 """ 

32 A permissions backend that checks for object-level permissions 

33 or for ownership. 

34 """ 

35 

36 perms_map = { 

37 "GET": ["%(app_label)s.view_%(model_name)s"], 

38 "OPTIONS": ["%(app_label)s.view_%(model_name)s"], 

39 "HEAD": ["%(app_label)s.view_%(model_name)s"], 

40 "POST": ["%(app_label)s.add_%(model_name)s"], 

41 "PUT": ["%(app_label)s.change_%(model_name)s"], 

42 "PATCH": ["%(app_label)s.change_%(model_name)s"], 

43 "DELETE": ["%(app_label)s.delete_%(model_name)s"], 

44 } 

45 

46 def has_object_permission(self, request, view, obj): 

47 if hasattr(obj, "owner") and obj.owner is not None: 

48 if request.user == obj.owner: 

49 return True 

50 else: 

51 return super().has_object_permission(request, view, obj) 

52 else: 

53 return True # no owner 

54 

55 

56class PaperlessAdminPermissions(BasePermission): 

57 def has_permission(self, request, view): 

58 return request.user.is_active and request.user.is_staff 

59 

60 

61def has_global_statistics_permission(user: User | None) -> bool: 

62 if ( 62 ↛ 67line 62 didn't jump to line 67 because the condition on line 62 was never true

63 user is None 

64 or not getattr(user, "is_active", False) 

65 or not getattr(user, "is_authenticated", False) 

66 ): 

67 return False 

68 

69 return getattr(user, "is_superuser", False) or user.has_perm( 

70 "paperless.view_global_statistics", 

71 ) 

72 

73 

74def has_system_status_permission(user: User | None) -> bool: 

75 if ( 

76 user is None 

77 or not getattr(user, "is_active", False) 

78 or not getattr(user, "is_authenticated", False) 

79 ): 

80 return False 

81 

82 return ( 

83 getattr(user, "is_superuser", False) 

84 or getattr(user, "is_staff", False) 

85 or user.has_perm("paperless.view_system_monitoring") 

86 ) 

87 

88 

89def get_groups_with_only_permission(obj, codename): 

90 ctype = ContentType.objects.get_for_model(obj) 

91 permission = Permission.objects.get(content_type=ctype, codename=codename) 

92 group_object_perm_group_ids = ( 

93 GroupObjectPermission.objects.filter( 

94 object_pk=obj.pk, 

95 content_type=ctype, 

96 ) 

97 .filter(permission=permission) 

98 .values_list("group_id") 

99 ) 

100 return Group.objects.filter(id__in=group_object_perm_group_ids).distinct() 

101 

102 

103def set_permissions_for_object( 

104 permissions: dict, 

105 object, 

106 *, 

107 merge: bool = False, 

108) -> None: 

109 """ 

110 Set permissions for an object. The permissions are given as a mapping of actions 

111 to a dict of user / group id lists, e.g. 

112 {"view": {"users": [1], "groups": [2]}, "change": {"users": [], "groups": []}}. 

113 

114 If merge is True, the permissions are merged with the existing permissions and 

115 no users or groups are removed. If False, the permissions are set to exactly 

116 the given list of users and groups. 

117 """ 

118 

119 for action, entry in permissions.items(): 

120 permission = f"{action}_{object.__class__.__name__.lower()}" 

121 if "users" in entry: 

122 # users 

123 users_to_add = User.objects.filter(id__in=entry["users"]) 

124 users_to_remove = ( 

125 get_users_with_perms( 

126 object, 

127 only_with_perms_in=[permission], 

128 with_group_users=False, 

129 ) 

130 if not merge 

131 else User.objects.none() 

132 ) 

133 if users_to_add.exists() and users_to_remove.exists(): 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true

134 users_to_remove = users_to_remove.exclude(id__in=users_to_add) 

135 if users_to_remove.exists(): 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true

136 for user in users_to_remove: 

137 remove_perm(permission, user, object) 

138 if users_to_add.exists(): 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true

139 for user in users_to_add: 

140 assign_perm(permission, user, object) 

141 if action == "change": 

142 # change gives view too 

143 assign_perm( 

144 f"view_{object.__class__.__name__.lower()}", 

145 user, 

146 object, 

147 ) 

148 if "groups" in entry: 

149 # groups 

150 groups_to_add = Group.objects.filter(id__in=entry["groups"]) 

151 groups_to_remove = ( 

152 get_groups_with_only_permission( 

153 object, 

154 permission, 

155 ) 

156 if not merge 

157 else Group.objects.none() 

158 ) 

159 if groups_to_add.exists() and groups_to_remove.exists(): 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true

160 groups_to_remove = groups_to_remove.exclude(id__in=groups_to_add) 

161 if groups_to_remove.exists(): 161 ↛ 162line 161 didn't jump to line 162 because the condition on line 161 was never true

162 for group in groups_to_remove: 

163 remove_perm(permission, group, object) 

164 if groups_to_add.exists(): 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true

165 for group in groups_to_add: 

166 assign_perm(permission, group, object) 

167 if action == "change": 

168 # change gives view too 

169 assign_perm( 

170 f"view_{object.__class__.__name__.lower()}", 

171 group, 

172 object, 

173 ) 

174 

175 

176def _resolve_permissions(codenames: set[str], ctype: ContentType) -> list[Permission]: 

177 """ 

178 Resolves `codenames` to Permission rows, raising like the single-object 

179 assign_perm() this bulk path replaces does (via a `.get()` internally) 

180 if any codename doesn't exist. SetPermissionsSerializer rejects unknown 

181 action names at the API, but a caller passing one directly would 

182 otherwise get a plain `.filter()` that silently builds zero rows and 

183 no-ops instead of reporting the bad input. 

184 """ 

185 permission_objs = list( 

186 Permission.objects.filter(content_type=ctype, codename__in=codenames), 

187 ) 

188 missing = codenames - {p.codename for p in permission_objs} 

189 if missing: 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true

190 raise Permission.DoesNotExist( 

191 f"Permission matching query does not exist for codename(s): " 

192 f"{', '.join(sorted(missing))}", 

193 ) 

194 return permission_objs 

195 

196 

197def _apply_bulk_permission_entry( 

198 *, 

199 perm_model: type[UserObjectPermission] | type[GroupObjectPermission], 

200 identity_model: type[User] | type[Group], 

201 identity_field: str, 

202 ids: list[int], 

203 codename: str, 

204 permission_objs: list[Permission], 

205 ctype: ContentType, 

206 object_pks: list[str], 

207 merge: bool, 

208) -> None: 

209 # Only the ids are needed to build permission rows (via `<field>_id=`), 

210 # so avoid fetching full User/Group rows for identities that may not 

211 # even end up being granted anything new. 

212 add_ids = set( 

213 identity_model.objects.filter(id__in=ids).values_list("id", flat=True), 

214 ) 

215 

216 if not merge: 216 ↛ 235line 216 didn't jump to line 235 because the condition on line 216 was always true

217 existing_ids = set( 

218 perm_model.objects.filter( 

219 content_type=ctype, 

220 object_pk__in=object_pks, 

221 permission__codename=codename, 

222 ) 

223 .values_list(f"{identity_field}_id", flat=True) 

224 .distinct(), 

225 ) 

226 remove_ids = existing_ids - add_ids 

227 if remove_ids: 227 ↛ 228line 227 didn't jump to line 228 because the condition on line 227 was never true

228 perm_model.objects.filter( 

229 content_type=ctype, 

230 object_pk__in=object_pks, 

231 permission__codename=codename, 

232 **{f"{identity_field}_id__in": remove_ids}, 

233 ).delete() 

234 

235 if not add_ids: 235 ↛ 238line 235 didn't jump to line 238 because the condition on line 235 was always true

236 return 

237 

238 rows = [ 

239 perm_model( 

240 content_type=ctype, 

241 object_pk=pk, 

242 permission=permission_obj, 

243 **{f"{identity_field}_id": identity_id}, 

244 ) 

245 for permission_obj in permission_objs 

246 for pk in object_pks 

247 for identity_id in add_ids 

248 ] 

249 # ignore_conflicts skips only rows that already exist as an exact 

250 # (identity, permission, object) match -- the same de-dup the 

251 # underlying (user|group, permission, object_pk) unique constraint 

252 # already enforces for the single-object assign_perm() this replaces, 

253 # so it doesn't change what counts as "already granted". batch_size 

254 # caps how many rows go into a single INSERT statement. 

255 perm_model.objects.bulk_create(rows, ignore_conflicts=True, batch_size=1000) 

256 

257 

258def set_permissions_for_objects( 

259 permissions: dict, 

260 model: type[Model], 

261 pks: QuerySet | list, 

262 *, 

263 merge: bool = False, 

264) -> None: 

265 """ 

266 Bulk equivalent of set_permissions_for_object: applies the same 

267 permission changes to every object identified by `pks` at once. 

268 

269 Takes a model + pks (rather than model instances) deliberately -- the 

270 permission rows built below only ever need `pk`, `content_type`, and 

271 identity ids, so callers shouldn't have to fetch full rows (with every 

272 other field) just to hand them to this function. 

273 

274 Deliberately does not use guardian's queryset/list-aware assign_perm: 

275 passing a list as the object routes to bulk_assign_perm, which skips 

276 creating a direct permission row for anyone who already has the 

277 permission via ANY group membership (it checks 

278 ObjectPermissionChecker.has_perm, which is group-inheritance-aware) -- 

279 unlike the single-object assign_perm this replaces, which always 

280 ensures a direct row via get_or_create regardless of group-derived 

281 access. Losing that guarantee would mean a later revocation of the 

282 group's grant silently strips access an admin explicitly asked to be 

283 direct. Bulk-creating rows straight against the permission models 

284 instead (see _apply_bulk_permission_entry) preserves the original 

285 always-create-a-direct-row semantics while still batching every object 

286 and every identity into one query per action, rather than one query per 

287 (object, user) pair. 

288 """ 

289 object_pks = [str(pk) for pk in pks] 

290 if not object_pks: # pragma: no cover 290 ↛ 291line 290 didn't jump to line 291 because the condition on line 290 was never true

291 return 

292 

293 model_name = model.__name__.lower() 

294 ctype = ContentType.objects.get_for_model(model) 

295 

296 # Every action is resolved up front, before anything is written, so an 

297 # unrecognized action name (see _resolve_permissions) aborts the whole 

298 # call instead of leaving the actions ahead of it already applied. 

299 # SetPermissionsSerializer rejects unknown actions at the API, so this 

300 # guards any other caller. 

301 permissions_by_action: dict[str, list[Permission]] = {} 

302 for action, entry in permissions.items(): 

303 if "users" not in entry and "groups" not in entry: 

304 continue 

305 implied_codenames = {f"{action}_{model_name}"} 

306 if action == "change": 

307 # change gives view too 

308 implied_codenames.add(f"view_{model_name}") 

309 permissions_by_action[action] = _resolve_permissions( 

310 implied_codenames, 

311 ctype, 

312 ) 

313 

314 for action, entry in permissions.items(): 

315 codename = f"{action}_{model_name}" 

316 permission_objs = permissions_by_action.get(action, []) 

317 

318 if "users" in entry: 

319 _apply_bulk_permission_entry( 

320 perm_model=UserObjectPermission, 

321 identity_model=User, 

322 identity_field="user", 

323 ids=entry["users"], 

324 codename=codename, 

325 permission_objs=permission_objs, 

326 ctype=ctype, 

327 object_pks=object_pks, 

328 merge=merge, 

329 ) 

330 

331 if "groups" in entry: 

332 _apply_bulk_permission_entry( 

333 perm_model=GroupObjectPermission, 

334 identity_model=Group, 

335 identity_field="group", 

336 ids=entry["groups"], 

337 codename=codename, 

338 permission_objs=permission_objs, 

339 ctype=ctype, 

340 object_pks=object_pks, 

341 merge=merge, 

342 ) 

343 

344 

345def permitted_object_ids( 

346 user: User | None, 

347 model: type[Model], 

348 perm: str, 

349 *, 

350 include_deleted: bool = False, 

351) -> QuerySet[int]: 

352 """ 

353 Generic version of ``permitted_document_ids`` for any model with an 

354 ``owner`` field and guardian object-level permissions. ``include_deleted`` 

355 only has an effect for models exposing a ``global_objects``/``deleted_at`` 

356 soft-delete pattern (currently only ``Document``); for every other model 

357 it is accepted but has no effect, since those models have no soft-delete 

358 concept. 

359 """ 

360 has_soft_delete = hasattr(model, "global_objects") 

361 manager = ( 

362 model.global_objects if include_deleted and has_soft_delete else model.objects 

363 ) 

364 base_qs = manager.all().only("id", "owner") 

365 

366 if user is None or not getattr(user, "is_authenticated", False): 

367 return base_qs.filter(owner__isnull=True).values_list("id", flat=True) 

368 

369 # Deactivated users get nothing, deactivated superusers included, so this 

370 # has to come before the superuser shortcut. guardian's 

371 # ObjectPermissionChecker denies inactive users, but get_objects_for_user 

372 # (the pattern this replaces) does not, so it would not be inherited. 

373 if not getattr(user, "is_active", False): 373 ↛ 374line 373 didn't jump to line 374 because the condition on line 373 was never true

374 return base_qs.none().values_list("id", flat=True) 

375 

376 if getattr(user, "is_superuser", False): 376 ↛ 383line 376 didn't jump to line 383 because the condition on line 376 was always true

377 return base_qs.values_list("id", flat=True) 

378 

379 # Guardian's UserObjectPermission/GroupObjectPermission always store a bare 

380 # codename, but has_perm()-style callers commonly pass the qualified 

381 # "app_label.codename" form. content_type already disambiguates the 

382 # codename, so just drop any prefix rather than silently under-permitting. 

383 perm = perm.rsplit(".", 1)[-1] 

384 

385 content_type = ContentType.objects.get_for_model(model) 

386 perm_filter = { 

387 "permission__codename": perm, 

388 "permission__content_type": content_type, 

389 } 

390 

391 user_perm_ids = ( 

392 UserObjectPermission.objects.filter(user=user, **perm_filter) 

393 .annotate(object_pk_int=Cast("object_pk", IntegerField())) 

394 .values_list("object_pk_int", flat=True) 

395 ) 

396 group_perm_ids = ( 

397 GroupObjectPermission.objects.filter(group__user=user, **perm_filter) 

398 .annotate(object_pk_int=Cast("object_pk", IntegerField())) 

399 .values_list("object_pk_int", flat=True) 

400 ) 

401 permitted_ids = user_perm_ids.union(group_perm_ids) 

402 

403 return base_qs.filter( 

404 Q(owner=user) | Q(owner__isnull=True) | Q(id__in=permitted_ids), 

405 ).values_list("id", flat=True) 

406 

407 

408ModelT = TypeVar("ModelT", bound=Model) 

409 

410 

411def user_is_unrestricted(user: User | None) -> bool: 

412 """ 

413 True when ``user`` means "no restriction at all" (an absent user, or an 

414 *active* superuser) without needing a database check to know it. 

415 

416 ``permitted_object_ids(None, ...)`` itself means the much narrower "only 

417 unowned rows", which is NOT the same thing as "no user filtering 

418 requested", so callers must special-case this before ever calling it. 

419 A deactivated superuser is deliberately NOT unrestricted here, matching 

420 permitted_object_ids's own is_active-before-is_superuser ordering. 

421 

422 Callers that can avoid a database round trip entirely when this is true 

423 (e.g. checking a single already-loaded object's visibility rather than 

424 filtering a queryset) should do so via this function directly, rather 

425 than through restrict_queryset_to_visible() below. 

426 """ 

427 if user is None: 427 ↛ 428line 427 didn't jump to line 428 because the condition on line 427 was never true

428 return True 

429 return ( 

430 getattr(user, "is_authenticated", False) 

431 and getattr(user, "is_active", False) 

432 and getattr(user, "is_superuser", False) 

433 ) 

434 

435 

436def restrict_queryset_to_visible( 

437 queryset: QuerySet[ModelT], 

438 user: User | None, 

439 perm: str, 

440) -> QuerySet[ModelT]: 

441 """ 

442 Restrict ``queryset`` to the rows ``user`` may see with ``perm``. 

443 

444 Delegates the visibility check to the database as a 

445 ``WHERE id IN (subquery)`` rather than materializing the full 

446 permitted-id set into a Python collection first: a caller that only 

447 needs to check a small handful of rows (a resolved-id list, a few 

448 RAG-neighbour candidate ids) never pays for scanning or holding the 

449 installation's entire taxonomy in memory to do it. 

450 

451 Returns ``queryset`` unchanged for user_is_unrestricted(user); every 

452 other case is delegated to ``permitted_object_ids`` rather than 

453 re-deciding the ordering here. 

454 """ 

455 if user_is_unrestricted(user): 455 ↛ 457line 455 didn't jump to line 457 because the condition on line 455 was always true

456 return queryset 

457 return queryset.filter(pk__in=permitted_object_ids(user, queryset.model, perm)) 

458 

459 

460def permitted_document_ids( 

461 user: User | None, 

462 *, 

463 perm: str = "view_document", 

464 include_deleted: bool = False, 

465) -> QuerySet[int]: 

466 """ 

467 Document-specific convenience wrapper around ``permitted_object_ids``. 

468 Return a queryset of document IDs the user has ``perm`` on (default 

469 ``"view_document"``). By default limited to non-deleted documents; pass 

470 ``include_deleted=True`` for callers that need to check permission on 

471 soft-deleted documents (e.g. trash restore). This intentionally avoids 

472 ``get_objects_for_user`` to keep the subquery small and index-friendly. 

473 """ 

474 return permitted_object_ids(user, Document, perm, include_deleted=include_deleted) 

475 

476 

477def get_document_count_filter_for_user(user, related_name: str = "documents"): 

478 """ 

479 Return the Q object used to filter document counts for the given user. 

480 

481 The filter is expressed as an ``id__in`` against a small subquery of permitted 

482 document IDs to keep the generated SQL simple and avoid large OR clauses. 

483 

484 ``related_name`` is the ORM path from the annotated model to Document (e.g. 

485 ``"documents"`` for Tag's direct M2M, or ``"fields__document"`` for CustomField, 

486 which only reaches Document via the CustomFieldInstance through-model). 

487 """ 

488 

489 if getattr(user, "is_superuser", False): 

490 # Superuser: no permission filtering needed 

491 return Q(**{f"{related_name}__deleted_at__isnull": True}) 

492 

493 permitted_ids = permitted_document_ids(user) 

494 return Q(**{f"{related_name}__id__in": permitted_ids}) 

495 

496 

497def annotate_document_count_by_ids( 

498 queryset: QuerySet[Any], 

499 through_model: Any, 

500 related_object_field: str, 

501 document_ids: Any, 

502 target_field: str = "document_id", 

503) -> QuerySet[Any]: 

504 """ 

505 Annotate a queryset with a document count for a relation to Document that 

506 goes through an M2M/through-model table (e.g. Tag via 

507 ``Document.tags.through``, or CustomField via ``CustomFieldInstance``), 

508 for an explicit, already-resolved set of document ids. 

509 

510 Counts are computed via a single, independent GROUP BY over the relation 

511 table -- with the id filter expressed as a plain ``WHERE`` rather than an 

512 aggregate ``FILTER`` -- then injected via ``Case``/``When``. This 

513 deliberately avoids two slower alternatives found while building this: 

514 

515 - A per-outer-row correlated subquery (one execution per row of the 

516 annotated queryset): fine at a handful of rows, catastrophic once the 

517 queryset has hundreds/thousands of rows. 

518 - ``Count(..., filter=Q(id__in=document_ids), distinct=True)`` applied 

519 directly to the M2M relation: Postgres can fail to plan the ``id__in`` 

520 check as a semi-join and instead re-checks subquery membership once per 

521 row of the (much larger) M2M join -- worse than the correlated subquery. 

522 

523 Aggregation is restricted to rows whose ``related_object_field`` is one of 

524 ``queryset``'s pks, so passing a subset (e.g. a handful of tag descendants) 

525 doesn't pay the cost of counting for every row matching ``document_ids``. 

526 

527 Args: 

528 queryset: base queryset to annotate (must contain pk) 

529 through_model: model representing the relation (e.g., Document.tags.through 

530 or CustomFieldInstance) 

531 related_object_field: field on the relation pointing back to queryset pk 

532 document_ids: the document ids to count against -- a concrete list/set, 

533 or a simple (already resolved) queryset of ids. Callers 

534 that need this filtered by a complex condition (e.g. a 

535 permission check) should resolve it to a concrete list 

536 first if the same ids will be reused across multiple 

537 calls, rather than passing the complex queryset itself 

538 into each -- see ``_get_selection_data_for_queryset``. 

539 target_field: field on the relation pointing to Document id 

540 """ 

541 

542 counts = ( 

543 through_model.objects.filter( 

544 **{ 

545 f"{related_object_field}__in": queryset.values("pk"), 

546 f"{target_field}__in": document_ids, 

547 }, 

548 ) 

549 .values(related_object_field) 

550 .annotate(c=Count(target_field, distinct=True)) 

551 ) 

552 counts_by_pk = {row[related_object_field]: row["c"] for row in counts} 

553 

554 if not counts_by_pk: 554 ↛ 559line 554 didn't jump to line 559 because the condition on line 554 was always true

555 return queryset.annotate( 

556 document_count=Value(0, output_field=IntegerField()), 

557 ) 

558 

559 return queryset.annotate( 

560 document_count=Case( 

561 *(When(pk=pk, then=Value(count)) for pk, count in counts_by_pk.items()), 

562 default=Value(0), 

563 output_field=IntegerField(), 

564 ), 

565 ) 

566 

567 

568def annotate_document_count_for_related_queryset( 

569 queryset: QuerySet[Any], 

570 through_model: Any, 

571 related_object_field: str, 

572 target_field: str = "document_id", 

573 user: User | None = None, 

574) -> QuerySet[Any]: 

575 """ 

576 Same as ``annotate_document_count_by_ids``, but resolves the document ids 

577 from the given user's view permissions rather than taking them directly. 

578 """ 

579 

580 return annotate_document_count_by_ids( 

581 queryset, 

582 through_model=through_model, 

583 related_object_field=related_object_field, 

584 document_ids=permitted_document_ids(user), 

585 target_field=target_field, 

586 ) 

587 

588 

589def get_objects_for_user_owner_aware( 

590 user: User | None, 

591 perms: str | list[str], 

592 Model: Any, 

593 *, 

594 include_deleted: bool = False, 

595) -> QuerySet[Any]: 

596 """ 

597 Returns objects the user owns, are unowned, or has explicit perms. 

598 When include_deleted is True, soft-deleted items are also included. 

599 

600 Legacy slow path (guardian-backed, O(n) style permission resolution). 

601 Most queryset-filtering call sites have migrated onto 

602 ``PermittedObjectsFilter``/``permitted_object_ids()``, but this function 

603 is kept because production callers still remain. Several callers remain 

604 across ``documents/``, ``paperless_mail/``, and ``paperless_ai/`` -- 

605 grep for this function name before removing it. 

606 """ 

607 manager = ( 

608 Model.global_objects 

609 if include_deleted and hasattr(Model, "global_objects") 

610 else Model.objects 

611 ) 

612 

613 objects_owned = manager.filter(owner=user) 

614 objects_unowned = manager.filter(owner__isnull=True) 

615 objects_with_perms = get_objects_for_user( 

616 user=user, 

617 perms=perms, 

618 klass=manager.all(), 

619 accept_global_perms=False, 

620 ) 

621 return objects_owned | objects_unowned | objects_with_perms 

622 

623 

624def has_perms_owner_aware(user, perms, obj): 

625 """ 

626 Legacy slow path (guardian-backed) single-object permission check. 

627 

628 The queryset-filtering side of this migrated onto 

629 ``PermittedObjectsFilter``/``permitted_object_ids()``, but this 

630 single-object check still has many production callers. Several callers 

631 remain across ``documents/``, ``paperless_mail/``, and ``paperless_ai/`` 

632 -- grep for this function name before removing it. 

633 """ 

634 checker = ObjectPermissionChecker(user) 

635 return obj.owner is None or obj.owner == user or checker.has_perm(perms, obj) 

636 

637 

638class ViewDocumentsPermissions(BasePermission): 

639 """ 

640 Permissions class that checks for model permissions for only viewing Documents. 

641 """ 

642 

643 perms_map = { 

644 "OPTIONS": ["documents.view_document"], 

645 "GET": ["documents.view_document"], 

646 "POST": ["documents.view_document"], 

647 } 

648 

649 def has_permission(self, request, view): 

650 if not request.user or (not request.user.is_authenticated): # pragma: no cover 650 ↛ 651line 650 didn't jump to line 651 because the condition on line 650 was never true

651 return False 

652 

653 return request.user.has_perms(self.perms_map.get(request.method, [])) 

654 

655 

656class TrashPermissions(BasePermission): 

657 """Check the global document permission for each trash operation.""" 

658 

659 perms_map = { 

660 "OPTIONS": ["documents.view_document"], 

661 "HEAD": ["documents.view_document"], 

662 "GET": ["documents.view_document"], 

663 "POST": ["documents.delete_document"], 

664 } 

665 

666 def has_permission(self, request, view): 

667 if not request.user or not request.user.is_authenticated: # pragma: no cover 667 ↛ 668line 667 didn't jump to line 668 because the condition on line 667 was never true

668 return False 

669 

670 return request.user.has_perms(self.perms_map.get(request.method, [])) 

671 

672 

673class PaperlessNotePermissions(BasePermission): 

674 """ 

675 Permissions class that checks for model permissions for Notes. 

676 """ 

677 

678 perms_map = { 

679 "OPTIONS": ["documents.view_note", "documents.view_document"], 

680 "GET": ["documents.view_note", "documents.view_document"], 

681 "POST": [ 

682 "documents.add_note", 

683 "documents.view_document", 

684 "documents.change_document", 

685 ], 

686 "DELETE": [ 

687 "documents.delete_note", 

688 "documents.view_document", 

689 "documents.change_document", 

690 ], 

691 } 

692 

693 def has_permission(self, request, view): 

694 if not request.user or (not request.user.is_authenticated): # pragma: no cover 

695 return False 

696 

697 perms = self.perms_map[request.method] 

698 

699 return request.user.has_perms(perms) 

700 

701 

702class AcknowledgeTasksPermissions(BasePermission): 

703 """ 

704 Permissions class that checks for model permissions for acknowledging tasks. 

705 """ 

706 

707 perms_map = { 

708 "POST": ["documents.change_paperlesstask"], 

709 } 

710 

711 def has_permission(self, request: Any, view: Any) -> bool: 

712 if not request.user or not request.user.is_authenticated: # pragma: no cover 712 ↛ 713line 712 didn't jump to line 713 because the condition on line 712 was never true

713 return False 

714 

715 perms = self.perms_map.get(request.method, []) 

716 

717 return request.user.has_perms(perms)