Enforce owner checks for upload attachments

src/upload_handler.py

@@ -8,7 +8,7 @@ import hashlib
 import mimetypes
 import threading
 from datetime import datetime, timedelta
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 from fastapi import HTTPException, UploadFile
 def secure_filename(filename: str) -> str:
     """Sanitize a filename (replaces werkzeug.utils.secure_filename)."""
@@ -225,6 +225,106 @@ class UploadHandler:
         """Validate that the upload ID matches the expected pattern."""
         pattern = r'^[0-9a-fA-F]{32}\.[A-Za-z0-9]+$'
         return re.fullmatch(pattern, upload_id) is not None
+
+    def _inside_upload_dir(self, path: str) -> bool:
+        """Check if path is inside the upload directory."""
+        base = os.path.realpath(self.upload_dir)
+        p = os.path.realpath(path)
+        try:
+            return os.path.commonpath([base, p]) == base
+        except Exception:
+            return False
+
+    def _load_upload_index(self) -> Dict[str, Any]:
+        uploads_db_path = os.path.join(self.upload_dir, "uploads.json")
+        if not os.path.exists(uploads_db_path):
+            return {}
+        try:
+            with open(uploads_db_path, "r") as f:
+                data = json.load(f)
+            return data if isinstance(data, dict) else {}
+        except Exception as e:
+            logger.warning(f"Failed to read uploads database: {e}")
+            return {}
+
+    def get_upload_info(self, upload_id: str) -> Optional[Dict[str, Any]]:
+        """Return the uploads.json metadata row for an upload ID, if present."""
+        if not self.validate_upload_id(upload_id):
+            return None
+        for info in self._load_upload_index().values():
+            if isinstance(info, dict) and info.get("id") == upload_id:
+                return dict(info)
+        return None
+
+    def _find_upload_path(self, upload_id: str) -> Optional[str]:
+        """Find an upload file by ID while staying inside upload_dir."""
+        if not self.validate_upload_id(upload_id):
+            return None
+
+        direct = os.path.join(self.upload_dir, upload_id)
+        if os.path.exists(direct) and self._inside_upload_dir(direct):
+            return direct
+
+        for root, _dirs, files in os.walk(self.upload_dir, followlinks=False):
+            if upload_id in files:
+                path = os.path.join(root, upload_id)
+                if self._inside_upload_dir(path):
+                    return path
+        return None
+
+    def resolve_upload(
+        self,
+        upload_id: str,
+        owner: Optional[str] = None,
+        auth_manager: Any = None,
+        allow_admin: bool = True,
+    ) -> Optional[Dict[str, Any]]:
+        """Resolve an upload ID to metadata only if the caller may read it.
+
+        This is the owner-aware lookup used by internal processors. Public
+        download routes already perform owner checks; chat/document paths must
+        do the same before reading file bytes server-side.
+        """
+        if not self.validate_upload_id(upload_id):
+            logger.warning(f"Invalid upload ID format: {upload_id}")
+            return None
+
+        auth_configured = bool(auth_manager and getattr(auth_manager, "is_configured", False))
+        if auth_configured and not owner:
+            return None
+
+        info = self.get_upload_info(upload_id) or {}
+        is_admin = False
+        if allow_admin and owner and auth_manager and hasattr(auth_manager, "is_admin"):
+            try:
+                is_admin = bool(auth_manager.is_admin(owner))
+            except Exception:
+                is_admin = False
+
+        if owner and not is_admin:
+            if info.get("owner") != owner:
+                logger.warning("Upload %s denied for owner %s", upload_id, owner)
+                return None
+        if not owner and info.get("owner") is not None:
+            logger.warning("Upload %s denied without an authenticated owner", upload_id)
+            return None
+
+        path = info.get("path")
+        if not path or not os.path.exists(path) or not self._inside_upload_dir(path):
+            path = self._find_upload_path(upload_id)
+        if not path:
+            return None
+        if not self._inside_upload_dir(path):
+            logger.warning(f"Upload path outside upload directory: {path}")
+            return None
+
+        resolved = dict(info)
+        resolved.setdefault("id", upload_id)
+        resolved["path"] = path
+        resolved.setdefault("name", os.path.basename(path))
+        resolved.setdefault("original_name", resolved["name"])
+        resolved.setdefault("mime", mimetypes.guess_type(path)[0] or "application/octet-stream")
+        return resolved
     
     def cleanup_rate_limits(self):
         """Remove stale entries from upload_rate_log."""