[Fix] 500 error by enforcing valid UUID patterns in admin URLs #682

openwisp_controller/config/admin.py

@@ -136,6 +136,11 @@ def get_extra_context(self, pk=None):
         if not issubclass(self.model, AbstractVpn):
             ctx['CONFIG_BACKEND_FIELD_SHOWN'] = app_settings.CONFIG_BACKEND_FIELD_SHOWN
         if pk:
+            UUID_PATTERN = re.compile(
+                '^[a-fA-F0-9]{8}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{12}$'
+            )
+            if not UUID_PATTERN.match(str(pk)):
+                raise Http404()
             ctx['download_url'] = reverse('{0}_download'.format(prefix), args=[pk])
             try:
                 has_config = True
@@ -165,9 +170,10 @@ def change_view(self, request, object_id, form_url='', extra_context=None):
     def get_urls(self):
         options = getattr(self.model, '_meta')
         url_prefix = '{0}_{1}'.format(options.app_label, options.model_name)
+        UUID_PATTERN = r'([a-fA-F0-9]{8}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{12})'
         return [
             re_path(
-                r'^download/(?P<pk>[^/]+)/$',
+                r'^download/{0}/$'.format(UUID_PATTERN),
                 self.admin_site.admin_view(self.download_view),
                 name='{0}_download'.format(url_prefix),
             ),
@@ -177,7 +183,7 @@ def get_urls(self):
                 name='{0}_preview'.format(url_prefix),
             ),
             re_path(
-                r'^(?P<pk>[^/]+)/context\.json$',
+                r'^{0}/context\.json$'.format(UUID_PATTERN),
                 self.admin_site.admin_view(self.context_view),
                 name='{0}_context'.format(url_prefix),
             ),

openwisp_controller/config/tests/pytest.py

@@ -11,21 +11,22 @@
 from ..base.channels_consumer import BaseDeviceConsumer
 from .utils import CreateDeviceMixin
 
-Device = load_model('config', 'Device')
+Device = load_model("config", "Device")
 
 
 @pytest.mark.asyncio
 @pytest.mark.django_db(transaction=True)
 class TestDeviceConsumer(CreateDeviceMixin):
     model = Device
