"""Email document parser for Mosaic — stdlib email parsing. SECURITY: Attachments are metadata-only. Never execute, open, or auto-extract attachment contents. """ import email import email.policy import re import hashlib import logging from typing import Optional from .base import ParseResult logger = logging.getLogger(__name__) # Suspicious MIME types that warrant flagging SUSPICIOUS_MIME_TYPES = { 'application/x-executable', 'application/x-msdos-program', 'application/x-msdownload', 'application/vnd.ms-excel.sheet.macroEnabled.12', 'application/vnd.ms-word.document.macroEnabled.12', 'application/x-shellscript', 'application/javascript', 'application/x-bat', 'application/x-com', 'application/x-dosexec', } # ANSI escape pattern ANSI_ESCAPE = re.compile(r'\x1b\[[0-9;]*[a-zA-Z]|\x1b\].*?\x07') class EmailDocParser: """Parse email documents — headers + body, attachments as metadata only.""" def parse(self, file_path: str) -> ParseResult: """Parse an email file (EML/MBOX format or HTML email page).""" try: with open(file_path, 'r', encoding='utf-8', errors='replace') as f: content = f.read() # Try standard email parsing first msg = email.message_from_string(content, policy=email.policy.default) # If it doesn't look like a real email, try HTML extraction if not msg['From'] and not msg['Subject']: return self._parse_html_email(content, file_path) return self._parse_email_message(msg) except Exception as e: return ParseResult(error=f"Email parse error: {str(e)}") def _parse_email_message(self, msg) -> ParseResult: """Parse a standard email.Message object.""" # Extract headers headers = {} for key in ('From', 'To', 'Cc', 'Bcc', 'Subject', 'Date', 'Message-ID', 'In-Reply-To', 'References'): val = msg.get(key) if val: headers[key] = str(val) # Extract body body_parts = [] if msg.is_multipart(): for part in msg.walk(): content_type = part.get_content_type() disposition = str(part.get('Content-Disposition', '')) if 'attachment' in disposition: continue # Skip attachments (metadata extracted separately) if content_type == 'text/plain': payload = part.get_payload(decode=True) if payload: body_parts.append(payload.decode('utf-8', errors='replace')) elif content_type == 'text/html': payload = part.get_payload(decode=True) if payload: # Simple HTML → text text = self._html_to_text(payload.decode('utf-8', errors='replace')) body_parts.append(text) else: payload = msg.get_payload(decode=True) if payload: body_parts.append(payload.decode('utf-8', errors='replace')) body = '\n\n'.join(body_parts) # Extract attachment metadata (NEVER the content) attachments = self._extract_attachment_metadata(msg) # Build formatted output text_parts = [] for key, val in headers.items(): text_parts.append(f"{key}: {val}") text_parts.append('') # blank line text_parts.append(body) text = '\n'.join(text_parts) # Strip ANSI text = ANSI_ESCAPE.sub('', text) metadata = { 'headers': headers, 'attachments': attachments, 'attachment_count': len(attachments), 'has_suspicious_attachments': any( a.get('suspicious') for a in attachments ), } return ParseResult( success=True, text=text, metadata=metadata, doc_type='email', title=headers.get('Subject', ''), char_count=len(text), ) def _extract_attachment_metadata(self, msg) -> list[dict]: """Extract attachment metadata ONLY — never content.""" attachments = [] if not msg.is_multipart(): return attachments for part in msg.walk(): disposition = str(part.get('Content-Disposition', '')) if 'attachment' not in disposition: continue filename = part.get_filename() or 'unnamed' content_type = part.get_content_type() payload = part.get_payload(decode=True) size = len(payload) if payload else 0 md5 = hashlib.md5(payload).hexdigest() if payload else '' att = { 'filename': filename, 'content_type': content_type, 'size': size, 'md5': md5, 'suspicious': content_type in SUSPICIOUS_MIME_TYPES, } if att['suspicious']: logger.warning("Suspicious attachment: %s (%s)", filename, content_type) attachments.append(att) return attachments def _parse_html_email(self, content: str, file_path: str) -> ParseResult: """Parse an HTML page that contains email content (e.g., WikiLeaks email viewer).""" try: from bs4 import BeautifulSoup soup = BeautifulSoup(content, 'html.parser') # Try to extract email fields from HTML structure metadata = {} text_parts = [] # Look for common WikiLeaks email display patterns for field in ('from', 'to', 'cc', 'subject', 'date'): el = soup.find(class_=re.compile(field, re.I)) if el: metadata[field] = el.get_text(strip=True) text_parts.append(f"{field.title()}: {metadata[field]}") # Get body body = soup.find(class_=re.compile('body|content|message', re.I)) if body: text_parts.append('') text_parts.append(body.get_text(separator='\n', strip=True)) else: # Fallback: get all text text_parts.append('') text_parts.append(soup.get_text(separator='\n', strip=True)) text = '\n'.join(text_parts) text = ANSI_ESCAPE.sub('', text) return ParseResult( success=True, text=text, metadata={'headers': metadata, 'attachments': [], 'parsed_from_html': True}, doc_type='email', title=metadata.get('subject', ''), char_count=len(text), ) except Exception as e: return ParseResult(error=f"HTML email parse error: {str(e)}") @staticmethod def _html_to_text(html: str) -> str: """Simple HTML to text conversion.""" try: from bs4 import BeautifulSoup soup = BeautifulSoup(html, 'html.parser') return soup.get_text(separator='\n', strip=True) except Exception: # Fallback: strip tags with regex return re.sub(r'<[^>]+>', '', html)