feat: Add permissions to subscriptions to only exposed user's

djangoldp_notification/models.py

@@ -1,25 +1,20 @@
 import logging
-from threading import Thread
 
 import requests
 from django.conf import settings
-from django.contrib.admin.models import LogEntry
-from django.contrib.sessions.models import Session
 from django.core.mail import send_mail
 from django.db import models
 from django.db.models.signals import post_save
 from django.dispatch import receiver
-from oidc_provider.models import Token
+from django.template import loader
 from django.urls import NoReverseMatch
 from django.utils.translation import ugettext_lazy as _
-
 from djangoldp.fields import LDPUrlField
 from djangoldp.models import Model
-
-from django.template import loader
+from threading import Thread
 
 from djangoldp_notification.middlewares import MODEL_MODIFICATION_USER_FIELD
-from .permissions import InboxPermissions
+from djangoldp_notification.permissions import InboxPermissions, SubscriptionsPermissions
 
 
 class Notification(Model):
@@ -50,6 +45,11 @@ class Subscription(Model):
     def __str__(self):
         return '{}'.format(self.object)
 
+    class Meta(Model.Meta):
+        anonymous_perms = []
+        authenticated_perms = ["add", "view", "delete"]
+        permission_classes = [SubscriptionsPermissions]
+
 # --- SUBSCRIPTION SYSTEM ---
 @receiver(post_save, dispatch_uid="callback_notif")
 def send_notification(sender, instance, created, **kwargs):

djangoldp_notification/permissions.py

@@ -1,4 +1,6 @@
+from django.contrib.auth import get_user_model
 from djangoldp.permissions import LDPPermissions
+from rest_framework.reverse import reverse
 
 
 class InboxPermissions(LDPPermissions):
@@ -35,4 +37,26 @@ class InboxPermissions(LDPPermissions):
             if not perm.split('.')[1].split('_')[0] in self.user_permissions(request.user, model, obj):
                 return False
 
-        return True
+        return True
+
+
+class SubscriptionsPermissions(LDPPermissions):
+    def has_permission(self, request, view):
+        if request.user.is_anonymous and not request.method == "OPTIONS":
+            return False
+
+        if request.method in ["GET", "PATCH", "DELETE", "PUT"]:
+            return True
+
+        return super().has_permission(request, view)
+
+    def has_object_permission(self, request, view, obj):
+        if request.user.is_anonymous and not request.method == "OPTIONS":
+            return False
+
+        reverse_path_key = "{}-notification-list".format(get_user_model()._meta.object_name.lower())
+        user_inbox = reverse(reverse_path_key, kwargs={"slug": request.user.slug}, request=request)
+        if obj.inbox == user_inbox:
+            return True
+
+        return False

djangoldp_notification/tests/conftest.py

@@ -0,0 +1,15 @@
+from importlib import import_module
+from os import environ
+
+from pytest import fail
+
+settings_module = environ.get("DJANGO_SETTINGS_MODULE")
+if settings_module is None or len(settings_module) == 0:
+    fail("DJANGO_SETTINGS_MODULE needs to be defined and point to your SIB app installation settings")
+
+try:
+    import_module(settings_module)
+except ImportError:
+    initial_module = [token for token in settings_module.split(".") if len(token) > 0][0]
+    fail("Unable to import {}. Try to configure PYTHONPATH to point the "
+         "directory containing the {} module".format(settings_module, initial_module))

djangoldp_notification/tests/test_models.py

@@ -0,0 +1,47 @@
+from collections import OrderedDict
+
+from djangoldp.factories import UserFactory
+from test_plus import APITestCase
+
+from djangoldp_notification.models import Subscription
+
+
+class TestSubscription(APITestCase):
+    user1 = None
+    user2 = None
+    circle_user1_url = "http://localhost:8000/circles/1"
+    circle_user2_url = "http://localhost:8000/circles/2"
+
+    def setUp(self):
+        self.user1 = UserFactory(username="karl_marx", password="password")
+        Subscription.objects.create(object=self.circle_user1_url, inbox="http://testserver/users/karl_marx/inbox/")
+
+        self.user2 = UserFactory(username="piotr_kropotkine", password="password")
+        Subscription.objects.create(object=self.circle_user2_url,
+                                    inbox="http://testserver/users/piotr_kropotkine/inbox/")
+
+    def test_not_logged_fails(self):
+        response = self.get("/subscriptions/")
+        self.assert_http_403_forbidden(response)
+        self.assertEqual(response.data.get("detail"), "Authentication credentials were not provided.")
+
+    def test_logged_in_succeeds(self):
+        with self.login(self.user1):
+            result = self.get("/subscriptions/").data.get("ldp:contains")
+            expected = [OrderedDict({
+                "@id": "http://localhost:8000/subscriptions/1/",
+                "object": self.circle_user1_url,
+                "inbox": "http://testserver/users/karl_marx/inbox/",
+                "permissions": [{'mode': {'@type': 'view'}}, {'mode': {'@type': 'delete'}}]
+            })]
+            self.assertSequenceEqual(result, expected)
+
+        with self.login(self.user2):
+            result = self.get("/subscriptions/").data.get("ldp:contains")
+            expected = [OrderedDict({
+                "@id": "http://localhost:8000/subscriptions/2/",
+                "object": self.circle_user2_url,
+                "inbox": "http://testserver/users/piotr_kropotkine/inbox/",
+                "permissions": [{'mode': {'@type': 'view'}}, {'mode': {'@type': 'delete'}}]
+            })]
+            self.assertSequenceEqual(result, expected)