[connection] Enforce multitenancy in DeviceConnection.credentials

openwisp · Jun 15, 2018 · 5420a2d · 5420a2d
1 parent c36cb8d
commit 5420a2d
Show file tree

Hide file tree

Showing 4 changed files with 37 additions and 2 deletions.
diff --git a/openwisp_controller/connection/admin.py b/openwisp_controller/connection/admin.py
@@ -25,11 +25,19 @@ def get_queryset(self, request):
         return qs.order_by('priority')
 
 
-class DeviceConnectionInline(admin.StackedInline):
+class DeviceConnectionInline(MultitenantAdminMixin, admin.StackedInline):
     model = DeviceConnection
     exclude = ['params', 'created', 'modified']
     readonly_fields = ['is_working', 'failure_reason', 'last_attempt']
     extra = 0
 
+    multitenant_shared_relations = ('credentials',)
+
+    def get_queryset(self, request):
+        """
+        Override MultitenantAdminMixin.get_queryset() because it breaks
+        """
+        return super(admin.StackedInline, self).get_queryset(request)
+
 
 DeviceAdmin.inlines += [DeviceConnectionInline, DeviceIpInline]
diff --git a/openwisp_controller/connection/models.py b/openwisp_controller/connection/models.py
@@ -102,6 +102,12 @@ class Meta:
         verbose_name_plural = _('Device connections')
 
     def clean(self):
+        cred_org = self.credentials.organization
+        if cred_org and cred_org != self.device.organization:
+            raise ValidationError({
+                'credentials': _('The organization of these credentials doesn\'t '
+                                 'match the organization of the device')
+            })
         if not self.update_strategy and hasattr(self.device, 'config'):
             try:
                 self.update_strategy = app_settings.CONFIG_UPDATE_MAPPING[self.device.config.backend]

diff --git a/openwisp_controller/connection/tests/base.py b/openwisp_controller/connection/tests/base.py
@@ -41,7 +41,10 @@ def _create_device_connection(self, **kwargs):
                     params={})
         opts.update(kwargs)
         if 'credentials' not in opts:
-            opts['credentials'] = self._create_credentials()
+            cred_opts = {}
+            if 'device' in opts:
+                cred_opts = {'organization': opts['device'].organization}
+            opts['credentials'] = self._create_credentials(**cred_opts)
         org = opts['credentials'].organization
         if 'device' not in opts:
             opts['device'] = self._create_device(organization=org)

diff --git a/openwisp_controller/connection/tests/test_models.py b/openwisp_controller/connection/tests/test_models.py
@@ -163,3 +163,21 @@ def test_address_list_link_local_ip(self):
         interfaces = get_interfaces()
         self.assertEqual(len(address_list), len(interfaces))
         self.assertIn(ipv6_linklocal, address_list[0])
+
+    def test_device_connection_credential_org_validation(self):
+        dc = self._create_device_connection()
+        shared = self._create_credentials(name='cred-shared',
+                                          organization=None)
+        dc.credentials = shared
+        dc.full_clean()
+        # ensure credentials of other orgs aren't accepted
+        org2 = self._create_org(name='org2')
+        cred2 = self._create_credentials(name='cred2',
+                                         organization=org2)
+        try:
+            dc.credentials = cred2
+            dc.full_clean()
+        except ValidationError as e:
+            self.assertIn('credentials', e.message_dict)
+        else:
+            self.fail('ValidationError not raised')