Repository: openwisp/openwisp-network-topology (language: python)

Issue:
[bug] Consumer doesn't seem to be checking if user is authorized
I think the following code is missing whether the user has permissions to view the Topology object and whether it's manager of the organization: https://github.com/openwisp/openwisp-network-topology/blob/da8e099c7c8502a395dbab116affe3b2f7730177/openwisp_network_topology/consumers.py#L10-L29
[bug] Consumer doesn't seem to be checking if user is authorized
I think the following code is missing whether the user has permissions to view the Topology object and whether it's manager of the organization: https://github.com/openwisp/openwisp-network-topology/blob/da8e099c7c8502a395dbab116affe3b2f7730177/openwisp_network_topology/consumers.py#L10-L29


Discussion / hints:

Fix patch:
diff --git a/openwisp_network_topology/consumers.py b/openwisp_network_topology/consumers.py
index 2a913132..73eb3c97 100644
--- a/openwisp_network_topology/consumers.py
+++ b/openwisp_network_topology/consumers.py
@@ -2,31 +2,41 @@
 
 from asgiref.sync import async_to_sync
 from channels.generic.websocket import WebsocketConsumer
+from django.core.exceptions import ValidationError
+from swapper import load_model
+
+from . import settings as app_settings
+
+Topology = load_model('topology', 'Topology')
 
 
 class TopologyConsumer(WebsocketConsumer):
     channel_layer_group = 'topology'
 
-    def _is_user_authenticated(self):
+    def _is_user_authorized_to_view_topology(self, user, topology_pk):
         try:
-            assert self.scope['user'].is_authenticated is True
-        except (KeyError, AssertionError):
-            self.close()
+            topology = Topology.objects.get(pk=topology_pk)
+        except (Topology.DoesNotExist, ValidationError):
             return False
-        else:
+        if not app_settings.TOPOLOGY_API_AUTH_REQUIRED:
             return True
+        return user.is_superuser or (
+            user.is_authenticated
+            and user.is_manager(topology.organization_id)
+            and user.has_perm(f'{Topology._meta.app_label}.view_topology')
+        )
 
     def connect(self):
-        try:
-            assert self._is_user_authenticated()
-            self.pk = self.scope['url_route']['kwargs']['pk']
-        except (AssertionError, KeyError):
-            self.close()
-        else:
+        user, topology_pk = self.scope.get('user'), self.scope.get('url_route').get(
+            'kwargs'
+        ).get('pk')
+        if self._is_user_authorized_to_view_topology(user, topology_pk):
             async_to_sync(self.channel_layer.group_add)(
-                f"{self.channel_layer_group}-{self.pk}", self.channel_name
+                f"{self.channel_layer_group}-{topology_pk}", self.channel_name
             )
             self.accept()
+        else:
+            self.close()
 
     def disconnect(self, close_code):
         self.close()


Test patch:
diff --git a/openwisp_network_topology/tests/test_websockets.py b/openwisp_network_topology/tests/test_websockets.py
index 864e0970..d221010b 100644
--- a/openwisp_network_topology/tests/test_websockets.py
+++ b/openwisp_network_topology/tests/test_websockets.py
@@ -1,11 +1,18 @@
+from unittest.mock import patch
+from uuid import uuid4
+
 import pytest
 from channels.db import database_sync_to_async
 from channels.testing import WebsocketCommunicator
 from django.conf import settings
+from django.contrib.auth.models import Permission
 from django.utils.module_loading import import_string
 from swapper import load_model
 
-from .utils import CreateGraphObjectsMixin, CreateOrgMixin
+from openwisp_users.tests.utils import TestOrganizationMixin
+
+from .. import settings as app_settings
+from .utils import CreateGraphObjectsMixin
 
 Topology = load_model('topology', 'Topology')
 Node = load_model('topology', 'Node')
@@ -14,7 +21,7 @@
 
 @pytest.mark.asyncio
 @pytest.mark.django_db(transaction=True)
-class TestTopologySockets(CreateGraphObjectsMixin, CreateOrgMixin):
+class TestTopologySockets(CreateGraphObjectsMixin, TestOrganizationMixin):
     node_model = Node
     link_model = Link
     topology_model = Topology
@@ -34,20 +41,86 @@ async def _get_communicator(self, admin_client, topology_id):
         )
         return communicator
 
