# ruff: noqa: W605 import functools import operator from django.contrib.contenttypes.models import ContentType from django.db import models from django.db.models import Q from cms.constants import ROOT_USER_LEVEL from cms.exceptions import NoPermissionsException from cms.models.query import PageQuerySet from cms.utils.compat.warnings import RemovedInDjangoCMS60Warning from cms.utils.i18n import get_fallback_languages class PageManager(models.Manager): def get_queryset(self): """Change standard model queryset to our own. """ return PageQuerySet(self.model) # Just some of the queryset methods are implemented here, access queryset # for more getting more supporting methods. def on_site(self, site=None): return self.get_queryset().on_site(site) def get_home(self, site=None): return self.get_queryset().get_home(site) def search(self, q, language=None, current_site_only=True, site=None): """Simple search function Plugins can define a 'search_fields' tuple similar to ModelAdmin classes """ from cms.plugin_pool import plugin_pool qs = self.get_queryset() if site or current_site_only: qs = qs.on_site(site) qt = Q(pagecontent_set__title__icontains=q) # find 'searchable' plugins and build query qp = Q() plugins = plugin_pool.registered_plugins for plugin in plugins: cmsplugin = plugin.model if not ( hasattr(cmsplugin, 'search_fields') and hasattr(cmsplugin, 'cmsplugin_ptr') ): continue field = cmsplugin.cmsplugin_ptr.field related_query_name = field.related_query_name() if related_query_name and not related_query_name.startswith('+'): for field in cmsplugin.search_fields: qp |= Q(**{ f'pagecontent_set__placeholders__cmsplugin__{related_query_name}__{field}__icontains': q}) if language: qt &= Q(pagecontent_set__language=language) qp &= Q(pagecontent_set__placeholders__cmsplugin__language=language) qs = qs.filter(qt | qp) return qs.distinct() class WithUserMixin: """empty mixin adds with_user """ def with_user(self, user): return self class PageUrlManager(WithUserMixin, models.Manager): def get_for_site(self, site, **kwargs): kwargs['site'] = site return self.filter(**kwargs) class PageContentManager(WithUserMixin, models.Manager): def get_title(self, page, language, language_fallback=False): """ Gets the latest content for a particular page and language. Falls back to another language if wanted. """ import warnings warnings.warn( "PageContentManager.get_title is deprecated and will be removed in a future release. " "Please use page.get_page_content() or page.get_admin_content() instead.", RemovedInDjangoCMS60Warning, stacklevel=2, ) try: title = self.get(language=language, page=page) return title except self.model.DoesNotExist: if language_fallback: try: titles = self.filter(page=page) fallbacks = get_fallback_languages(language, site_id=page.site_id) for lang in fallbacks: for title in titles: if lang == title.language: return title return None except self.model.DoesNotExist: pass else: raise return None class ContentAdminQuerySet(models.QuerySet): def current_content(self, **kwargs): """If a versioning package is installed, this returns the currently valid content that matches the filter given in kwargs. Used to find content to be copied, e.g.. Without versioning every page is current.""" return self.filter(**kwargs) def latest_content(self, **kwargs): """If a versioning package is installed, returns the latest version that matches the filter given in kwargs including discarded or unpublished page content. Without versioning every page content is the latest.""" return self.filter(**kwargs) class ContentAdminManager(WithUserMixin, models.Manager): def get_queryset(self): return ContentAdminQuerySet(self.model, using=self._db) def current_content(self, **kwargs): """Syntactic sugar: admin_manager.current_content()""" return self.get_queryset().current_content(**kwargs) def latest_content(self, **kwargs): """Syntactic sugar: admin_manager.latest_content()""" return self.get_queryset().latest_content(**kwargs) class PlaceholderForObjQS(models.QuerySet): # This Queryset prepopulates the source field cache when the get_for_obj manager is used # __iter__ is called when the queryset has been evaluated and is iterated # get prepopulates the cache for any get requests _source_object = None def __init__(self, *args, **kwargs): self._source_object = kwargs.pop('source_object', self._source_object) super().__init__(*args, **kwargs) def __iter__(self): if self._fields: # values_list etc yield from super().__iter__() else: # Placeholder objects for obj in super().__iter__(): obj._state.fields_cache["source"] = self._source_object yield obj def get(self, *args, **kwargs): obj = super().get(*args, **kwargs) obj._state.fields_cache["source"] = self._source_object return obj def _chain(self): # Also clone source_object when chaining querysets! clone = super()._chain() clone._source_object = self._source_object return clone class PlaceholderManager(models.Manager): def get_for_obj(self, obj): """ Get all placeholders for given object """ content_type = ContentType.objects.get_for_model(obj) return (PlaceholderForObjQS(source_object=obj, model=self.model, using=self._db, hints=self._hints) .filter(content_type=content_type, object_id=obj.pk)) ################################################################################ # Permissions ################################################################################ class BasicPagePermissionManager(models.Manager): """Global page permission manager accessible under objects. !IMPORTANT: take care, PagePermissionManager and GlobalPagePermissionManager both inherit from this manager """ def with_user(self, user): """Get all objects for given user, also takes look if user is in some group. """ return self.filter(Q(user=user) | Q(group__user=user)) def get_with_permission(self, user, site_id, perm): raise NotImplementedError def get_with_add_pages_permission(self, user, site_id): return self.get_with_permission(user, site_id, 'can_add') def get_with_change_pages_permission(self, user, site_id): return self.get_with_permission(user, site_id, 'can_change') def get_with_change_permissions(self, user, site_id): return self.get_with_permission(user, site_id, 'can_change_permissions') def get_with_view_permissions(self, user, site_id): return self.get_with_permission(user, site_id, 'can_view') class GlobalPagePermissionManager(BasicPagePermissionManager): def get_with_site(self, user, site_id): # if the user has add rights to this site explicitly this_site = Q(sites__in=[site_id]) # if the user can add to all sites all_sites = Q(sites__isnull=True) return self.with_user(user).filter(this_site | all_sites) def get_with_permission(self, user, site_id, perm): """ Provide a single point of entry for deciding whether any given global permission exists. """ # if the user has add rights to this site explicitly this_site = Q(**{perm: True, 'sites__in': [site_id]}) # if the user can add to all sites all_sites = Q(**{perm: True, 'sites__isnull': True}) return self.with_user(user).filter(this_site | all_sites) def user_has_permissions(self, user, site_id, perms): # if the user has add rights to this site explicitly this_site = Q(sites__in=[site_id]) # if the user can add to all sites all_sites = Q(sites__isnull=True) queryset = self.with_user(user).filter(this_site | all_sites) queries = [Q(**{perm: True}) for perm in perms] return queryset.filter(functools.reduce(operator.or_, queries)).exists() class PagePermissionManager(BasicPagePermissionManager): """Page permission manager accessible under objects. """ def get_with_permission(self, user, site_id, perm): """ Provide a single point of entry for deciding whether any given global permission exists. """ query = {perm: True, 'page__site': site_id} return self.with_user(user).filter(**query) def get_with_site(self, user, site_id): return self.with_user(user).filter(page__site=site_id) def user_has_permissions(self, user, site_id, perms): queryset = self.with_user(user).filter(page__site=site_id) queries = [Q(**{perm: True}) for perm in perms] return queryset.filter(functools.reduce(operator.or_, queries)).exists() def subordinate_to_user(self, user, site): r""" Get all page permission objects on which user/group is lover in hierarchy then given user and given user can change permissions on them. !IMPORTANT, but exclude objects with given user, or any group containing this user - he can't be able to change his own permissions, because if he does, and removes some permissions from himself, he will not be able to add them anymore. Example: A / \ user B,E / \ C,X D,Y Gives permission nodes C,X,D,Y under user, so he can edit permissions if he haves can_change_permission. Example: A,Y / \ user B,E,X / \ C,X D,Y Gives permission nodes C,D under user, so he can edit, but not anymore to X,Y, because this users are on the same level or higher in page hierarchy. (but only if user have can_change_permission) Example: A / \ user B,E / \ \ C,X D,Y user / \ I J,A User permissions can be assigned to multiple page nodes, so merge of all of them is required. In this case user can see permissions for users C,X,D,Y,I,J but not A, because A user in higher in hierarchy. If permission object holds group, this permission object can be visible to user only if all the group members are lover in hierarchy. If any of members is higher than given user, this entry must stay invisible. If user is superuser, or haves global can_change_permission permissions, show him everything. Result of this is used in admin for page permissions inline. """ # get user level from cms.utils.page_permissions import get_change_permissions_perm_tuples from cms.utils.permissions import get_user_permission_level try: user_level = get_user_permission_level(user, site) except NoPermissionsException: return self.none() if user_level == ROOT_USER_LEVEL: return self.all() # get all permissions from cms.models import PermissionTuple allow_list = Q() for perm_tuple in get_change_permissions_perm_tuples(user, site, check_global=False): allow_list |= PermissionTuple(perm_tuple).allow_list("page") # get permission set, but without objects targeting user, or any group # in which he can be qs = self.filter( allow_list, page__depth__gte=user_level, ) qs = qs.exclude(user=user).exclude(group__user=user) return qs def for_page(self, page): """Returns queryset containing all instances somehow connected to given page. This includes permissions to page itself and permissions inherited from higher pages. NOTE: this returns just PagePermission instances, to get complete access list merge return of this function with Global permissions. """ # permissions should be managed on the draft page only from cms.models import ( ACCESS_CHILDREN, ACCESS_DESCENDANTS, ACCESS_PAGE, ACCESS_PAGE_AND_CHILDREN, ACCESS_PAGE_AND_DESCENDANTS, ) paths = page.get_ancestor_paths() # Ancestors query = ( Q(page__path__in=paths) & (Q(grant_on=ACCESS_DESCENDANTS) | Q(grant_on=ACCESS_PAGE_AND_DESCENDANTS)) ) if page.parent: # Direct parent query |= ( Q(page=page.parent) & (Q(grant_on=ACCESS_CHILDREN) | Q(grant_on=ACCESS_PAGE_AND_CHILDREN)) ) query |= Q(page=page) & ( Q(grant_on=ACCESS_PAGE_AND_DESCENDANTS) | Q(grant_on=ACCESS_PAGE_AND_CHILDREN) | Q(grant_on=ACCESS_PAGE) ) return self.filter(query).order_by('page__depth')