memora

Instructions

Implementation Reference

• memora/server.py:1356-1456 (handler)

  Implementation of the memory_upload_image tool, which handles image file validation, R2 storage upload, and returning the image reference.

  async def memory_upload_image(
      file_path: str,
      memory_id: int,
      image_index: int = 0,
      caption: Optional[str] = None,
  ) -> Dict[str, Any]:
      """Upload an image file directly to R2 storage.
  
      Uploads a local image file to R2 and returns the r2:// reference URL
      that can be used in memory metadata.
  
      Args:
          file_path: Absolute path to the image file to upload
          memory_id: Memory ID this image belongs to (used for organizing in R2)
          image_index: Index of image within the memory (default: 0)
          caption: Optional caption for the image
  
      Returns:
          Dictionary with r2_url (the r2:// reference) and image object ready for metadata
      """
      from pathlib import Path as _Path
  
      from PIL import Image as _PILImage
  
      from .image_storage import get_image_storage_instance
  
      image_storage = get_image_storage_instance()
      if not image_storage:
          return {
              "error": "r2_not_configured",
              "message": "R2 storage is not configured. Set MEMORA_STORAGE_URI to s3:// and configure AWS credentials.",
          }
  
      # --- Path validation (defense in depth) ---
      raw_path = _Path(file_path)
  
      # 1. Reject symlinks anywhere in the path chain
      for part in [raw_path] + list(raw_path.parents):
          if part.is_symlink():
              return {"error": "invalid_path", "message": "Symlinks are not supported"}
  
      try:
          resolved = raw_path.resolve(strict=True)
      except (OSError, ValueError):
          return {"error": "file_not_found", "message": "File not found"}
  
      # 2. Validate extension — aligned with image_storage.py ext_map
      _UPLOAD_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
      if resolved.suffix.lower() not in _UPLOAD_EXTENSIONS:
          return {"error": "invalid_type", "message": "File must be an image (jpg, jpeg, png, gif, webp)"}
  
      # 3. Block known sensitive directories
      _BLOCKED_PATTERNS = [".ssh", ".gnupg", ".aws", ".config/gcloud", "id_rsa", "id_ed25519", ".env"]
      path_str = str(resolved).lower()
      for pattern in _BLOCKED_PATTERNS:
          if pattern in path_str:
              return {"error": "blocked_path", "message": "Cannot upload files from sensitive directories"}
  
      # 4. Verify file is actually an image and derive MIME from content
      _PILLOW_TO_MIME = {"JPEG": "image/jpeg", "PNG": "image/png", "GIF": "image/gif", "WEBP": "image/webp"}
      try:
          with _PILImage.open(str(resolved)) as img:
              img.verify()
              pillow_format = img.format
      except Exception:
          return {"error": "invalid_image", "message": "File is not a valid image"}
  
      content_type = _PILLOW_TO_MIME.get(pillow_format)
      if not content_type:
          return {"error": "unsupported_format", "message": f"Unsupported image format: {pillow_format}"}
  
      try:
          # Read file and upload
          with open(str(resolved), "rb") as f:
              image_data = f.read()
  
          r2_url = image_storage.upload_image(
              image_data=image_data,
              content_type=content_type,
              memory_id=memory_id,
              image_index=image_index,
          )
  
          # Build image object for metadata
          image_obj = {"src": r2_url}
          if caption:
              image_obj["caption"] = caption
  
          # Don't echo local file_path in response (path disclosure fix)
          return {
              "r2_url": r2_url,
              "image": image_obj,
              "content_type": content_type,
              "size_bytes": len(image_data),
          }
  
      except Exception as e:
          logger.error("Failed to upload image for memory %s: %s", memory_id, e)
          return {"error": "upload_failed", "message": "Image upload failed. Check server logs for details."}