Implement per site restriction for users

HISTORY.rst

@@ -15,6 +15,7 @@ History
 * Removed meta-mixin compatibility code
 * Changed slug size to 255 chars
 * Fixed pagination setting in list views
+* Added API to set default sites if user has permission only for a subset of sites
 
 0.7.0 (2016-03-19)
 ++++++++++++++++++

README.rst

@@ -340,6 +340,28 @@ To add the blog Sitemap, add the following code to the project ``urls.py``::
         }),
     )
 
+Multisite
++++++++++
+
+django CMS blog provides full support for multisite setups.
+
+Each blog post can be assigned to none, one or more sites: if no site is selected, then
+it's visible on all sites.
+
+This is matched with and API that allows to restrict users to only be able to edit
+blog posts only for some sites.
+
+To implement this API, you must add a ``get_sites`` method on the user model which
+returns a queryset of sites the user is allowed to add posts to.
+
+Example::
+
+    class CustomUser(AbstractUser):
+        sites = models.ManyToManyField('sites.Site')
+
+        def get_sites(self):
+            return self.sites
+
 
 django CMS 3.2+ Wizard
 ++++++++++++++++++++++

cms_helper.py

@@ -2,6 +2,8 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import, print_function, unicode_literals
 
+import os
+
 from tempfile import mkdtemp
 
 
@@ -20,7 +22,6 @@ def gettext(s): return s
         'taggit',
         'taggit_autosuggest',
         'aldryn_apphooks_config',
-        'tests.test_utils',
         'aldryn_search',
     ],
     LANGUAGE_CODE='en',
@@ -91,7 +92,7 @@ def gettext(s): return s
     SITE_ID=1,
     HAYSTACK_CONNECTIONS={
         'default': {}
-    }
+    },
 )
 
 try:
@@ -124,9 +125,9 @@ def gettext(s): return s
             'ROUTING': 'knocker.routing.channel_routing',
         },
     }
-
 except ImportError:
     pass
+os.environ['AUTH_USER_MODEL'] = 'tests.test_utils.CustomUser'
 
 
 def run():

djangocms_blog/admin.py

