Python: keep attachments close (#7038)

* Python: keep attachments close

* Python: close attachment edge cases

python/samples/05-end-to-end/chatkit-integration/app.py

@@ -587,12 +587,17 @@ async def upload_file(attachment_id: str, file: UploadFile = File(...)):  # noqa
     """
     logger.info(f"Receiving file upload for attachment: {attachment_id}")
 
+    try:
+        file_path = attachment_store.get_file_path(attachment_id)
+    except ValueError:
+        logger.warning(f"Rejected invalid attachment ID: {attachment_id!r}")
+        return JSONResponse(status_code=400, content={"error": "Invalid attachment ID."})
+
     try:
         # Read file contents
         contents = await file.read()
 
         # Save to disk
-        file_path = attachment_store.get_file_path(attachment_id)
         file_path.write_bytes(contents)
 
         logger.info(f"Saved {len(contents)} bytes to {file_path}")
@@ -625,7 +630,11 @@ async def preview_image(attachment_id: str):
 
     try:
         file_path = attachment_store.get_file_path(attachment_id)
+    except ValueError:
+        logger.warning(f"Rejected invalid attachment ID: {attachment_id!r}")
+        return JSONResponse(status_code=400, content={"error": "Invalid attachment ID."})
 
+    try:
         if not file_path.exists():
             return JSONResponse(status_code=404, content={"error": "File not found"})
 

python/samples/05-end-to-end/chatkit-integration/attachment_store.py

@@ -48,7 +48,7 @@ class FileBasedAttachmentStore(AttachmentStore[dict[str, Any]]):
             base_url: Base URL for generating upload and preview URLs
             data_store: Optional data store to persist attachment metadata
         """
-        self.uploads_dir = Path(uploads_dir)
+        self.uploads_dir = Path(uploads_dir).resolve()
         self.base_url = base_url.rstrip("/")
         self.data_store = data_store
 
@@ -56,8 +56,24 @@ class FileBasedAttachmentStore(AttachmentStore[dict[str, Any]]):
         self.uploads_dir.mkdir(parents=True, exist_ok=True)
 
     def get_file_path(self, attachment_id: str) -> Path:
-        """Get the filesystem path for an attachment."""
-        return self.uploads_dir / attachment_id
+        """Get the filesystem path for an attachment.
+
+        Args:
+            attachment_id: Identifier used as the attachment filename.
+
+        Returns:
+            The resolved path within the uploads directory.
+
+        Raises:
+            ValueError: If the attachment ID does not resolve to a direct child of the uploads directory.
+        """
+        if not attachment_id or attachment_id in {".", ".."} or "/" in attachment_id or "\\" in attachment_id:
+            raise ValueError(f"Invalid attachment ID: {attachment_id!r}")
+
+        file_path = (self.uploads_dir / attachment_id).resolve()
+        if not file_path.is_relative_to(self.uploads_dir) or file_path.parent != self.uploads_dir:
+            raise ValueError(f"Invalid attachment ID: {attachment_id!r}")
+        return file_path
 
     async def delete_attachment(self, attachment_id: str, context: dict[str, Any]) -> None:
         """Delete an attachment and its file from disk."""