-    async def test_consumer_connection(self, admin_user, admin_client):
+    async def _assert_connection_superuser(self, admin_client, conn=True):
         org = await database_sync_to_async(self._create_org)()
         t = await database_sync_to_async(self._create_topology)(organization=org)
         communicator = await self._get_communicator(admin_client, t.pk)
         connected, _ = await communicator.connect()
-        assert connected is True
+        assert connected is conn
+        await communicator.disconnect()
+
+    async def _assert_connection_org_manager(self, client, view_perm=True, conn=True):
+        org = await database_sync_to_async(self._create_org)()
+        test_user = await database_sync_to_async(self._create_user)(
+            username='test-user-org-manager', email='test@orgmanger.com', is_staff=True
+        )
+        await database_sync_to_async(self._create_org_user)(
+            is_admin=True, user=test_user, organization=org
+        )
+        t = await database_sync_to_async(self._create_topology)(organization=org)
+        if view_perm:
+            topology_view_permission = await database_sync_to_async(
+                Permission.objects.get
+            )(codename='view_topology')
+            await database_sync_to_async(test_user.user_permissions.add)(
+                topology_view_permission
+            )
+        await database_sync_to_async(client.force_login)(test_user)
+        communicator = await self._get_communicator(client, t.pk)
+        connected, _ = await communicator.connect()
+        assert connected is conn
         await communicator.disconnect()
 
-    async def test_unauthenticated_users(self, client):
-        client.cookies['sessionid'] = 'random'
+    async def _assert_connection_unauth_user(self, client, conn=True):
+        client.cookies['sessionid'] = 'unauthenticated_user'
         org = await database_sync_to_async(self._create_org)()
         t = await database_sync_to_async(self._create_topology)(organization=org)
         communicator = await self._get_communicator(client, t.pk)
         connected, _ = await communicator.connect()
+        assert connected is conn
+        await communicator.disconnect()
+
+    async def test_consumer_connection_superuser(self, admin_client):
+        await self._assert_connection_superuser(admin_client)
+
+    async def test_consumer_connection_org_manager_view_perm(self, client):
+        await self._assert_connection_org_manager(client)
+
+    async def test_consumer_connection_org_manager_without_view_perm(self, client):
+        await self._assert_connection_org_manager(client, view_perm=False, conn=False)
+
+    async def test_consumer_connection_unauth_user(self, client):
+        await self._assert_connection_unauth_user(client, conn=False)
+
+    @patch.object(app_settings, 'TOPOLOGY_API_AUTH_REQUIRED', False)
+    async def test_consumer_connection_auth_disabled_superuser(self, admin_client):
+        await self._assert_connection_superuser(admin_client)
+
+    @patch.object(app_settings, 'TOPOLOGY_API_AUTH_REQUIRED', False)
+    async def test_consumer_connection_auth_disabled_org_manager(self, client):
+        await self._assert_connection_org_manager(client)
+
+    @patch.object(app_settings, 'TOPOLOGY_API_AUTH_REQUIRED', False)
+    async def test_consumer_connection_auth_disabled_unauth_user(self, client):
+        await self._assert_connection_unauth_user(client)
+
+    async def test_consumer_connection_invalid_topology_pk(
+        self, admin_user, admin_client
+    ):
+        org = await database_sync_to_async(self._create_org)()
+        await database_sync_to_async(self._create_topology)(organization=org)
+        communicator = await self._get_communicator(admin_client, 'invalid_topology_pk')
+        connected, _ = await communicator.connect()
+        assert connected is False
+        await communicator.disconnect()
+
+    async def test_consumer_connection_unexisting_topology(
+        self, admin_user, admin_client
+    ):
+        org = await database_sync_to_async(self._create_org)()
+        await database_sync_to_async(self._create_topology)(organization=org)
+        communicator = await self._get_communicator(admin_client, uuid4())
+        connected, _ = await communicator.connect()
         assert connected is False
         await communicator.disconnect()
 
diff --git a/requirements-test.txt b/requirements-test.txt
index e23ec9bc..529af3fe 100644
--- a/requirements-test.txt
+++ b/requirements-test.txt
@@ -3,6 +3,7 @@ pytest-asyncio~=0.14.0
 pytest-django~=4.5.2
 freezegun~=1.1.0
 selenium~=3.141.0
+urllib3~=1.26.1
 # Needed to test integrations
 openwisp-monitoring @ https://github.com/openwisp/openwisp-monitoring/tarball/master
 openwisp-controller @ https://github.com/openwisp/openwisp-controller/tarball/master