@@ -75,6 +75,7 @@ class PostAdmin(PlaceholderAdminMixin, FrontendEditableAdminMixin,
     app_config_values = {
         'default_published': 'publish'
     }
+    _sites = None
 
     def get_urls(self):
         """
@@ -90,6 +91,7 @@ def get_urls(self):
     def publish_post(self, request, pk):
         """
         Admin view to publish a single post
+
         :param request: request
         :param pk: primary key of the post to publish
         :return: Redirect to the post itself (if found) or fallback urls
@@ -116,9 +118,47 @@ def formfield_for_dbfield(self, db_field, **kwargs):
             field.max_length = 70
         return field
 
+    def has_restricted_sites(self, request):
+        """
+        Whether the current user has permission on one site only
+
+        :param request: current request
+        :return: boolean: user has permission on only one site
+        """
+        sites = self.get_restricted_sites(request)
+        return sites and sites.count() == 1
+
+    def get_restricted_sites(self, request):
+        """
+        The sites on which the user has permission on.
+
+        To return the permissions, the method check for the ``get_sites``
+        method on the user instance (e.g.: ``return request.user.get_sites()``)
+        which must return the queryset of enabled sites.
+        If the attribute does not exists, the user is considered enabled
+        for all the websites.
+
+        :param request: current request
+        :return: boolean or a queryset of available sites
+        """
+        if self._sites is None:
+            try:
+                self._sites = request.user.get_sites()
+            except AttributeError:  # pragma: no cover
+                self._sites = False
+        return self._sites
+
+    def _set_config_defaults(self, request, form, obj=None):
+        form = super(PostAdmin, self)._set_config_defaults(request, form, obj)
+        sites = self.get_restricted_sites(request)
+        if 'sites' in form.base_fields and sites.exists():
+            form.base_fields['sites'].queryset = self.get_restricted_sites(request).all()
+        return form
+
     def get_fieldsets(self, request, obj=None):
         """
         Customize the fieldsets according to the app settings
+
         :param request: request
         :param obj: post
         :return: fieldsets configuration
@@ -142,7 +182,7 @@ def get_fieldsets(self, request, obj=None):
                 fsets[0][1]['fields'].append('abstract')
             if not get_setting('USE_PLACEHOLDER'):
                 fsets[0][1]['fields'].append('post_text')
-        if get_setting('MULTISITE'):
+        if get_setting('MULTISITE') and not self.has_restricted_sites(request):
             fsets[1][1]['fields'][0].append('sites')
         if request.user.is_superuser:
             fsets[1][1]['fields'][0].append('author')
@@ -158,6 +198,19 @@ def save_model(self, request, obj, form, change):
         obj._set_default_author(request.user)
         super(PostAdmin, self).save_model(request, obj, form, change)
 
+    def get_queryset(self, request):
+        qs = super(PostAdmin, self).get_queryset(request)
+        sites = self.get_restricted_sites(request)
+        if sites.exists():
+            qs = qs.filter(sites__in=sites.all())
+        return qs
+
+    def save_related(self, request, form, formsets, change):
+        super(PostAdmin, self).save_related(request, form, formsets, change)
+        obj = form.instance
+        sites = self.get_restricted_sites(request)
+        obj.sites = sites.all()
+
     class Media:
         css = {
             'all': ('%sdjangocms_blog/css/%s' % (settings.STATIC_URL, 'djangocms_blog_admin.css'),)

tests/test_models.py

@@ -164,19 +164,60 @@ def test_admin_fieldsets(self):
             fsets = post_admin.get_fieldsets(request)
             self.assertFalse('sites' in fsets[1][1]['fields'][0])
 
-        request = self.get_page_request('/', self.user, r'/en/blog/?app_config=%s' % self.app_config_1.pk, edit=False)
+        request = self.get_page_request(
+            '/', self.user, r'/en/blog/?app_config=%s' % self.app_config_1.pk, edit=False
+        )
         fsets = post_admin.get_fieldsets(request)
         self.assertTrue('author' in fsets[1][1]['fields'][0])
 
         with self.login_user_context(self.user):
-            request = self.get_request('/', 'en', user=self.user, path=r'/en/blog/?app_config=%s' % self.app_config_1.pk)
+            request = self.get_request(
+                '/', 'en', user=self.user, path=r'/en/blog/?app_config=%s' % self.app_config_1.pk
+            )
             msg_mid = MessageMiddleware()
             msg_mid.process_request(request)
             post_admin = admin.site._registry[Post]
             response = post_admin.add_view(request)
             self.assertContains(response, '<option value="%s">%s</option>' % (
                 self.category_1.pk, self.category_1.safe_translation_getter('name', language_code='en')
             ))
+            self.assertContains(response, 'id="id_sites" name="sites"')
+
+        self.user.sites.add(self.site_1)
+        with self.login_user_context(self.user):
+            request = self.get_request('/', 'en', user=self.user,
+                                       path=r'/en/blog/?app_config=%s' % self.app_config_1.pk)
+            msg_mid = MessageMiddleware()
+            msg_mid.process_request(request)
+            post_admin = admin.site._registry[Post]
+            post_admin._sites = None
+            response = post_admin.add_view(request)
+            response.render()
+            self.assertNotContains(response, 'id="id_sites" name="sites"')
+            post_admin._sites = None
+        self.user.sites.clear()
+
+    def test_admin_queryset(self):
+        posts = self.get_posts()
+        posts[0].sites.add(self.site_1)
+        posts[1].sites.add(self.site_2)
+
+        request = self.get_request('/', 'en', user=self.user,
+                                   path=r'/en/blog/?app_config=%s' % self.app_config_1.pk)
+        post_admin = admin.site._registry[Post]
+        post_admin._sites = None
+        qs = post_admin.get_queryset(request)
+        self.assertEqual(qs.count(), 4)
+
+        self.user.sites.add(self.site_2)
+        request = self.get_request('/', 'en', user=self.user,
+                                   path=r'/en/blog/?app_config=%s' % self.app_config_1.pk)
+        post_admin = admin.site._registry[Post]
+        post_admin._sites = None
+        qs = post_admin.get_queryset(request)
+        self.assertEqual(qs.count(), 1)
+        self.assertEqual(qs[0], posts[1])
+        self.user.sites.clear()
 
     def test_admin_auto_author(self):
         pages = self.get_pages()
@@ -235,17 +276,26 @@ def test_admin_fieldsets_filter(self):
         post_admin = admin.site._registry[Post]
         request = self.get_page_request('/', self.user_normal, r'/en/blog/?app_config=%s' % self.app_config_1.pk)
 
+        post_admin._sites = None
         fsets = post_admin.get_fieldsets(request)
         self.assertFalse('author' in fsets[1][1]['fields'][0])
+        self.assertTrue('sites' in fsets[1][1]['fields'][0])
+        post_admin._sites = None
 
         def filter_function(fs, request, obj=None):
             if request.user == self.user_normal:
                 fs[1][1]['fields'][0].append('author')
             return fs
 
+        self.user_normal.sites.add(self.site_1)
+        request = self.get_page_request('/', self.user_normal, r'/en/blog/?app_config=%s' % self.app_config_1.pk)
+        post_admin._sites = None
         with self.settings(BLOG_ADMIN_POST_FIELDSET_FILTER=filter_function):
             fsets = post_admin.get_fieldsets(request)
             self.assertTrue('author' in fsets[1][1]['fields'][0])
+            self.assertFalse('sites' in fsets[1][1]['fields'][0])
+        post_admin._sites = None
+        self.user_normal.sites.clear()
 
     def test_admin_post_text(self):
         pages = self.get_pages()

tests/test_utils/admin.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+from django.contrib import admin
+from django.contrib.auth.admin import UserAdmin
+from django.utils.translation import ugettext_lazy as _
+
+from .models import CustomUser
+
+
+class CustomUserAdmin(UserAdmin):
+    model = CustomUser
+
+
+    fieldsets = (
+        (None, {'fields': ('username', 'password')}),
+        (_('Personal info'), {'fields': ('first_name', 'last_name', 'email')}),
+        (_('Permissions'), {'fields': ('is_active', 'is_staff', 'is_superuser',
+                                       'groups', 'user_permissions')}),
+        (_('Important dates'), {'fields': ('last_login', 'date_joined')}),
+        (_('Sites'), {'fields': ('sites',)})
+    )
+
+admin.site.register(CustomUser, CustomUserAdmin)

tests/test_utils/migrations/0001_initial.py

@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+import django.core.validators
+import django.utils.timezone
+import django.contrib.auth.models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('auth', '0006_require_contenttypes_0002'),
+        ('sites', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='CustomUser',
+            fields=[
+                ('id', models.AutoField(serialize=False, auto_created=True, primary_key=True, verbose_name='ID')),
+                ('password', models.CharField(verbose_name='password', max_length=128)),
+                ('last_login', models.DateTimeField(verbose_name='last login', null=True, blank=True)),
+                ('is_superuser', models.BooleanField(verbose_name='superuser status', help_text='Designates that this user has all permissions without explicitly assigning them.', default=False)),
+                ('username', models.CharField(help_text='Required. 30 characters or fewer. Letters, digits and @/./+/-/_ only.', unique=True, verbose_name='username', validators=[django.core.validators.RegexValidator('^[\\w.@+-]+$', 'Enter a valid username. This value may contain only letters, numbers and @/./+/-/_ characters.', 'invalid')], max_length=30, error_messages={'unique': 'A user with that username already exists.'})),
+                ('first_name', models.CharField(verbose_name='first name', max_length=30, blank=True)),
+                ('last_name', models.CharField(verbose_name='last name', max_length=30, blank=True)),
+                ('email', models.EmailField(verbose_name='email address', max_length=254, blank=True)),
+                ('is_staff', models.BooleanField(verbose_name='staff status', help_text='Designates whether the user can log into this admin site.', default=False)),
+                ('is_active', models.BooleanField(verbose_name='active', help_text='Designates whether this user should be treated as active. Unselect this instead of deleting accounts.', default=True)),
+                ('date_joined', models.DateTimeField(verbose_name='date joined', default=django.utils.timezone.now)),
+                ('groups', models.ManyToManyField(help_text='The groups this user belongs to. A user will get all permissions granted to each of their groups.', related_query_name='user', related_name='user_set', verbose_name='groups', blank=True, to='auth.Group')),
+                ('sites', models.ManyToManyField(to='sites.Site')),
+                ('user_permissions', models.ManyToManyField(help_text='Specific permissions for this user.', related_query_name='user', related_name='user_set', verbose_name='user permissions', blank=True, to='auth.Permission')),
+            ],
+            options={
+                'verbose_name': 'user',
+                'abstract': False,
+                'verbose_name_plural': 'users',
+            },
+            managers=[
+                ('objects', django.contrib.auth.models.UserManager()),
+            ],
+        ),
+    ]

tests/test_utils/models.py

@@ -1 +1,10 @@
 # -*- coding: utf-8 -*-
+from django.contrib.auth.models import AbstractUser
+from django.db import models
+
+
+class CustomUser(AbstractUser):
+    sites = models.ManyToManyField('sites.Site')
+
+    def get_sites(self):
+        return self.sites