Improve asset file retrieval

Author: etj

geonode/assets/local.py

@@ -256,7 +256,7 @@ def create_response(
             return HttpResponse("Asset does not contain any data", status=500)
 
         if len(asset.location) > 1:
-            logger.warning("TODO: Asset contains more than one file. Download needs to be implemented")
+            logger.warning("TODO: Asset contains more than one file. Only first file will be returned")
 
         file0 = asset.location[0]
         if not path:  # use the file definition
@@ -266,10 +266,6 @@ def create_response(
             localfile = file0
 
         else:  # a specific file is requested
-            if "/../" in path:  # we may want to improve fraudolent request detection
-                logger.warning(f"Tentative path traversal for asset {asset.id}")
-                return HttpResponse(f"File not found for asset {asset.id}", status=400)
-
             if os.path.isfile(file0):
                 dir0 = os.path.dirname(file0)
             elif os.path.isdir(file0):
@@ -280,6 +276,13 @@ def create_response(
             localfile = os.path.join(dir0, path)
             logger.debug(f"Requested path {dir0} + {path}")
 
+            # check the requested file is within the asset's directory
+            localfile = os.path.realpath(localfile)
+            realassetpath = os.path.realpath(dir0)
+            if not localfile.startswith(realassetpath):
+                logger.error(f"Tentative path traversal for asset {asset.id} on path [{path}]")
+                return HttpResponse(f"File not found for asset {asset.id}", status=400)
+
         if os.path.isfile(localfile):
             filename = os.path.basename(localfile)
             orig_base, ext = os.path.splitext(filename)

geonode/assets/tests.py

@@ -415,6 +415,49 @@ def test_download_with_attachment_for_anonymous(self):
                 if resource:
                     resource.delete()
 
+    def test_cross_download(self):
+
+        admin, _ = get_user_model().objects.get_or_create(username="admin")
+        user1, _ = get_user_model().objects.get_or_create(username="user1")
+        user2, _ = get_user_model().objects.get_or_create(username="user2")
+
+        asset_handler = asset_handler_registry.get_default_handler()
+
+        assets = []
+        for user, file in ((user1, ONE_JSON), (user2, TWO_JSON)):
+            asset1 = asset_handler.create(
+                title="Test Asset1",
+                description="Description of test asset",
+                type="NeverMind",
+                owner=user1,
+                files=[file],
+                clone_files=True,
+            )
+            asset1.save()
+            self.assertIsInstance(asset1, LocalAsset)
+            assets.append(asset1)
+
+        # check that user1 can access file1
+        self.client.force_login(user1)
+
+        args = {"pk": assets[0].pk, "path": "one.json"}
+        url = reverse("assets-link", kwargs=args)
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+
+        # check that user1 can NOT access file2 using asset1 link/download
+        a2path = assets[1].location[0]
+        logger.debug(f"Asset path {a2path}")
+        a2dir = a2path.split("/")[-2]
+        logger.debug(f"Asset dir  {a2dir}")
+        forged_path = f"../{a2dir}/two.json"  # path will be automatically urlencoded into "%2e%2e%2f{a2dir}%2ftwo.json"
+
+        args = {"pk": assets[0].pk, "path": forged_path}
+        url = reverse("assets-link", kwargs=args)
+        logger.debug(f"Reverse link URL is {url}")
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 400)
+
     def _setup_test(self, u, _file=ONE_JSON):
         asset_handler = asset_handler_registry.get_default_handler()
         asset = asset_handler.create(

geonode/assets/views.py

@@ -63,7 +63,7 @@ def list(self, request, *args, **kwargs):
         queryset = self.filter_queryset(self.get_queryset())
 
         user = request.user
-        is_admin = user.is_superuser if user and user.is_authenticated else False
+        is_admin = user and user.is_authenticated and user.is_superuser
 
         if is_admin:
             pass
@@ -104,5 +104,5 @@ def download(self, request, pk=None, path=None, *args, **kwargs):
         methods=["get"],
     )
     def link(self, request, pk=None, path=None, *args, **kwargs):
-        logger.warning(f"REQUESTED ASSET LINK FOR PK:{pk} PATH:{path}")
+        logger.debug(f"REQUESTED ASSET LINK FOR PK:{pk} PATH:{path}")
         return self._get_file(request, pk, attachment=False, path=path)