Coverage for open_webui/models/access_grants.py: 38%
278 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 05:07 +0000
1import logging
2import time
3import uuid
4from typing import Optional
6from open_webui.internal.db import Base, get_async_db_context
7from pydantic import BaseModel, ConfigDict
8from sqlalchemy import BigInteger, Column, Text, UniqueConstraint, and_, delete, or_, select
9from sqlalchemy.dialects.postgresql import JSONB
10from sqlalchemy.ext.asyncio import AsyncSession
12log = logging.getLogger(__name__)
14PRINCIPAL_TYPE_ANYONE = 'anyone'
15PRINCIPAL_TYPE_GROUP = 'group'
16PRINCIPAL_TYPE_USER = 'user'
17WILDCARD_PRINCIPAL_ID = '*'
20####################
21# AccessGrant DB Schema
22####################
25class AccessGrant(Base):
26 __tablename__ = 'access_grant'
28 id = Column(Text, primary_key=True)
29 resource_type = Column(Text, nullable=False) # "knowledge", "model", "prompt", "tool", "note", "channel", "file"
30 resource_id = Column(Text, nullable=False)
31 principal_type = Column(Text, nullable=False) # "user", "group", or "anyone"
32 principal_id = Column(Text, nullable=False) # user_id, group_id, or "*" (wildcard for public)
33 permission = Column(Text, nullable=False) # "read" or "write"
34 created_at = Column(BigInteger, nullable=False)
36 __table_args__ = (
37 UniqueConstraint(
38 'resource_type',
39 'resource_id',
40 'principal_type',
41 'principal_id',
42 'permission',
43 name='uq_access_grant_grant',
44 ),
45 )
48class AccessGrantModel(BaseModel):
49 model_config = ConfigDict(from_attributes=True)
51 id: str
52 resource_type: str
53 resource_id: str
54 principal_type: str
55 principal_id: str
56 permission: str
57 created_at: int
60class AccessGrantResponse(BaseModel):
61 """Slim grant model for API responses — resource context is implicit from the parent."""
63 id: str
64 principal_type: str
65 principal_id: str
66 permission: str
68 @classmethod
69 def from_grant(cls, grant: 'AccessGrantModel') -> 'AccessGrantResponse':
70 return cls(
71 id=grant.id,
72 principal_type=grant.principal_type,
73 principal_id=grant.principal_id,
74 permission=grant.permission,
75 )
78####################
79# Conversion utilities
80####################
83def access_control_to_grants(
84 resource_type: str,
85 resource_id: str,
86 access_control: Optional[dict],
87) -> list[dict]:
88 """
89 Convert an old-style access_control JSON dict to a flat list of grant dicts.
91 Semantics:
92 - None → public read (user:* read) — except files which are private
93 - {} → private/owner-only (no grants)
94 - {read: {group_ids, user_ids}, write: {group_ids, user_ids}} → specific grants
96 Returns a list of dicts with keys: resource_type, resource_id, principal_type, principal_id, permission
97 """
98 grants = []
100 if access_control is None:
101 # NULL → public read (user:* for read)
102 # Exception: files with NULL are private (owner-only), no grants needed
103 if resource_type != 'file':
104 grants.append(
105 {
106 'resource_type': resource_type,
107 'resource_id': resource_id,
108 'principal_type': 'user',
109 'principal_id': '*',
110 'permission': 'read',
111 }
112 )
113 return grants
115 # {} → private/owner-only, no grants
116 if not access_control:
117 return grants
119 # Parse structured permissions
120 for permission in ['read', 'write']:
121 perm_data = access_control.get(permission, {})
122 if not perm_data:
123 continue
125 for group_id in perm_data.get('group_ids', []):
126 grants.append(
127 {
128 'resource_type': resource_type,
129 'resource_id': resource_id,
130 'principal_type': 'group',
131 'principal_id': group_id,
132 'permission': permission,
133 }
134 )
136 for user_id in perm_data.get('user_ids', []):
137 grants.append(
138 {
139 'resource_type': resource_type,
140 'resource_id': resource_id,
141 'principal_type': 'user',
142 'principal_id': user_id,
143 'permission': permission,
144 }
145 )
147 return grants
150def normalize_access_grants(access_grants: Optional[list]) -> list[dict]:
151 """
152 Normalize direct access_grants payloads from API forms.
154 Keeps only valid grants and removes duplicates by
155 (principal_type, principal_id, permission).
156 """
157 if not access_grants:
158 return []
160 deduped = {}
161 for grant in access_grants:
162 if isinstance(grant, BaseModel):
163 grant = grant.model_dump()
164 if not isinstance(grant, dict): 164 ↛ 165line 164 didn't jump to line 165 because the condition on line 164 was never true
165 continue
167 principal_type = grant.get('principal_type')
168 principal_id = grant.get('principal_id')
169 permission = grant.get('permission')
171 if principal_type not in (PRINCIPAL_TYPE_USER, PRINCIPAL_TYPE_GROUP, PRINCIPAL_TYPE_ANYONE): 171 ↛ 173line 171 didn't jump to line 173 because the condition on line 171 was always true
172 continue
173 if permission not in ('read', 'write'):
174 continue
175 if not isinstance(principal_id, str) or not principal_id:
176 continue
177 if principal_type == PRINCIPAL_TYPE_ANYONE and (principal_id != WILDCARD_PRINCIPAL_ID or permission != 'read'):
178 continue
180 key = (principal_type, principal_id, permission)
181 deduped[key] = {
182 'id': (grant.get('id') if isinstance(grant.get('id'), str) and grant.get('id') else str(uuid.uuid4())),
183 'principal_type': principal_type,
184 'principal_id': principal_id,
185 'permission': permission,
186 }
188 return list(deduped.values())
191def has_public_read_access_grant(access_grants: Optional[list]) -> bool:
192 """
193 Returns True when a direct grant list includes wildcard public-read.
194 """
195 for grant in normalize_access_grants(access_grants):
196 if (
197 grant['principal_type'] == PRINCIPAL_TYPE_USER
198 and grant['principal_id'] == WILDCARD_PRINCIPAL_ID
199 and grant['permission'] == 'read'
200 ):
201 return True
202 return False
205def has_public_write_access_grant(access_grants: Optional[list]) -> bool:
206 """
207 Returns True when a direct grant list includes wildcard public-write.
208 """
209 for grant in normalize_access_grants(access_grants):
210 if (
211 grant['principal_type'] == PRINCIPAL_TYPE_USER
212 and grant['principal_id'] == WILDCARD_PRINCIPAL_ID
213 and grant['permission'] == 'write'
214 ):
215 return True
216 return False
219def has_anyone_read_access_grant(access_grants: Optional[list]) -> bool:
220 """
221 Returns True when a direct grant list includes no-auth anyone-read.
222 """
223 for grant in normalize_access_grants(access_grants): 223 ↛ 224line 223 didn't jump to line 224 because the loop on line 223 never started
224 if (
225 grant['principal_type'] == PRINCIPAL_TYPE_ANYONE
226 and grant['principal_id'] == WILDCARD_PRINCIPAL_ID
227 and grant['permission'] == 'read'
228 ):
229 return True
230 return False
233def has_user_access_grant(access_grants: Optional[list]) -> bool:
234 """
235 Returns True when a direct grant list includes any non-wildcard user grant.
236 """
237 for grant in normalize_access_grants(access_grants):
238 if grant['principal_type'] == PRINCIPAL_TYPE_USER and grant['principal_id'] != WILDCARD_PRINCIPAL_ID:
239 return True
240 return False
243def strip_user_access_grants(access_grants: Optional[list]) -> list:
244 """
245 Remove all non-wildcard user grants from the list.
246 Keeps group grants and the public wildcard (user:*) intact.
247 """
248 if not access_grants:
249 return []
250 return [
251 grant
252 for grant in access_grants
253 if not (
254 (grant.get('principal_type') if isinstance(grant, dict) else getattr(grant, 'principal_type', None))
255 == PRINCIPAL_TYPE_USER
256 and (grant.get('principal_id') if isinstance(grant, dict) else getattr(grant, 'principal_id', None))
257 != WILDCARD_PRINCIPAL_ID
258 )
259 ]
262def strip_anyone_access_grants(access_grants: Optional[list]) -> list:
263 """
264 Remove no-auth anyone grants from the list.
265 """
266 if not access_grants:
267 return []
268 return [
269 grant
270 for grant in access_grants
271 if (grant.get('principal_type') if isinstance(grant, dict) else getattr(grant, 'principal_type', None))
272 != PRINCIPAL_TYPE_ANYONE
273 ]
276def grants_to_access_control(grants: list) -> Optional[dict]:
277 """
278 Convert a list of grant objects (AccessGrantModel or AccessGrantResponse)
279 back to the old-style access_control JSON dict for backward compatibility.
281 Semantics:
282 - [] (empty) → {} (private/owner-only)
283 - Contains user:*:read → None (public), but write grants are preserved
284 - Otherwise → {read: {group_ids, user_ids}, write: {group_ids, user_ids}}
286 Note: "public" (user:*:read) still allows additional write permissions
287 to coexist. When the wildcard read is present the function returns None
288 for the legacy dict, so callers that need write info should inspect the
289 grants list directly.
290 """
291 if not grants:
292 return {} # No grants = private/owner-only
294 result = {
295 'read': {'group_ids': [], 'user_ids': []},
296 'write': {'group_ids': [], 'user_ids': []},
297 }
299 is_public = False
300 for grant in grants:
301 if grant.principal_type == 'user' and grant.principal_id == '*' and grant.permission == 'read':
302 is_public = True
303 continue # Don't add wildcard to user_ids list
305 if grant.permission not in ('read', 'write'):
306 continue
308 if grant.principal_type == 'group':
309 if grant.principal_id not in result[grant.permission]['group_ids']:
310 result[grant.permission]['group_ids'].append(grant.principal_id)
311 elif grant.principal_type == 'user':
312 if grant.principal_id not in result[grant.permission]['user_ids']:
313 result[grant.permission]['user_ids'].append(grant.principal_id)
315 if is_public:
316 return None # Public read access
318 return result
321####################
322# Table Operations
323####################
326class AccessGrantsTable:
327 async def grant_access(
328 self,
329 resource_type: str,
330 resource_id: str,
331 principal_type: str,
332 principal_id: str,
333 permission: str,
334 db: Optional[AsyncSession] = None,
335 ) -> Optional[AccessGrantModel]:
336 """Add a single access grant. Idempotent (ignores duplicates)."""
337 async with get_async_db_context(db) as db:
338 # Check for existing grant
339 result = await db.execute(
340 select(AccessGrant).filter_by(
341 resource_type=resource_type,
342 resource_id=resource_id,
343 principal_type=principal_type,
344 principal_id=principal_id,
345 permission=permission,
346 )
347 )
348 existing = result.scalars().first()
349 if existing:
350 return AccessGrantModel.model_validate(existing)
352 grant = AccessGrant(
353 id=str(uuid.uuid4()),
354 resource_type=resource_type,
355 resource_id=resource_id,
356 principal_type=principal_type,
357 principal_id=principal_id,
358 permission=permission,
359 created_at=int(time.time()),
360 )
361 db.add(grant)
362 await db.commit()
363 return AccessGrantModel.model_validate(grant)
365 async def revoke_access(
366 self,
367 resource_type: str,
368 resource_id: str,
369 principal_type: str,
370 principal_id: str,
371 permission: str,
372 db: Optional[AsyncSession] = None,
373 ) -> bool:
374 """Remove a single access grant."""
375 async with get_async_db_context(db) as db:
376 result = await db.execute(
377 delete(AccessGrant).filter_by(
378 resource_type=resource_type,
379 resource_id=resource_id,
380 principal_type=principal_type,
381 principal_id=principal_id,
382 permission=permission,
383 )
384 )
385 await db.commit()
386 return result.rowcount > 0
388 async def revoke_all_access(
389 self,
390 resource_type: str,
391 resource_id: str,
392 db: Optional[AsyncSession] = None,
393 ) -> int:
394 """Remove all access grants for a resource."""
395 async with get_async_db_context(db) as db:
396 result = await db.execute(
397 delete(AccessGrant).filter_by(
398 resource_type=resource_type,
399 resource_id=resource_id,
400 )
401 )
402 await db.commit()
403 return result.rowcount
405 async def set_access_control(
406 self,
407 resource_type: str,
408 resource_id: str,
409 access_control: Optional[dict],
410 db: Optional[AsyncSession] = None,
411 ) -> list[AccessGrantModel]:
412 """
413 Replace all grants for a resource from an access_control JSON dict.
414 This is the primary bridge for backward compat with the frontend.
415 """
416 async with get_async_db_context(db) as db:
417 # Delete all existing grants for this resource
418 await db.execute(
419 delete(AccessGrant).filter_by(
420 resource_type=resource_type,
421 resource_id=resource_id,
422 )
423 )
425 # Convert JSON to grant dicts
426 grant_dicts = access_control_to_grants(resource_type, resource_id, access_control)
428 # Insert new grants
429 results = []
430 for grant_dict in grant_dicts:
431 grant = AccessGrant(
432 id=str(uuid.uuid4()),
433 **grant_dict,
434 created_at=int(time.time()),
435 )
436 db.add(grant)
437 results.append(grant)
439 await db.commit()
441 return [AccessGrantModel.model_validate(g) for g in results]
443 async def set_access_grants(
444 self,
445 resource_type: str,
446 resource_id: str,
447 access_grants: Optional[list],
448 db: Optional[AsyncSession] = None,
449 ) -> list[AccessGrantModel]:
450 """
451 Replace all grants for a resource from a direct access_grants list.
452 """
453 async with get_async_db_context(db) as db:
454 await db.execute(
455 delete(AccessGrant).filter_by(
456 resource_type=resource_type,
457 resource_id=resource_id,
458 )
459 )
461 normalized_grants = normalize_access_grants(access_grants)
463 results = []
464 for grant_dict in normalized_grants:
465 grant = AccessGrant(
466 id=str(uuid.uuid4()),
467 resource_type=resource_type,
468 resource_id=resource_id,
469 principal_type=grant_dict['principal_type'],
470 principal_id=grant_dict['principal_id'],
471 permission=grant_dict['permission'],
472 created_at=int(time.time()),
473 )
474 db.add(grant)
475 results.append(grant)
477 await db.commit()
478 return [AccessGrantModel.model_validate(g) for g in results]
480 async def get_access_control(
481 self,
482 resource_type: str,
483 resource_id: str,
484 db: Optional[AsyncSession] = None,
485 ) -> Optional[dict]:
486 """
487 Reconstruct the old-style access_control JSON dict from grants.
488 For backward compat with the frontend.
489 """
490 async with get_async_db_context(db) as db:
491 result = await db.execute(
492 select(AccessGrant).filter_by(
493 resource_type=resource_type,
494 resource_id=resource_id,
495 )
496 )
497 grants = result.scalars().all()
498 grant_models = [AccessGrantModel.model_validate(g) for g in grants]
499 return grants_to_access_control(grant_models)
501 async def get_grants_by_resource(
502 self,
503 resource_type: str,
504 resource_id: str,
505 db: Optional[AsyncSession] = None,
506 ) -> list[AccessGrantModel]:
507 """Get all grants for a specific resource."""
508 async with get_async_db_context(db) as db:
509 result = await db.execute(
510 select(AccessGrant).filter_by(
511 resource_type=resource_type,
512 resource_id=resource_id,
513 )
514 )
515 grants = result.scalars().all()
516 return [AccessGrantModel.model_validate(g) for g in grants]
518 async def get_grants_by_resources(
519 self,
520 resource_type: str,
521 resource_ids: list[str],
522 db: Optional[AsyncSession] = None,
523 ) -> dict[str, list[AccessGrantModel]]:
524 """Batch-fetch grants for multiple resources. Returns {resource_id: [grants]}."""
525 if not resource_ids:
526 return {}
527 async with get_async_db_context(db) as db:
528 result = await db.execute(
529 select(AccessGrant).filter(
530 AccessGrant.resource_type == resource_type,
531 AccessGrant.resource_id.in_(resource_ids),
532 )
533 )
534 grants = result.scalars().all()
535 result_dict: dict[str, list[AccessGrantModel]] = {rid: [] for rid in resource_ids}
536 for g in grants:
537 result_dict[g.resource_id].append(AccessGrantModel.model_validate(g))
538 return result_dict
540 async def has_anyone_access(
541 self,
542 resource_type: str,
543 resource_id: str,
544 permission: str = 'read',
545 db: Optional[AsyncSession] = None,
546 ) -> bool:
547 """Check for a no-auth anyone:* grant. Callers must opt in explicitly."""
548 async with get_async_db_context(db) as db:
549 result = await db.execute(
550 select(AccessGrant)
551 .filter(
552 AccessGrant.resource_type == resource_type,
553 AccessGrant.resource_id == resource_id,
554 AccessGrant.principal_type == PRINCIPAL_TYPE_ANYONE,
555 AccessGrant.principal_id == WILDCARD_PRINCIPAL_ID,
556 AccessGrant.permission == permission,
557 )
558 .limit(1)
559 )
560 return result.scalars().first() is not None
562 async def has_access(
563 self,
564 user_id: str,
565 resource_type: str,
566 resource_id: str,
567 permission: str = 'read',
568 user_group_ids: Optional[set[str]] = None,
569 db: Optional[AsyncSession] = None,
570 ) -> bool:
571 """
572 Check if a user has the specified permission on a resource.
574 Access is granted if any of the following is true:
575 - There's a grant for user:* (public) with the requested permission
576 - There's a grant for the specific user with the requested permission
577 - There's a grant for any of the user's groups with the requested permission
578 """
579 async with get_async_db_context(db) as db:
580 # Build conditions for matching grants
581 conditions = [
582 # Public access
583 and_(
584 AccessGrant.principal_type == 'user',
585 AccessGrant.principal_id == '*',
586 ),
587 # Direct user access
588 and_(
589 AccessGrant.principal_type == 'user',
590 AccessGrant.principal_id == user_id,
591 ),
592 ]
594 # Group access
595 if user_group_ids is None:
596 from open_webui.models.groups import Groups
598 user_groups = await Groups.get_groups_by_member_id(user_id, db=db)
599 user_group_ids = {group.id for group in user_groups}
601 if user_group_ids:
602 conditions.append(
603 and_(
604 AccessGrant.principal_type == 'group',
605 AccessGrant.principal_id.in_(user_group_ids),
606 )
607 )
609 result = await db.execute(
610 select(AccessGrant)
611 .filter(
612 AccessGrant.resource_type == resource_type,
613 AccessGrant.resource_id == resource_id,
614 AccessGrant.permission == permission,
615 or_(*conditions),
616 )
617 .limit(1)
618 )
619 grant = result.scalars().first()
620 return grant is not None
622 async def get_accessible_resource_ids(
623 self,
624 user_id: str,
625 resource_type: str,
626 resource_ids: list[str],
627 permission: str = 'read',
628 user_group_ids: Optional[set[str]] = None,
629 db: Optional[AsyncSession] = None,
630 ) -> set[str]:
631 """
632 Batch check: return the subset of resource_ids that the user can access.
634 This replaces calling has_access() in a loop (N+1) with a single query.
635 """
636 if not resource_ids:
637 return set()
639 async with get_async_db_context(db) as db:
640 conditions = [
641 and_(
642 AccessGrant.principal_type == 'user',
643 AccessGrant.principal_id == '*',
644 ),
645 and_(
646 AccessGrant.principal_type == 'user',
647 AccessGrant.principal_id == user_id,
648 ),
649 ]
651 if user_group_ids is None: 651 ↛ 652line 651 didn't jump to line 652 because the condition on line 651 was never true
652 from open_webui.models.groups import Groups
654 user_groups = await Groups.get_groups_by_member_id(user_id, db=db)
655 user_group_ids = {group.id for group in user_groups}
657 if user_group_ids:
658 conditions.append(
659 and_(
660 AccessGrant.principal_type == 'group',
661 AccessGrant.principal_id.in_(user_group_ids),
662 )
663 )
665 result = await db.execute(
666 select(AccessGrant.resource_id)
667 .filter(
668 AccessGrant.resource_type == resource_type,
669 AccessGrant.resource_id.in_(resource_ids),
670 AccessGrant.permission == permission,
671 or_(*conditions),
672 )
673 .distinct()
674 )
675 rows = result.all()
676 return {row[0] for row in rows}
678 async def get_users_with_access(
679 self,
680 resource_type: str,
681 resource_id: str,
682 permission: str = 'read',
683 db: Optional[AsyncSession] = None,
684 ) -> list:
685 """
686 Get all users who have the specified permission on a resource.
687 Returns a list of UserModel instances.
688 """
689 from open_webui.models.groups import Groups
690 from open_webui.models.users import UserModel, Users
692 async with get_async_db_context(db) as db:
693 result = await db.execute(
694 select(AccessGrant).filter_by(
695 resource_type=resource_type,
696 resource_id=resource_id,
697 permission=permission,
698 )
699 )
700 grants = result.scalars().all()
702 # Check for public access
703 for grant in grants:
704 if grant.principal_type == 'user' and grant.principal_id == '*':
705 result = await Users.get_users(filter={'roles': ['!pending']}, db=db)
706 return result.get('users', [])
708 user_ids_with_access = set()
710 for grant in grants:
711 if grant.principal_type == 'user':
712 user_ids_with_access.add(grant.principal_id)
713 elif grant.principal_type == 'group':
714 group_user_ids = await Groups.get_group_user_ids_by_id(grant.principal_id, db=db)
715 if group_user_ids:
716 user_ids_with_access.update(group_user_ids)
718 if not user_ids_with_access:
719 return []
721 return await Users.get_users_by_user_ids(list(user_ids_with_access), db=db)
723 def has_permission_filter(
724 self,
725 db,
726 query,
727 DocumentModel,
728 filter: dict,
729 resource_type: str,
730 permission: str = 'read',
731 ):
732 """
733 Apply access control filtering to a SQLAlchemy query by JOINing with access_grant.
735 This replaces the old JSON-column-based filtering with a proper relational JOIN.
737 Note: This method builds SQLAlchemy expressions and does NOT perform I/O itself,
738 so it remains synchronous. The caller is responsible for executing the query
739 asynchronously with `await db.execute(...)`.
740 """
741 group_ids = filter.get('group_ids', [])
742 user_id = filter.get('user_id')
744 if permission == 'read_only': 744 ↛ 745line 744 didn't jump to line 745 because the condition on line 744 was never true
745 return self._has_read_only_permission_filter(db, query, DocumentModel, filter, resource_type)
747 # Build principal conditions
748 principal_conditions = []
750 if group_ids or user_id:
751 # Public access: user:* read
752 principal_conditions.append(
753 and_(
754 AccessGrant.principal_type == 'user',
755 AccessGrant.principal_id == '*',
756 )
757 )
759 if user_id:
760 # Owner always has access
761 principal_conditions.append(DocumentModel.user_id == user_id)
763 # Direct user grant
764 principal_conditions.append(
765 and_(
766 AccessGrant.principal_type == 'user',
767 AccessGrant.principal_id == user_id,
768 )
769 )
771 if group_ids: 771 ↛ 773line 771 didn't jump to line 773 because the condition on line 771 was never true
772 # Group grants
773 principal_conditions.append(
774 and_(
775 AccessGrant.principal_type == 'group',
776 AccessGrant.principal_id.in_(group_ids),
777 )
778 )
780 if not principal_conditions:
781 return query
783 # LEFT JOIN access_grant and filter
784 # We use a subquery approach to avoid duplicates from multiple matching grants
785 from sqlalchemy import exists as sa_exists
787 grant_exists = (
788 select(AccessGrant.id)
789 .where(
790 AccessGrant.resource_type == resource_type,
791 AccessGrant.resource_id == DocumentModel.id,
792 AccessGrant.permission == permission,
793 or_(
794 and_(
795 AccessGrant.principal_type == 'user',
796 AccessGrant.principal_id == '*',
797 ),
798 *(
799 [
800 and_(
801 AccessGrant.principal_type == 'user',
802 AccessGrant.principal_id == user_id,
803 )
804 ]
805 if user_id
806 else []
807 ),
808 *(
809 [
810 and_(
811 AccessGrant.principal_type == 'group',
812 AccessGrant.principal_id.in_(group_ids),
813 )
814 ]
815 if group_ids
816 else []
817 ),
818 ),
819 )
820 .correlate(DocumentModel)
821 .exists()
822 )
824 # Owner OR has a matching grant
825 owner_or_grant = [grant_exists]
826 if user_id: 826 ↛ 829line 826 didn't jump to line 829 because the condition on line 826 was always true
827 owner_or_grant.append(DocumentModel.user_id == user_id)
829 query = query.filter(or_(*owner_or_grant))
830 return query
832 def _has_read_only_permission_filter(
833 self,
834 db,
835 query,
836 DocumentModel,
837 filter: dict,
838 resource_type: str,
839 ):
840 """
841 Filter for items where user has read BUT NOT write access.
842 A public (user:*) read grant counts as read access, so publicly shared
843 read-only items are listed rather than being reachable only by direct link.
845 Note: This method builds SQLAlchemy expressions and does NOT perform I/O itself,
846 so it remains synchronous. The caller is responsible for executing the query
847 asynchronously with `await db.execute(...)`.
848 """
849 group_ids = filter.get('group_ids', [])
850 user_id = filter.get('user_id')
852 from sqlalchemy import exists as sa_exists
854 read_grant_exists = (
855 select(AccessGrant.id)
856 .where(
857 AccessGrant.resource_type == resource_type,
858 AccessGrant.resource_id == DocumentModel.id,
859 AccessGrant.permission == 'read',
860 or_(
861 and_(
862 AccessGrant.principal_type == 'user',
863 AccessGrant.principal_id == '*',
864 ),
865 *(
866 [
867 and_(
868 AccessGrant.principal_type == 'user',
869 AccessGrant.principal_id == user_id,
870 )
871 ]
872 if user_id
873 else []
874 ),
875 *(
876 [
877 and_(
878 AccessGrant.principal_type == 'group',
879 AccessGrant.principal_id.in_(group_ids),
880 )
881 ]
882 if group_ids
883 else []
884 ),
885 ),
886 )
887 .correlate(DocumentModel)
888 .exists()
889 )
891 write_grant_exists = (
892 select(AccessGrant.id)
893 .where(
894 AccessGrant.resource_type == resource_type,
895 AccessGrant.resource_id == DocumentModel.id,
896 AccessGrant.permission == 'write',
897 or_(
898 and_(
899 AccessGrant.principal_type == 'user',
900 AccessGrant.principal_id == '*',
901 ),
902 *(
903 [
904 and_(
905 AccessGrant.principal_type == 'user',
906 AccessGrant.principal_id == user_id,
907 )
908 ]
909 if user_id
910 else []
911 ),
912 *(
913 [
914 and_(
915 AccessGrant.principal_type == 'group',
916 AccessGrant.principal_id.in_(group_ids),
917 )
918 ]
919 if group_ids
920 else []
921 ),
922 ),
923 )
924 .correlate(DocumentModel)
925 .exists()
926 )
928 conditions = [read_grant_exists, ~write_grant_exists]
930 # Not owner
931 if user_id:
932 conditions.append(DocumentModel.user_id != user_id)
934 query = query.filter(and_(*conditions))
935 return query
938AccessGrants = AccessGrantsTable()