Merge pull request #3186 from RachHavoc/rcm/cwe-22-fix

Fix CWE 22 in Payload API

Author: deacon-mp

app/api/v2/handlers/payload_api.py

@@ -2,6 +2,7 @@
 import itertools
 import os
 import pathlib
+import re
 from io import IOBase
 
 import aiohttp_apispec
@@ -47,7 +48,7 @@ async def get_payloads(self, request: web.Request):
             if add_path
             else self.file_svc.remove_xored_extension(p.name)
             for p in itertools.chain.from_iterable(p_dir.glob('[!.]*') for p_dir in payload_dirs)
-            if p.is_file()
+            if p.is_file() and not p.is_symlink() and not p.name.startswith('.')
         }
 
         payloads = list(payloads)
@@ -69,12 +70,24 @@ async def post_payloads(self, request: web.Request):
         # accessing the file using the prefilled request["form"] dictionary.
         file_field: web.FileField = request["form"]["file"]
 
-        file_name, file_path = await self.__generate_file_name_and_path(file_field)
+        # Sanitize the file name to prevent directory traversal
+        sanitized_filename = self.sanitize_filename(file_field.filename)
 
-        # The file_field.file is of type IOBase: It uses blocking methods.
-        # Putting blocking code into a dedicated method and thread...
+        # Generate the file name and path
+        file_name, file_path = await self.__generate_file_name_and_path(sanitized_filename)
+
+        # Save the file to a temporary location first
+        temp_file_path = pathlib.Path(file_path).parent / f"temp_{file_name}"
         loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
-        await loop.run_in_executor(None, self.__save_file, file_path, file_field.file)
+        await loop.run_in_executor(None, self.__save_file, str(temp_file_path), file_field.file)
+
+        # Validate the saved file to ensure it is not a symbolic link
+        if temp_file_path.is_symlink():
+            temp_file_path.unlink()  
+            raise web.HTTPBadRequest(reason="Uploaded file is a symbolic link and is not allowed.")
+
+        # Move the validated file to the final destination
+        temp_file_path.rename(file_path)
 
         body: dict[list[str]] = {"payloads": [file_name]}
         return web.json_response(body)
@@ -90,34 +103,38 @@ async def post_payloads(self, request: web.Request):
     @aiohttp_apispec.match_info_schema(PayloadDeleteRequestSchema)
     async def delete_payloads(self, request: web.Request):
         file_name: str = request.match_info.get("name")
-        file_path: str = os.path.join('data/payloads/', file_name)
-
-        response: web.HTTPException = None
         try:
-            os.remove(file_path)
+            safe_path = self.validate_and_canonicalize_path(file_name)
+            if pathlib.Path(safe_path).is_symlink():
+                raise ValueError(f"Invalid path: {file_name} is a symbolic link.")
+            os.remove(safe_path)
             response = web.HTTPNoContent()
+        except ValueError as e:
+            response = web.HTTPNotFound(reason=str(e))
         except FileNotFoundError:
             response = web.HTTPNotFound()
+        except PermissionError:
+            response = web.HTTPForbidden(reason="Permission denied.")
         return response
 
     @classmethod
-    async def __generate_file_name_and_path(cls, file_field: web.FileField) -> [str, str]:
+    async def __generate_file_name_and_path(cls, sanitized_filename: str) -> [str, str]:
         """
         Finds whether an uploaded file already exists in the payload directory.
         In the case, generates a new file name with an incremental suffix to avoid overriding the existing one.
         Otherwise, the original file name is used.
 
-        :param file_field: The upload payload object.
+        :param sanitized_filename: The sanitized file name.
         :return: A tuple containing the generated file name and path for future storage.
         """
-        file_name_candidate: str = file_field.filename
+        file_name_candidate: str = sanitized_filename
         file_path: str = os.path.join('data/payloads/', file_name_candidate)
         suffix: int = 1
 
         # Generating a file suffix in the case it already exists.
         while os.path.exists(file_path):
-            file_name_candidate = f"{pathlib.Path(file_field.filename).stem}_" \
-                                  f"{suffix}{pathlib.Path(file_field.filename).suffix}"
+            file_name_candidate = f"{pathlib.Path(sanitized_filename).stem}_" \
+                                  f"{suffix}{pathlib.Path(sanitized_filename).suffix}"
             file_path = os.path.join('data/payloads/', file_name_candidate)
             suffix += 1
         file_name: str = file_name_candidate
@@ -142,3 +159,35 @@ def __save_file(target_file_path: str, io_base_src: IOBase):
                     buffered_io_base_dest.write(chunk)
                 else:
                     read_chunk = False
+
+    @staticmethod
+    def validate_and_canonicalize_path(input_path: str, base_directory: str = "data/payloads/") -> str:
+        """
+        Validates and canonicalizes a file path to ensure it is within the designated directory.
+
+        :param input_path: The input file path to validate.
+        :param base_directory: The base directory to constrain paths to.
+        :return: The canonicalized absolute path if valid.
+        :raises ValueError: If the path resolves outside the base directory.
+        """
+        base_dir = pathlib.Path(base_directory).resolve()
+
+        try:
+            resolved_path = (base_dir / pathlib.Path(input_path).name).resolve()
+            resolved_path.relative_to(base_dir)
+        except ValueError:
+            raise ValueError(f"Invalid path: {input_path} resolves outside the designated directory {base_directory}")
+        except Exception as e:
+            raise ValueError(f"Invalid path: {input_path}. Error: {e}")
+
+        return str(resolved_path)
+
+    @staticmethod
+    def sanitize_filename(filename: str) -> str:
+        """
+        Sanitizes a file name to remove potentially dangerous characters.
+
+        :param filename: The original file name.
+        :return: A sanitized file name.
+        """
+        return re.sub(r'[^\w\.-]', '_', filename)