[fix] Fixed Geo API endpoints to allow bearer authentication

- Added tests for Pki API endpoints for bearer authentication
- Added tests for Connection API endpoints for bearer authentication
- Added tests for Config API endpoints for bearer authentication

openwisp_controller/config/api/views.py

@@ -4,19 +4,14 @@
 from django.http import Http404
 from django.urls.base import reverse
 from rest_framework import pagination
-from rest_framework.authentication import SessionAuthentication
 from rest_framework.generics import (
     ListCreateAPIView,
     RetrieveAPIView,
     RetrieveUpdateDestroyAPIView,
 )
-from rest_framework.permissions import IsAuthenticated
 from swapper import load_model
 
-from openwisp_users.api.authentication import BearerAuthentication
-from openwisp_users.api.mixins import FilterByOrganizationManaged
-from openwisp_users.api.permissions import DjangoModelPermissions
-
+from ...mixins import ProtectedAPIMixin
 from ..admin import BaseConfigAdmin
 from .serializers import (
     DeviceDetailSerializer,
@@ -42,14 +37,6 @@ class ListViewPagination(pagination.PageNumberPagination):
     max_page_size = 100
 
 
-class ProtectedAPIMixin(FilterByOrganizationManaged):
-    authentication_classes = [BearerAuthentication, SessionAuthentication]
-    permission_classes = [
-        IsAuthenticated,
-        DjangoModelPermissions,
-    ]
-
-
 class TemplateListCreateView(ProtectedAPIMixin, ListCreateAPIView):
     serializer_class = TemplateSerializer
     queryset = Template.objects.order_by('-created')

openwisp_controller/config/tests/test_api.py

@@ -6,7 +6,9 @@
 from swapper import load_model
 
 from openwisp_controller.tests.utils import TestAdminMixin
+from openwisp_users.tests.test_api import AuthenticationMixin
 from openwisp_users.tests.utils import TestOrganizationMixin
+from openwisp_utils.tests import capture_any_output
 
 from .utils import CreateConfigTemplateMixin, CreateDeviceGroupMixin, TestVpnX509Mixin
 
@@ -24,6 +26,7 @@ class TestConfigApi(
     CreateConfigTemplateMixin,
     TestVpnX509Mixin,
     CreateDeviceGroupMixin,
+    AuthenticationMixin,
     TestCase,
 ):
     def setUp(self):
@@ -700,6 +703,104 @@ def _assert_response(org_slug, common_name):
             vpnclient = config.vpnclient_set.select_related('cert').first()
             _assert_response(org_slug=org.slug, common_name=vpnclient.cert.common_name)
 
+    @capture_any_output()
+    def test_bearer_authentication(self):
+        self.client.logout()
+        token = self._obtain_auth_token(username='admin', password='tester')
+        vpn = self._create_vpn()
+        template = self._create_template(type='vpn', vpn=vpn, default=True)
+        device_group = self._create_device_group()
+        device = self._create_device(group=device_group)
+        config = self._create_config(device=device)
+        vpnclient_cert = config.vpnclient_set.first().cert
+
+        with self.subTest('Test TemplateListCreateView'):
+            response = self.client.get(
+                reverse('config_api:template_list'),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test TemplateDetailView'):
+            response = self.client.get(
+                reverse('config_api:template_detail', args=[template.id]),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test DownloadTemplateconfiguration'):
+            response = self.client.get(
+                reverse('config_api:download_template_config', args=[template.id]),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test VpnListCreateView'):
+            response = self.client.get(
+                reverse('config_api:vpn_list'),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test VpnDetailView'):
+            response = self.client.get(
+                reverse('config_api:vpn_detail', args=[vpn.id]),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test DownloadVpnView'):
+            response = self.client.get(
+                reverse('config_api:download_vpn_config', args=[vpn.id]),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test DeviceListCreateView'):
+            response = self.client.get(
+                reverse('config_api:device_list'),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test DeviceDetailView'):
+            response = self.client.get(
+                reverse('config_api:device_detail', args=[device.id]),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test DeviceGroupListCreateView'):
+            response = self.client.get(
+                reverse('config_api:devicegroup_list'),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test DeviceGroupDetailView'):
+            response = self.client.get(
+                reverse('config_api:devicegroup_detail', args=[device_group.id]),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test DeviceGroupCommonName'):
+            response = self.client.get(
+                reverse(
+                    'config_api:devicegroup_x509_commonname',
+                    args=[vpnclient_cert.common_name],
+                ),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test DownloadDeviceView'):
+            response = self.client.get(
+                reverse('config_api:download_device_config', args=[device.id]),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
 
 class TestConfigApiTransaction(
     TestAdminMixin,

openwisp_controller/connection/api/views.py

@@ -9,13 +9,11 @@
     RetrieveUpdateDestroyAPIView,
     get_object_or_404,
 )
-from rest_framework.permissions import IsAuthenticated
 from swapper import load_model
 
 from openwisp_users.api.authentication import BearerAuthentication
-from openwisp_users.api.mixins import FilterByOrganizationManaged
-from openwisp_users.api.permissions import DjangoModelPermissions
 
+from ...mixins import ProtectedAPIMixin
 from .serializer import (
     CommandSerializer,
     CredentialSerializer,
@@ -84,14 +82,6 @@ def get_object(self):
         return obj
 
 
-class ProtectedAPIMixin(FilterByOrganizationManaged):
-    authentication_classes = [BearerAuthentication, SessionAuthentication]
-    permission_classes = [
-        IsAuthenticated,
-        DjangoModelPermissions,
-    ]
-
-
 class CredentialListCreateView(ProtectedAPIMixin, ListCreateAPIView):
     queryset = Credentials.objects.order_by('-created')
     serializer_class = CredentialSerializer

openwisp_controller/connection/tests/test_api.py

@@ -286,7 +286,9 @@ def test_non_superuser(self):
             self.assertEqual(response.data['count'], 1)
 
 
-class TestConnectionApi(TestAdminMixin, TestCase, CreateConnectionsMixin):
+class TestConnectionApi(
+    TestAdminMixin, AuthenticationMixin, TestCase, CreateConnectionsMixin
+):
     def setUp(self):
         super().setUp()
         self._login()
@@ -469,3 +471,41 @@ def test_delete_deviceconnection_detail(self):
         with self.assertNumQueries(9):
             response = self.client.delete(path)
         self.assertEqual(response.status_code, 204)
+
+    def test_bearer_authentication(self):
+        self.client.logout()
+        token = self._obtain_auth_token(username='admin', password='tester')
+        credentials = self._create_credentials(auto_add=True)
+        device_conn = self._create_device_connection(credentials=credentials)
+        device = device_conn.device
+
+        with self.subTest('Test CredentialListCreateView'):
+            response = self.client.get(
+                reverse('connection_api:credential_list'),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test CredentialDetailView'):
+            response = self.client.get(
+                reverse('connection_api:credential_detail', args=[credentials.id]),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test DeviceConnenctionListCreateView'):
+            response = self.client.get(
+                reverse('connection_api:deviceconnection_list', args=[device.id]),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test DeviceConnectionDetailView'):
+            response = self.client.get(
+                reverse(
+                    'connection_api:deviceconnection_detail',
+                    args=[device.id, device_conn.id],
+                ),
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)

openwisp_controller/geo/api/views.py

@@ -6,8 +6,9 @@
 from rest_framework_gis.pagination import GeoJsonPagination
 from swapper import load_model
 
-from openwisp_users.api.mixins import FilterByOrganizationManaged, FilterByParentManaged
+from openwisp_users.api.mixins import FilterByParentManaged
 
+from ...mixins import ProtectedAPIMixin
 from .serializers import (
     GeoJsonLocationSerializer,
     LocationDeviceSerializer,
@@ -34,13 +35,18 @@ class ListViewPagination(pagination.PageNumberPagination):
     max_page_size = 100
 
 
-class DeviceLocationView(generics.RetrieveUpdateAPIView):
+class DeviceLocationView(ProtectedAPIMixin, generics.RetrieveUpdateAPIView):
     serializer_class = LocationSerializer
     permission_classes = (DevicePermission,)
     queryset = Device.objects.select_related(
         'devicelocation', 'devicelocation__location'
     )
 
+    def get_queryset(self):
+        # It is required to override ProtectedAPIMixin.get_queryset
+        # which filters the queryset for organizations managed.
+        return self.queryset
+
     def get_location(self, device):
         try:
             return device.devicelocation.location
@@ -82,7 +88,7 @@ class Meta:
         fields = ['organization_slug']
 
 
-class GeoJsonLocationList(FilterByOrganizationManaged, generics.ListAPIView):
+class GeoJsonLocationList(ProtectedAPIMixin, generics.ListAPIView):
     queryset = Location.objects.filter(devicelocation__isnull=False).annotate(
         device_count=Count('devicelocation')
     )
@@ -92,7 +98,9 @@ class GeoJsonLocationList(FilterByOrganizationManaged, generics.ListAPIView):
     filterset_class = GeoJsonLocationFilter
 
 
-class LocationDeviceList(FilterByParentManaged, generics.ListAPIView):
+class LocationDeviceList(
+    FilterByParentManaged, ProtectedAPIMixin, generics.ListAPIView
+):
     serializer_class = LocationDeviceSerializer
     pagination_class = ListViewPagination
     queryset = Device.objects.none()

openwisp_controller/geo/tests/test_api.py

@@ -1,8 +1,10 @@
 import json
 
+from django.contrib.auth import get_user_model
 from django.contrib.gis.geos import Point
 from django.test import TestCase
 from django.urls import reverse
+from rest_framework.authtoken.models import Token
 from swapper import load_model
 
 from openwisp_controller.config.tests.utils import CreateConfigTemplateMixin
@@ -15,6 +17,8 @@
 Location = load_model('geo', 'Location')
 DeviceLocation = load_model('geo', 'DeviceLocation')
 OrganizationUser = load_model('openwisp_users', 'OrganizationUser')
+Group = load_model('openwisp_users', 'Group')
+User = get_user_model()
 
 
 class TestApi(TestGeoMixin, TestCase):
@@ -29,6 +33,11 @@ def test_permission_404(self):
         self.assertEqual(r.status_code, 404)
 
     def test_permission_403(self):
+        user = User.objects.create(
+            username='tester',
+            password='tester',
+        )
+        self.client.force_login(user)
         dl = self._create_object_location()
         url = reverse(self.url_name, args=[dl.device.pk])
         r = self.client.get(url)
@@ -98,6 +107,40 @@ def test_put_update_coordinates(self):
         )
         self.assertEqual(self.location_model.objects.count(), 1)
 
+    @capture_any_output()
+    def test_bearer_authentication(self):
+        user = User.objects.create(
+            username='admin', password='password', is_staff=True, is_superuser=True
+        )
+        token = Token.objects.create(user=user).key
+        device = self._create_object()
+
+        with self.subTest('Test DeviceLocationView'):
+            response = self.client.get(
+                reverse(self.url_name, args=[device.pk]),
+                data={'key': device.key},
+                content_type='application/json',
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test GeoJsonLocationListView'):
+            response = self.client.get(
+                reverse('geo_api:location_geojson'),
+                content_type='application/json',
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest('Test LocationDeviceList'):
+            location = self._create_location(organization=device.organization)
+            response = self.client.get(
+                reverse('geo_api:location_device_list', args=[location.id]),
+                content_type='application/json',
+                HTTP_AUTHORIZATION=f'Bearer {token}',
+            )
+            self.assertEqual(response.status_code, 200)
+
 
 class TestMultitenantApi(
     TestOrganizationMixin, TestGeoMixin, TestCase, CreateConfigTemplateMixin
@@ -111,10 +154,11 @@ def setUp(self):
         # create 2 orgs
         self._create_org(name='org_b', slug='org_b')
         org_a = self._create_org(name='org_a', slug='org_a')
+        user = self._create_operator()
+        admin_group = Group.objects.get(name='Administrator')
+        admin_group.user_set.add(user)
         # create an operator for org_a
-        ou = OrganizationUser.objects.create(
-            user=self._create_operator(), organization=org_a
-        )
+        ou = OrganizationUser.objects.create(user=user, organization=org_a)
         ou.is_admin = True
         ou.save()
         # create a superuser
@@ -156,7 +200,7 @@ def test_location_device_list(self):
         with self.subTest('Test location device list for unauthenticated user'):
             self.client.logout()
             r = self.client.get(reverse(url, args=[location_a.id]))
-            self.assertEqual(r.status_code, 403)
+            self.assertEqual(r.status_code, 401)
 
     @capture_any_output()
     def test_geojson_list(self):
@@ -197,4 +241,4 @@ def test_geojson_list(self):
         with self.subTest('Test geojson list unauthenticated user'):
             self.client.logout()
             r = self.client.get(reverse(url))
-            self.assertEqual(r.status_code, 403)
+            self.assertEqual(r.status_code, 401)