+    UUID_PATTERN = "[a-fA-F0-9]{8}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{12}"
     application = ProtocolTypeRouter(
         {
-            'websocket': AllowedHostsOriginValidator(
+            "websocket": AllowedHostsOriginValidator(
                 AuthMiddlewareStack(
                     URLRouter(
                         [
                             re_path(
-                                r'^ws/controller/device/(?P<pk>[^/]+)/$',
+                                f"^ws/controller/device/(?P<pk>{UUID_PATTERN})/$",
                                 BaseDeviceConsumer.as_asgi(),
                             )
                         ]
@@ -36,34 +37,34 @@ class TestDeviceConsumer(CreateDeviceMixin):
     )
 
     async def _get_communicator(self, admin_client, device_id):
-        session_id = admin_client.cookies['sessionid'].value
-        communicator = WebsocketCommunicator(
+       session_id = admin_client.cookies["sessionid"].value
+       communicator = WebsocketCommunicator(
             self.application,
-            path=f'ws/controller/device/{device_id}/',
+            path=f"ws/controller/device/{device_id}/",
             headers=[
                 (
-                    b'cookie',
-                    f'sessionid={session_id}'.encode('ascii'),
+                    b"cookie",
+                    f"sessionid={session_id}".encode("ascii"),
                 )
             ],
         )
-        connected, subprotocol = await communicator.connect()
-        assert connected is True
-        return communicator
+       connected, subprotocol = await communicator.connect()
+       assert connected is True
+       return communicator
 
     @database_sync_to_async
     def _add_model_permissions(self, user, add=True, change=True, delete=True):
         permissions = []
         if add:
-            permissions.append(f'add_{self.model._meta.model_name}')
+            permissions.append(f"add_{self.model._meta.model_name}")
         if change:
-            permissions.append(f'change_{self.model._meta.model_name}')
+            permissions.append(f"change_{self.model._meta.model_name}")
         if delete:
-            permissions.append(f'delete_{self.model._meta.model_name}')
+            permissions.append(f"delete_{self.model._meta.model_name}")
         user.user_permissions.set(Permission.objects.filter(codename__in=permissions))
 
     async def test_unauthenticated_user(self, client):
-        client.cookies['sessionid'] = 'random'
+        client.cookies["sessionid"] = "random"
         device = await database_sync_to_async(self._create_device)()
         with pytest.raises(AssertionError):
             await self._get_communicator(client, device.id)
@@ -91,14 +92,14 @@ async def test_user_authorization(self, client, django_user_model):
 
     async def test_silent_disconnection(self, admin_user, admin_client):
         device = await database_sync_to_async(self._create_device)()
-        session_id = admin_client.cookies['sessionid'].value
+        session_id = admin_client.cookies["sessionid"].value
         communicator = WebsocketCommunicator(
             self.application,
-            path=f'ws/controller/device/{device.pk}/',
+            path=f"ws/controller/device/{device.pk}/",
             headers=[
                 (
-                    b'cookie',
-                    f'sessionid={session_id}'.encode('ascii'),
+                    b"cookie",
+                    f"sessionid={session_id}".encode("ascii"),
                 )
             ],
         )

openwisp_controller/config/tests/utils.py

@@ -408,4 +408,4 @@ def _create_device_group(self, **kwargs):
         device_group = DeviceGroup(**options)
         device_group.full_clean()
         device_group.save()
-        return device_group
+        return device_group

openwisp_controller/config/utils.py

@@ -1,4 +1,5 @@
 import logging
+import re
 
 from django.core.exceptions import ValidationError
 from django.db.models import Q
@@ -9,6 +10,9 @@
 
 logger = logging.getLogger(__name__)
 
+# Define UUID_PATTERN for URL regular expressions
+UUID_PATTERN = r'[0-9a-fA-F]{8}\-[0-9a-fA-F]{4}\-[0-9a-fA-F]{4}\-[0-9a-fA-F]{4}\-[0-9a-fA-F]{12}'
+
 
 def get_object_or_404(model, **kwargs):
     """
@@ -116,22 +120,22 @@ def get_controller_urls(views_module):
     """
     urls = [
         re_path(
-            'controller/device/checksum/(?P<pk>[^/]+)/$',
+            f'controller/device/checksum/(?P<pk>{UUID_PATTERN})/$',
             views_module.device_checksum,
             name='device_checksum',
         ),
         re_path(
-            'controller/device/download-config/(?P<pk>[^/]+)/$',
+            f'controller/device/download-config/(?P<pk>{UUID_PATTERN})/$',
             views_module.device_download_config,
             name='device_download_config',
         ),
         re_path(
-            'controller/device/update-info/(?P<pk>[^/]+)/$',
+            f'controller/device/update-info/(?P<pk>{UUID_PATTERN})/$',
             views_module.device_update_info,
             name='device_update_info',
         ),
         re_path(
-            'controller/device/report-status/(?P<pk>[^/]+)/$',
+            f'controller/device/report-status/(?P<pk>{UUID_PATTERN})/$',
             views_module.device_report_status,
             name='device_report_status',
         ),
@@ -141,33 +145,33 @@ def get_controller_urls(views_module):
             name='device_register',
         ),
         re_path(
-            'controller/vpn/checksum/(?P<pk>[^/]+)/$',
+            f'controller/vpn/checksum/(?P<pk>{UUID_PATTERN})/$',
             views_module.vpn_checksum,
             name='vpn_checksum',
         ),
         re_path(
-            'controller/vpn/download-config/(?P<pk>[^/]+)/$',
+            f'controller/vpn/download-config/(?P<pk>{UUID_PATTERN})/$',
             views_module.vpn_download_config,
             name='vpn_download_config',
         ),
         # legacy URLs
         re_path(
-            'controller/checksum/(?P<pk>[^/]+)/$',
+            f'controller/checksum/(?P<pk>{UUID_PATTERN})/$',
             views_module.device_checksum,
             name='checksum_legacy',
         ),
         re_path(
-            'controller/download-config/(?P<pk>[^/]+)/$',
+            f'controller/download-config/(?P<pk>{UUID_PATTERN})/$',
             views_module.device_download_config,
             name='download_config_legacy',
         ),
         re_path(
-            'controller/update-info/(?P<pk>[^/]+)/$',
+            f'controller/update-info/(?P<pk>{UUID_PATTERN})/$',
             views_module.device_update_info,
             name='update_info_legacy',
         ),
         re_path(
-            'controller/report-status/(?P<pk>[^/]+)/$',
+            f'controller/report-status/(?P<pk>{UUID_PATTERN})/$',
             views_module.device_report_status,
             name='report_status_legacy',
         ),

openwisp_controller/connection/channels/routing.py

@@ -4,9 +4,10 @@
 
 
 def get_routes(consumer=ow_consumer):
+    UUID_PATTERN = '[a-fA-F0-9]{8}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{4}-?[a-fA-F0-9]{12}'
     return [
         re_path(
-            r'^ws/controller/device/(?P<pk>[^/]+)/command$',
+            f'^ws/controller/device/(?P<pk>{UUID_PATTERN})/command$',
             consumer.CommandConsumer.as_asgi(),
         )
     ]