Add permission checks to markNotificationAsRead + deleteNotification

backend/dataall/modules/notifications/db/notification_repositories.py

@@ -3,7 +3,31 @@
 from sqlalchemy import func, and_, or_
 
 from dataall.modules.notifications.db import notification_models as models
-from dataall.base.db import paginate
+from dataall.base.db import paginate, exceptions
+from dataall.base.context import get_context
+from functools import wraps
+
+
+class NotificationAccess:
+    @staticmethod
+    def is_recipient(f):
+        @wraps(f)
+        def wrapper(*args, **kwds):
+            uri = kwds.get('notificationUri')
+            if not uri:
+                raise KeyError(f"{f.__name__} doesn't have parameter uri.")
+            context = get_context()
+            with context.db_engine.scoped_session() as session:
+                notification = session.query(models.Notification).get(uri)
+                if notification and (notification.recipient in context.groups + [context.username]):
+                    return f(*args, **kwds)
+                else:
+                    raise exceptions.UnauthorizedOperation(
+                        action='UPDATE NOTIFICATION',
+                        message=f'User {context.username} is not the recipient user/group of the notification {uri}',
+                    )
+
+        return wrapper
 
 
 class NotificationRepository:
@@ -88,13 +112,15 @@ def count_deleted_notifications(session, username, groups):
         return int(count)
 
     @staticmethod
+    @NotificationAccess.is_recipient
     def read_notification(session, notificationUri):
         notification = session.query(models.Notification).get(notificationUri)
         notification.is_read = True
         session.commit()
         return True
 
     @staticmethod
+    @NotificationAccess.is_recipient
     def delete_notification(session, notificationUri):
         notification = session.query(models.Notification).get(notificationUri)
         if notification: