Merge pull request #5312 from bjester/code-ql-fixes

Address user-supplied CodeQL issues

Author: bjester

contentcuration/contentcuration/urls.py

@@ -13,6 +13,8 @@
     1. Add an import:  from blog import urls as blog_urls
     2. Add a URL to urlpatterns:  re_path(r'^blog/', include(blog_urls))
 """
+import uuid
+
 import django_js_reverse.views as django_js_reverse_views
 from django.conf import settings
 from django.conf.urls.i18n import i18n_patterns
@@ -53,8 +55,11 @@
 
 class StagingPageRedirectView(RedirectView):
     def get_redirect_url(self, *args, **kwargs):
-        channel_id = kwargs["channel_id"]
-        return "/channels/{}/#/staging".format(channel_id)
+        try:
+            channel_id = uuid.UUID(kwargs["channel_id"]).hex
+            return "/channels/{}/#/staging".format(channel_id)
+        except ValueError:
+            return None
 
 
 router = routers.DefaultRouter(trailing_slash=False)

contentcuration/contentcuration/utils/files.py

@@ -108,7 +108,13 @@ def get_thumbnail_encoding(filename, dimension=THUMBNAIL_WIDTH):
             inbuffer = default_storage.open(filename, "rb")
 
         else:
-            inbuffer = open(filename, "rb")
+            # Normalize the path and ensure it is indeed within STATIC_ROOT
+            normalized_path = os.path.normpath(filename)
+            static_root = os.path.abspath(settings.STATIC_ROOT)
+            abs_path = os.path.abspath(normalized_path)
+            if not abs_path.startswith(static_root + os.sep):
+                raise ValueError("Attempted access to file outside of STATIC_ROOT")
+            inbuffer = open(abs_path, "rb")
 
         if not inbuffer:
             raise AssertionError

contentcuration/contentcuration/views/internal.py

@@ -221,9 +221,9 @@ def api_create_channel_endpoint(request):
                 "channel_id": obj.pk,
             }
         )
-    except KeyError:
+    except KeyError as e:
         return HttpResponseBadRequest(
-            "Required attribute missing from data: {}".format(data)
+            "Required attribute missing from data | {}".format(str(e))
         )
     except Exception as e:
         handle_server_error(e, request)
@@ -294,9 +294,9 @@ def api_commit_channel(request):
         )
     except (Channel.DoesNotExist, PermissionDenied):
         return HttpResponseNotFound("No channel matching: {}".format(channel_id))
-    except KeyError:
+    except KeyError as e:
         return HttpResponseBadRequest(
-            "Required attribute missing from data: {}".format(data)
+            "Required attribute missing from data | {}".format(str(e))
         )
     except Exception as e:
         handle_server_error(e, request)
@@ -351,9 +351,9 @@ def api_add_nodes_to_tree(request):
         return HttpResponseNotFound("No content matching: {}".format(parent_id))
     except ValidationError as e:
         return HttpResponseBadRequest(content=str(e))
-    except KeyError:
+    except KeyError as e:
         return HttpResponseBadRequest(
-            "Required attribute missing from data: {}".format(data)
+            "Required attribute missing from data | {}".format(str(e))
         )
     except NodeValidationError as e:
         return HttpResponseBadRequest(str(e))

contentcuration/contentcuration/views/zip.py

@@ -117,9 +117,7 @@ def get(self, request, zipped_filename, embedded_filepath):  # noqa: C901
         Handles GET requests and serves a static file from within the zip file.
         """
         if not VALID_STORAGE_FILENAME.match(zipped_filename):
-            return HttpResponseNotFound(
-                "'{}' is not a valid URL for this zip file".format(zipped_filename)
-            )
+            return HttpResponseNotFound("Invalid URL for this zip file")
 
         storage = default_storage
 
@@ -132,9 +130,7 @@ def get(self, request, zipped_filename, embedded_filepath):  # noqa: C901
 
         # if the zipfile does not exist on disk, return a 404
         if not storage.exists(zipped_path):
-            return HttpResponseNotFound(
-                '"%(filename)s" does not exist in storage' % {"filename": zipped_path}
-            )
+            return HttpResponseNotFound("Zipfile does not exist in storage")
 
         # if client has a cached version, use that (we can safely assume nothing has changed, due to MD5)
         if request.META.get("HTTP_IF_MODIFIED_SINCE"):
@@ -153,9 +149,7 @@ def get(self, request, zipped_filename, embedded_filepath):  # noqa: C901
                     info = zf.getinfo(embedded_filepath)
                 except KeyError:
                     return HttpResponseNotFound(
-                        '"{}" does not exist inside "{}"'.format(
-                            embedded_filepath, zipped_filename
-                        )
+                        "Embedded file does not exist inside zip"
                     )
 
                 # try to guess the MIME type of the embedded file being referenced

contentcuration/kolibri_public/views_v1.py

@@ -6,6 +6,7 @@
 from django.db.models import TextField
 from django.db.models import Value
 from django.http import HttpResponseNotFound
+from django.utils.html import escape
 from django.utils.translation import gettext_lazy as _
 from django.views.decorators.cache import cache_page
 from kolibri_content.constants.schema_versions import MIN_CONTENT_SCHEMA_VERSION
@@ -75,9 +76,7 @@ def get_public_channel_list(request, version):
     try:
         channel_list = _get_channel_list(version, request.query_params)
     except LookupError:
-        return HttpResponseNotFound(
-            _("Api endpoint {} is not available").format(version)
-        )
+        return HttpResponseNotFound(_("API version is unavailable"))
     return Response(PublicChannelSerializer(channel_list, many=True).data)
 
 
@@ -92,12 +91,10 @@ def get_public_channel_lookup(request, version, identifier):
             identifier=identifier.strip().replace("-", ""),
         )
     except LookupError:
-        return HttpResponseNotFound(
-            _("Api endpoint {} is not available").format(version)
-        )
+        return HttpResponseNotFound(_("API version is unavailable"))
     if not channel_list.exists():
         return HttpResponseNotFound(
-            _("No channel matching {} found").format(identifier)
+            _("No channel matching {} found").format(escape(identifier))
         )
     return Response(PublicChannelSerializer(channel_list, many=True).data)
 
@@ -108,7 +105,9 @@ def get_channel_name_by_id(request, channel_id):
     """ Endpoint: /public/channels/<channel_id> """
     channel = Channel.objects.filter(pk=channel_id).first()
     if not channel:
-        return HttpResponseNotFound("Channel with id {} not found".format(channel_id))
+        return HttpResponseNotFound(
+            "Channel with id {} not found".format(escape(channel_id))
+        )
     channel_info = {
         "name": channel.name,
         "description": channel.description,