Root cause
Path & link handling introduced
feat: add email attachment extraction support
How AI contributed: Direct introductionPublished Apr 29, 2026 in
fatbobman/mail-mcp-bridge
A flaw has been found in fatbobman mail-mcp-bridge up to 1.3.3. Affected is an unknown function of the file src/mail_mcp_server.py. Executing a manipulation of the argument message_ids can lead to path traversal. The attack can be executed remotely. The exploit has been published and may be used. Upgrading to version 1.3.4 is able to address this issue. This patch is called 638b162b26532e32fa8d8047f638537dbdfe197a. Upgrading the affected component is recommended.
message_id.strip('<>') is joined as a directory then shutil.rmtree; parent had no attachment module.
Root cause
feat: add email attachment extraction support
Fix
The fix maps each message ID to a URL-safe encoded directory name, resolves it beneath the attachment root, and makes extraction and cleanup use that checked path instead of joining raw IDs.
Fix by fatbobman · no AI marker found
Case facts
diff --git a/cleanup_attachments.py b/cleanup_attachments.pynew file mode 100644index 0000000..aeb574f--- /dev/null+++ b/cleanup_attachments.py@@ -0,0 +1,152 @@+#!/usr/bin/env python3+"""+Clean up temporary attachment directories created by extract_attachments++Provides functionality to remove attachment directories for specific messages+"""++import os+from pathlib import Path+from typing import Dict, List, Any+++def get_attachment_base_dir() -> str:+ """+ Get the base directory for saving attachments++ Reads from MAIL_ATTACHMENT_PATH environment variable (defaults to /tmp),+ then appends 'mail-mcp-attachments' subdirectory.++ Returns:+ Full directory path (e.g., /tmp/mail-mcp-attachments)+ """+ base_temp = os.environ.get('MAIL_ATTACHMENT_PATH', '/tmp')+ return os.path.join(base_temp, 'mail-mcp-attachments')+++def cleanup_attachments(message_ids: List[str], base_dir: str = None) -> Dict[str, Any]:+ """+ Remove attachment directories for specified message IDs++ Args:+ message_ids: List of RFC Message-IDs to clean up+ base_dir: Optional override for base directory++ Returns:+ Dictionary containing:+ {+ "success": True/False,+ "base_dir": "...",+ "cleaned": [+ {+ "message_id": "...",+ "path": "...",+ "files_removed": 3,+ "size_freed": 12345+ }+ ],+ "not_found": ["..."],+ "error": "..." (if failed)+ }+ """+ base_dir = base_dir or get_attachment_base_dir()+ base_path = Path(base_dir)++ if not base_path.exists():+ return {+ "success": True,+ "base_dir": base_dir,+ "cleaned": [],+ "not_found": message_ids,+ "note": "Base directory does not exist"+ }++ cleaned = []+ not_found = []++ for message_id in message_ids:+ # Clean message_id for use as directory name (remove angle brackets)+ clean_message_id = message_id.strip('<>')+ message_dir = base_path / clean_message_id++ if message_dir.exists() and message_dir.is_dir():+ # Calculate size before deletion+ total_size = 0+ file_count = 0+ for item in message_dir.rglob('*'):+ if item.is_file():+ total_size += item.stat().st_size+ file_count += 1++ # Remove directory+ try:+ import shutil+ shutil.rmtree(message_dir)++ cleaned.append({+ "message_id": message_id,+ "path": str(message_dir),+ "files_removed": file_count,+ "size_freed": total_size+ })+ except Exception as e:+ return {+ "success": False,+ "base_dir": base_dir,+ "error": f"Failed to remove {message_dir}: {str(e)}"+ }+ else:+ not_found.append(message_id)++ return {+ "success": True,+ "base_dir": base_dir,+ "cleaned": cleaned,+ "not_found": not_found+ }+++def main():+ """Command line test tool"""+ import sys++ if len(sys.argv) < 2:+ print("Usage: python3 cleanup_attachments.py <message_id1> [message_id2 ...]")+ print("\nExample:")+ print(" python3 cleanup_attachments.py \\<[email protected]\\>")+ print("\nEnvironment Variables:")+ print(" MAIL_ATTACHMENT_PATH Base directory for attachments (default: /tmp/mail-mcp-attachments)")+ sys.exit(1)++ message_ids = sys.argv[1:]++ print(f"Cleaning up attachments for {len(message_ids)} message(s)...\n")++ result = cleanup_attachments(message_ids)++ if result['success']:+ print(f"✅ Cleanup successful\n")+ print(f"Base directory: {result['base_dir']}")++ if result['cleaned']:+ print(f"\n🗑️ Cleaned ({len(result['cleaned'])}):")+ for item in result['cleaned']:+ print(f" ✓ {item['message_id']}")+ print(f" Path: {item['path']}")+ print(f" Files removed: {item['files_removed']}")+ print(f" Space freed: {item['size_freed']:,} bytes ({item['size_freed'] / 1024:.1f} KB)")++ if result.get('note'):+ print(f"\nℹ️ {result['note']}")++ if result['not_found']:+ print(f"\n⚠️ Not found ({len(result['not_found'])}):")+ for message_id in result['not_found']:+ print(f" - {message_id}")+ else:+ print(f"❌ {result['error']}")+ sys.exit(1)+++if __name__ == '__main__':+ main()diff --git a/extract_attachments.py b/extract_attachments.pynew file mode 100644index 0000000..2d91851--- /dev/null+++ b/extract_attachments.py@@ -0,0 +1,281 @@+#!/usr/bin/env python3+"""+Extract attachments from .emlx files and save to a temporary directory++Supports extracting specific attachments from emails for AI analysis+"""++import email+import email.policy+from email.header import decode_header+from pathlib import Path+from typing import Dict, List, Any, Optional+import os+import shutil+++def decode_header_value(value: str) -> str:+ """+ Decode encoded email header values (e.g. =?utf-8?B?...?=)++ Args:+ value: Encoded header value++ Returns:+ Decoded string+ """+ if not value:+ return ""++ decoded_parts = []+ for content, encoding in decode_header(value):+ if isinstance(content, bytes):+ if encoding:+ try:+ decoded_parts.append(content.decode(encoding))+ except (LookupError, UnicodeDecodeError):+ decoded_parts.append(content.decode('utf-8', errors='replace'))+ else:+ decoded_parts.append(content.decode('utf-8', errors='replace'))+ else:+ decoded_parts.append(str(content))++ return ''.join(decoded_parts)+++def get_attachment_base_dir() -> str:+ """+ Get the base directory for saving attachments++ Reads from MAIL_ATTACHMENT_PATH environment variable (defaults to /tmp),+ then appends 'mail-mcp-attachments' subdirectory.++ Returns:+ Full directory path (e.g., /tmp/mail-mcp-attachments)+ """+ base_temp = os.environ.get('MAIL_ATTACHMENT_PATH', '/tmp')+ return os.path.join(base_temp, 'mail-mcp-attachments')+++def extract_attachments(+ file_path: str,+ message_id: str,+ filenames: List[str],+ output_dir: Optional[str] = None+) -> Dict[str, Any]:+ """+ Extract specific attachments from an .emlx file++ Args:+ file_path: Absolute path to .emlx file+ message_id: RFC Message-ID (used for creating subdirectory)+ filenames: List of attachment filenames to extract+ output_dir: Optional override for base output directory++ Returns:+ Dictionary containing:+ {+ "success": True/False,+ "base_dir": "...",+ "message_dir": "...",+ "extracted": [+ {+ "filename": "...",+ "path": "...",+ "mime_type": "...",+ "size_bytes": 12345+ }+ ],+ "not_found": ["..."],+ "error": "..." (if failed)+ }+ """+ file_path_obj = Path(file_path)++ if not file_path_obj.exists():+ return {+ "success": False,+ "error": f"File not found: {file_path}"+ }++ if not filenames:+ return {+ "success": False,+ "error": "No filenames specified for extraction"+ }++ try:+ # Read email content+ with open(file_path_obj, 'rb') as f:+ lines = f.readlines()++ # Parse .emlx format+ if len(lines) < 2:+ return {+ "success": False,+ "error": "Invalid file format or empty file"+ }++ # Extract email content (skip size line and plist metadata)+ email_lines = []+ for line in lines[1:]:+ line_str = line.decode('utf-8', errors='ignore')+ if '<?xml version' in line_str or '<!DOCTYPE plist' in line_str or '<plist version' in line_str:+ break+ email_lines.append(line)++ raw_content = b''.join(email_lines)+ msg = email.message_from_bytes(raw_content, policy=email.policy.compat32)++ # Create output directory structure+ base_dir = output_dir or get_attachment_base_dir()+ # Clean message_id for use as directory name (remove angle brackets)+ clean_message_id = message_id.strip('<>')+ message_dir = Path(base_dir) / clean_message_id+ message_dir.mkdir(parents=True, exist_ok=True)++ # Track extraction results+ extracted = []+ not_found = []+ filenames_set = set(filenames)++ # Walk through MIME parts and extract attachments+ for part in msg.walk():+ content_type = part.get_content_type()+ content_disposition = str(part.get('Content-Disposition', ''))+ filename = part.get_filename()++ # Check if this is an attachment we're looking for+ is_attachment = (+ 'attachment' in content_disposition or+ (filename and content_type not in ['text/plain', 'text/html'])+ )++ if is_attachment and filename:+ # Decode filename if encoded+ decoded_filename = decode_header_value(filename)++ # Check if this filename is in our requested list+ if decoded_filename in filenames_set:+ # Get attachment payload+ payload = part.get_payload(decode=True)++ # If payload is empty or too small, try to find in Attachments directory+ if not payload or len(payload) < 100:+ # Try to find attachment in Mail's file system+ # The attachment is usually in: /path/to/.../Data/{X}/{Y}/{Z}/Attachments/{message_number}/{attachment_id}/{filename}+ # Path structure: .../Data/{X}/{Y}/{Z}/Messages/{number}.partial.emlx+ try:+ messages_dir = file_path_obj.parent+ attachments_dir = messages_dir.parent / "Attachments"+ message_num = file_path_obj.stem.replace('.partial', '')++ # Search in attachment subdirectories+ if attachments_dir.exists():+ # Search for the file in any subdirectory+ for att_dir in attachments_dir.iterdir():+ if att_dir.is_dir() and att_dir.name == message_num:+ for sub_dir in att_dir.iterdir():+ if sub_dir.is_dir():+ # Try exact match first+ potential_file = sub_dir / decoded_filename+ if potential_file.exists():+ # Read from file system+ payload = potential_file.read_bytes()+ break++ # Try fuzzy match (Mail replaces / with _ in filenames)+ # and other special characters+ if not payload:+ # Try different variations+ alt_filename = decoded_filename.replace('/', '_').replace('\\', '_')+ potential_file = sub_dir / alt_filename+ if potential_file.exists():+ payload = potential_file.read_bytes()+ break+ except Exception:+ pass # Fall through to empty payload handling++ if payload:+ # Save to file+ # Sanitize filename for filesystem (replace problematic characters)+ safe_filename = decoded_filename.replace('/', '_').replace('\\', '_').replace(':', '_')+ output_path = message_dir / safe_filename+ with open(output_path, 'wb') as f:+ f.write(payload)++ extracted.append({+ "filename": decoded_filename, # Report original filename+ "safe_filename": safe_filename, # Actual filesystem name+ "path": str(output_path),+ "mime_type": content_type,+ "size_bytes": len(payload)+ })++ # Remove from set to track what we've found+ filenames_set.discard(decoded_filename)++ # Any remaining filenames in the set were not found+ not_found = list(filenames_set)++ return {+ "success": True,+ "base_dir": base_dir,+ "message_dir": str(message_dir),+ "extracted": extracted,+ "not_found": not_found+ }++ except Exception as e:+ return {+ "success": False,+ "error": f"Extraction failed: {str(e)}"+ }+++def main():+ """Command line test tool"""+ import sys++ if len(sys.argv) < 4:+ print("Usage: python3 extract_attachments.py <emlx_file> <message_id> <filename1> [filename2 ...]")+ print("\nExample:")+ print(" python3 extract_attachments.py ~/Library/Mail/V10/.../email.emlx \\<[email protected]\\> report.pdf")+ print("\nEnvironment Variables:")+ print(" MAIL_ATTACHMENT_PATH Base directory for attachments (default: /tmp/mail-mcp-attachments)")+ sys.exit(1)++ emlx_file = sys.argv[1]+ message_id = sys.argv[2]+ filenames = sys.argv[3:]++ print(f"Extracting attachments from: {emlx_file}")+ print(f"Message-ID: {message_id}")+ print(f"Requested files: {', '.join(filenames)}\n")++ result = extract_attachments(emlx_file, message_id, filenames)++ if result['success']:+ print(f"✅ Extraction successful\n")+ print(f"Base directory: {result['base_dir']}")+ print(f"Message directory: {result['message_dir']}")++ if result['extracted']:+ print(f"\n📦 Extracted ({len(result['extracted'])}):")+ for item in result['extracted']:+ print(f" ✓ {item['filename']}")+ print(f" Path: {item['path']}")+ print(f" Type: {item['mime_type']}")+ print(f" Size: {item['size_bytes']:,} bytes")++ if result['not_found']:+ print(f"\n❌ Not found ({len(result['not_found'])}):")+ for filename in result['not_found']:+ print(f" - {filename}")+ else:+ print(f"❌ {result['error']}")+ sys.exit(1)+++if __name__ == '__main__':+ main()diff --git a/src/attachment_paths.py b/src/attachment_paths.pynew file mode 100644index 0000000..3beff21--- /dev/null+++ b/src/attachment_paths.py@@ -0,0 +1,48 @@+"""+Helpers for safely mapping message IDs onto attachment directories.+"""++import base64+from pathlib import Path+from typing import Union+++def normalize_message_id(message_id: str) -> str:+ """+ Normalize a Message-ID for use as a directory name.++ Message-ID values are untrusted input. We only allow a single path segment+ after stripping optional angle brackets.+ """+ clean_message_id = message_id.strip().strip('<>')++ if not clean_message_id:+ raise ValueError("Message-ID is empty")++ return clean_message_id+++def get_message_dir_name(message_id: str) -> str:+ """+ Encode a Message-ID into a single safe directory name.+ """+ clean_message_id = normalize_message_id(message_id)+ encoded = base64.urlsafe_b64encode(clean_message_id.encode("utf-8")).decode("ascii")+ return f"mid_{encoded.rstrip('=')}"+++def get_message_dir(base_dir: Union[str, Path], message_id: str) -> Path:+ """+ Resolve the directory for a Message-ID and ensure it stays under base_dir.+ """+ base_path = Path(base_dir).resolve(strict=False)+ message_dir_name = get_message_dir_name(message_id)+ message_dir = (base_path / message_dir_name).resolve(strict=False)++ try:+ message_dir.relative_to(base_path)+ except ValueError as exc:+ raise ValueError("Message-ID resolves outside the attachment directory") from exc++ return message_dirdiff --git a/src/cleanup_attachments.py b/src/cleanup_attachments.pyindex aeb574f..aae2750 100644--- a/src/cleanup_attachments.py+++ b/src/cleanup_attachments.py@@ -11,0 +12,5 @@ from typing import Dict, List, Any+try:+ from .attachment_paths import get_message_dir+except ImportError:+ from attachment_paths import get_message_dir+@@ -48,0 +54,6 @@ def cleanup_attachments(message_ids: List[str], base_dir: str = None) -> Dict[st+ "invalid": [+ {+ "message_id": "...",+ "error": "..."+ }+ ],@@ -65,0 +77 @@ def cleanup_attachments(message_ids: List[str], base_dir: str = None) -> Dict[st+ invalid = []@@ -68,3 +80,8 @@ def cleanup_attachments(message_ids: List[str], base_dir: str = None) -> Dict[st- # Clean message_id for use as directory name (remove angle brackets)- clean_message_id = message_id.strip('<>')- message_dir = base_path / clean_message_id+ try:+ message_dir = get_message_dir(base_path, message_id)+ except ValueError as e:+ invalid.append({+ "message_id": message_id,+ "error": str(e)+ })+ continue@@ -105 +122,2 @@ def cleanup_attachments(message_ids: List[str], base_dir: str = None) -> Dict[st- "not_found": not_found+ "not_found": not_found,+ "invalid": invaliddiff --git a/src/extract_attachments.py b/src/extract_attachments.pyindex 2d91851..fb5b35a 100644--- a/src/extract_attachments.py+++ b/src/extract_attachments.py@@ -14 +14,5 @@ import os-import shutil++try:+ from .attachment_paths import get_message_dir+except ImportError:+ from attachment_paths import get_message_dir@@ -132,3 +136 @@ def extract_attachments(- # Clean message_id for use as directory name (remove angle brackets)- clean_message_id = message_id.strip('<>')- message_dir = Path(base_dir) / clean_message_id+ message_dir = get_message_dir(base_dir, message_id)AI-assisted change 7867f06ec4c0a93f2ada3f7a4ec9338fec8833ccb653690a25d8b9d2d1a50dd6 · Fix aea57c5a606401552f14d0a39e192e21cdcff549fd284934ecd7cdcd3808cd89