email-forensics

电子邮件取证

全面的电子邮件取证技能，用于分析电子邮件、邮箱存档和电子邮件元数据。支持调查网络钓鱼攻击、商业电子邮件入侵（BEC）和电子邮件欺骗，并从电子邮件数据中提取具有取证价值的工件。

功能

• 邮箱解析：解析 PST、OST、MBOX、EML 和 MSG 文件
• 邮件头分析：深入分析邮件头和路由信息
• 附件提取：提取并分析电子邮件附件
• 网络钓鱼检测：识别网络钓鱼指标和手法
• 欺骗检测：检测电子邮件欺骗和冒充
• 链接分析：提取并分析电子邮件内容中的 URL
• 时间线生成：创建基于电子邮件的通信时间线
• 会话重建：重建电子邮件会话线程
• 元数据提取：提取发件人、收件人和路由元数据
• 身份验证分析：分析 SPF、DKIM 和 DMARC 结果

快速开始

from email_forensics import EmailAnalyzer, MailboxParser, PhishingDetector

# Parse mailbox file
parser = MailboxParser("/evidence/mailbox.pst")
emails = parser.get_all_messages()

# Analyze single email
analyzer = EmailAnalyzer()
analysis = analyzer.analyze_file("/evidence/suspicious.eml")

# Detect phishing
detector = PhishingDetector()
results = detector.scan_email(analysis)

用法

任务 1：邮箱解析

输入：邮箱文件（PST、OST、MBOX）

流程：

1. 加载并验证邮箱文件
2. 解析文件夹结构
3. 提取邮件
4. 为元数据编制索引
5. 生成邮箱摘要

输出：已解析的邮箱及邮件清单

示例：

from email_forensics import MailboxParser

# Parse Outlook PST file
parser = MailboxParser("/evidence/user_mailbox.pst")

# Get mailbox info
info = parser.get_mailbox_info()
print(f"Mailbox type: {info.format}")
print(f"Total messages: {info.message_count}")
print(f"Total folders: {info.folder_count}")
print(f"Date range: {info.oldest_date} - {info.newest_date}")

# List folders
folders = parser.get_folders()
for folder in folders:
    print(f"Folder: {folder.name}")
    print(f"  Path: {folder.path}")
    print(f"  Messages: {folder.message_count}")
    print(f"  Unread: {folder.unread_count}")

# Get messages from folder
inbox = parser.get_messages(folder_path="Inbox")
for msg in inbox:
    print(f"[{msg.date}] From: {msg.sender}")
    print(f"  Subject: {msg.subject}")
    print(f"  To: {msg.recipients}")
    print(f"  Has attachments: {msg.has_attachments}")

# Search messages
results = parser.search(
    query="confidential",
    search_body=True,
    search_subject=True
)
for r in results:
    print(f"Match: {r.subject}")
    print(f"  Folder: {r.folder}")
    print(f"  Match context: {r.context}")

# Export messages
parser.export_messages(
    folder_path="Inbox",
    output_dir="/evidence/exported/",
    format="eml"
)

# Generate mailbox report
parser.generate_report("/evidence/mailbox_report.html")

任务 2：电子邮件邮件头分析

输入：电子邮件（EML、MSG 或原始邮件头）

流程：

1. 解析所有邮件头字段
2. 分析路由路径
3. 验证身份认证
4. 检测异常
5. 生成邮件头分析结果

输出：全面的邮件头分析结果

示例：

from email_forensics import HeaderAnalyzer

# Analyze email headers
analyzer = HeaderAnalyzer()
analysis = analyzer.analyze_file("/evidence/suspicious.eml")

# Get basic headers
print(f"From: {analysis.from_address}")
print(f"To: {analysis.to_addresses}")
print(f"Subject: {analysis.subject}")
print(f"Date: {analysis.date}")
print(f"Message-ID: {analysis.message_id}")

# Analyze routing path
routing = analysis.get_routing_path()
for hop in routing:
    print(f"Hop {hop.number}:")
    print(f"  From: {hop.from_server}")
    print(f"  By: {hop.by_server}")
    print(f"  Time: {hop.timestamp}")
    print(f"  Delay: {hop.delay_seconds}s")

# Get authentication results
auth = analysis.get_authentication()
print(f"SPF: {auth.spf_result}")
print(f"  SPF domain: {auth.spf_domain}")
print(f"DKIM: {auth.dkim_result}")
print(f"  DKIM domain: {auth.dkim_domain}")
print(f"DMARC: {auth.dmarc_result}")

# Detect anomalies
anomalies = analysis.detect_anomalies()
for a in anomalies:
    print(f"ANOMALY: {a.type}")
    print(f"  Description: {a.description}")
    print(f"  Severity: {a.severity}")

# Get original sender (envelope)
envelope = analysis.get_envelope_info()
print(f"Envelope From: {envelope.mail_from}")
print(f"Envelope To: {envelope.rcpt_to}")

# Get X-headers
x_headers = analysis.get_x_headers()
for header, value in x_headers.items():
    print(f"{header}: {value}")

# Export analysis
analysis.export_report("/evidence/header_analysis.html")

任务 3：网络钓鱼检测

输入：电子邮件

流程：

1. 分析发件人真实性
2. 检查 URL 是否存在恶意指标
3. 分析附件风险
4. 检测社会工程学手法
5. 计算风险评分

输出：包含风险评估的网络钓鱼分析结果

示例：

from email_forensics import PhishingDetector, EmailAnalyzer

# Initialize detector
detector = PhishingDetector()

# Analyze email
analyzer = EmailAnalyzer()
email = analyzer.parse_file("/evidence/suspicious.eml")

# Run phishing detection
result = detector.analyze(email)

print(f"Risk Score: {result.risk_score}/100")
print(f"Classification: {result.classification}")
print(f"Confidence: {result.confidence}")

# Get indicators
for indicator in result.indicators:
    print(f"INDICATOR: {indicator.type}")
    print(f"  Description: {indicator.description}")
    print(f"  Weight: {indicator.weight}")
    print(f"  Evidence: {indicator.evidence}")

# Check sender authenticity
sender = result.sender_analysis
print(f"Sender: {sender.display_name} <{sender.address}>")
print(f"  Display name mismatch: {sender.display_name_mismatch}")
print(f"  Domain reputation: {sender.domain_reputation}")
print(f"  First-time sender: {sender.first_time_sender}")

# Analyze URLs
for url in result.url_analysis:
    print(f"URL: {url.url}")
    print(f"  Domain: {url.domain}")
    print(f"  Display text: {url.display_text}")
    print(f"  Mismatch: {url.text_url_mismatch}")
    print(f"  Shortened: {url.is_shortened}")
    print(f"  Risk: {url.risk_level}")

# Check attachments
for att in result.attachment_analysis:
    print(f"Attachment: {att.filename}")
    print(f"  Type: {att.content_type}")
    print(f"  Risk: {att.risk_level}")
    print(f"  Double extension: {att.has_double_extension}")

# Export report
detector.generate_report(result, "/evidence/phishing_report.html")

任务 4：附件分析

输入：包含附件的电子邮件

流程：

1. 提取所有附件
2. 识别文件类型
3. 计算哈希值
4. 检查恶意软件指标
5. 提取元数据

输出：包含提取文件的附件分析结果

示例：

from email_forensics import AttachmentAnalyzer

# Initialize analyzer
analyzer = AttachmentAnalyzer()

# Extract from single email
attachments = analyzer.extract_from_email(
    email_path="/evidence/email.eml",
    output_dir="/evidence/attachments/"
)

for att in attachments:
    print(f"Attachment: {att.filename}")
    print(f"  Content-Type: {att.content_type}")
    print(f"  Size: {att.size}")
    print(f"  MD5: {att.md5}")
    print(f"  SHA256: {att.sha256}")
    print(f"  Detected type: {att.detected_type}")
    print(f"  Type mismatch: {att.type_mismatch}")
    print(f"  Extracted to: {att.output_path}")

# Analyze specific attachment
detailed = analyzer.analyze_file("/evidence/attachments/document.pdf")
print(f"Metadata: {detailed.metadata}")
print(f"Embedded objects: {detailed.embedded_objects}")
print(f"Scripts: {detailed.contains_scripts}")
print(f"Macros: {detailed.contains_macros}")

# Extract from mailbox
mailbox_attachments = analyzer.extract_from_mailbox(
    mailbox_path="/evidence/mailbox.pst",
    output_dir="/evidence/all_attachments/",
    filter_types=["application/pdf", "application/msword"]
)

# Find suspicious attachments
suspicious = analyzer.find_suspicious(attachments)
for s in suspicious:
    print(f"SUSPICIOUS: {s.filename}")
    print(f"  Reason: {s.reason}")

# Check against malware hashes
malware = analyzer.check_malware_hashes("/hashsets/malware.txt")

# Generate attachment report
analyzer.generate_report("/evidence/attachment_report.html")

任务 5：创建电子邮件时间线

输入：邮箱或电子邮件集合

流程：

1. 解析所有邮件
2. 提取时间戳
3. 构建时间顺序时间线
4. 识别通信模式
5. 可视化活动

输出：电子邮件通信时间线

示例：

from email_forensics import EmailTimeline

# Initialize timeline
timeline = EmailTimeline()

# Add email sources
timeline.add_mailbox("/evidence/user1.pst")
timeline.add_mailbox("/evidence/user2.pst")
timeline.add_folder("/evidence/exported_emails/")

# Build timeline
events = timeline.build()

for event in events:
    print(f"[{event.timestamp}] {event.direction}")
    print(f"  From: {event.sender}")
    print(f"  To: {event.recipients}")
    print(f"  Subject: {event.subject}")

# Filter by date range
filtered = timeline.filter_by_date(
    start="2024-01-01",
    end="2024-01-31"
)

# Filter by participants
participant_emails = timeline.filter_by_participant("[email protected]")

# Get communication patterns
patterns = timeline.analyze_patterns()
print(f"Total messages: {patterns.total_messages}")
print(f"Unique senders: {patterns.unique_senders}")
print(f"Unique recipients: {patterns.unique_recipients}")
print(f"Peak hours: {patterns.peak_hours}")
print(f"Top correspondents: {patterns.top_correspondents}")

# Detect anomalies
anomalies = timeline.detect_anomalies()
for a in anomalies:
    print(f"ANOMALY: {a.description}")
    print(f"  Time: {a.timestamp}")

# Export timeline
timeline.export_csv("/evidence/email_timeline.csv")
timeline.generate_visualization("/evidence/email_timeline.html")

任务 6：电子邮件会话重建

输入：电子邮件

流程：

1. 按会话分组
2. 分析 In-Reply-To 邮件头
3. 构建会话层级
4. 识别缺失邮件
5. 重建完整会话线程

输出：重建后的电子邮件会话线程

示例：

from email_forensics import ThreadReconstructor

# Initialize reconstructor
reconstructor = ThreadReconstructor()

# Load emails
reconstructor.load_mailbox("/evidence/mailbox.pst")

# Reconstruct all threads
threads = reconstructor.reconstruct_all()

for thread in threads:
    print(f"Thread: {thread.subject}")
    print(f"  Messages: {thread.message_count}")
    print(f"  Participants: {thread.participants}")
    print(f"  Duration: {thread.start_date} - {thread.end_date}")
    print(f"  Complete: {thread.is_complete}")

    # Print thread hierarchy
    for msg in thread.messages:
        indent = "  " * msg.depth
        print(f"{indent}[{msg.date}] {msg.sender}: {msg.subject}")

# Find specific thread
thread = reconstructor.find_thread(subject_contains="Project Alpha")

# Find threads with missing messages
incomplete = reconstructor.find_incomplete_threads()
for t in incomplete:
    print(f"Incomplete: {t.subject}")
    print(f"  Missing IDs: {t.missing_message_ids}")

# Export threads
reconstructor.export_threads(
    output_dir="/evidence/threads/",
    format="mbox"
)

# Generate thread report
reconstructor.generate_report("/evidence/threads_report.html")

任务 7：欺骗检测

输入：电子邮件

流程：

1. 验证发件人邮件头
2. 检查身份认证记录
3. 分析显示名称伪装手法
4. 比较信封发件人与邮件头发件人
5. 检测冒充行为

输出：欺骗分析结果

示例：

from email_forensics import SpoofingDetector

# Initialize detector
detector = SpoofingDetector()

# Analyze email
result = detector.analyze_file("/evidence/suspicious.eml")

print(f"Spoofing detected: {result.is_spoofed}")
print(f"Confidence: {result.confidence}")

# Header vs Envelope analysis
print(f"Header From: {result.header_from}")
print(f"Envelope From: {result.envelope_from}")
print(f"Mismatch: {result.from_mismatch}")

# Display name analysis
display = result.display_name_analysis
print(f"Display Name: {display.name}")
print(f"Homograph attack: {display.homograph_detected}")
print(f"Executive impersonation: {display.executive_impersonation}")
print(f"Brand impersonation: {display.brand_impersonation}")

# Authentication analysis
auth = result.authentication_analysis
print(f"SPF Pass: {auth.spf_pass}")
print(f"DKIM Pass: {auth.dkim_pass}")
print(f"DMARC Pass: {auth.dmarc_pass}")

# Reply-To analysis
reply_to = result.reply_to_analysis
print(f"Reply-To: {reply_to.address}")
print(f"Reply-To differs from From: {reply_to.differs_from_sender}")

# Get all indicators
for indicator in result.indicators:
    print(f"INDICATOR: {indicator.name}")
    print(f"  Evidence: {indicator.evidence}")
    print(f"  Severity: {indicator.severity}")

# Export report
detector.generate_report(result, "/evidence/spoofing_analysis.html")

任务 8：链接分析

输入：电子邮件内容

流程：

1. 提取所有 URL
2. 分析 URL 组成部分
3. 与威胁情报进行比对
4. 检测 URL 混淆手法
5. 识别重定向链

输出：URL 分析结果

示例：

from email_forensics import LinkAnalyzer

# Initialize analyzer
analyzer = LinkAnalyzer()

# Extract links from email
links = analyzer.extract_from_email("/evidence/email.eml")

for link in links:
    print(f"URL: {link.url}")
    print(f"  Display text: {link.display_text}")
    print(f"  Domain: {link.domain}")
    print(f"  TLD: {link.tld}")
    print(f"  Text matches URL: {link.text_matches_url}")
    print(f"  Is shortened: {link.is_shortened}")
    print(f"  Is IP-based: {link.is_ip_based}")
    print(f"  Risk score: {link.risk_score}")

# Unshorten URLs
unshortened = analyzer.unshorten_urls(links)
for u in unshortened:
    print(f"Short: {u.short_url}")
    print(f"Final: {u.final_url}")
    print(f"Redirects: {u.redirect_count}")

# Check against threat intelligence
threats = analyzer.check_threat_intel(
    links,
    feed_path="/feeds/malicious_urls.txt"
)
for t in threats:
    print(f"THREAT: {t.url}")
    print(f"  Category: {t.category}")
    print(f"  Source: {t.intel_source}")

# Detect URL obfuscation
obfuscated = analyzer.detect_obfuscation(links)
for o in obfuscated:
    print(f"OBFUSCATED: {o.url}")
    print(f"  Technique: {o.obfuscation_type}")
    print(f"  Decoded: {o.decoded_url}")

# Analyze link destinations (safe fetch)
destinations = analyzer.analyze_destinations(links, safe_mode=True)

# Export link analysis
analyzer.generate_report("/evidence/link_analysis.html")

任务 9：商业电子邮件入侵分析

输入：电子邮件或邮箱

流程：

1. 识别 BEC 指标
2. 检测紧迫性措辞
3. 分析财务请求
4. 检查发件人是否合法
5. 评估 BEC 可能性

输出：BEC 分析结果

示例：

from email_forensics import BECDetector

# Initialize BEC detector
detector = BECDetector()

# Analyze single email
result = detector.analyze_email("/evidence/wire_request.eml")

print(f"BEC Score: {result.bec_score}/100")
print(f"Classification: {result.classification}")

# Check BEC indicators
for indicator in result.indicators:
    print(f"INDICATOR: {indicator.type}")
    print(f"  Description: {indicator.description}")
    print(f"  Evidence: {indicator.evidence}")
    print(f"  Weight: {indicator.weight}")

# Language analysis
language = result.language_analysis
print(f"Urgency detected: {language.urgency_score}")
print(f"Authority claims: {language.authority_score}")
print(f"Financial keywords: {language.financial_keywords}")
print(f"Secrecy requests: {language.secrecy_score}")

# Sender analysis
sender = result.sender_analysis
print(f"Claimed identity: {sender.claimed_identity}")
print(f"Actual sender: {sender.actual_address}")
print(f"Executive impersonation: {sender.executive_impersonation}")

# Request analysis
request = result.request_analysis
print(f"Action requested: {request.action}")
print(f"Amount mentioned: {request.amount}")
print(f"Account details: {request.has_account_details}")
print(f"Wire transfer request: {request.wire_transfer}")

# Scan mailbox for BEC
mailbox_results = detector.scan_mailbox("/evidence/mailbox.pst")
for r in mailbox_results.high_risk:
    print(f"HIGH RISK: {r.subject}")
    print(f"  BEC Score: {r.bec_score}")

# Generate BEC report
detector.generate_report("/evidence/bec_analysis.html")

任务 10：电子邮件搜索与导出

输入：邮箱文件或电子邮件集合

流程：

1. 为电子邮件内容编制索引
2. 执行搜索查询
3. 筛选结果
4. 导出匹配项
5. 生成搜索报告

输出：搜索结果及导出的电子邮件

示例：

from email_forensics import EmailSearcher

# Initialize searcher
searcher = EmailSearcher("/evidence/mailbox.pst")

# Build search index
searcher.build_index()

# Search by keywords
results = searcher.search(
    query="confidential project",
    search_body=True,
    search_subject=True,
    search_attachments=True
)

for r in results:
    print(f"Match: {r.subject}")
    print(f"  From: {r.sender}")
    print(f"  Date: {r.date}")
    print(f"  Score: {r.relevance_score}")
    print(f"  Snippet: {r.snippet}")

# Search by sender
sender_emails = searcher.search_by_sender("[email protected]")

# Search by date range
date_range = searcher.search_by_date(
    start="2024-01-01",
    end="2024-01-31"
)

# Search by attachment name
with_attachments = searcher.search_by_attachment(
    filename_pattern="*.pdf"
)

# Complex query
complex_results = searcher.advanced_search(
    sender_contains="@example.com",
    subject_contains="wire transfer",
    date_after="2024-01-01",
    has_attachments=True
)

# Export search results
searcher.export_results(
    results,
    output_dir="/evidence/search_results/",
    format="eml",
    include_attachments=True
)

# Generate search report
searcher.generate_report("/evidence/search_report.html")

配置

环境变量

变量	描述	必填	默认值
EMAIL_PARSER	电子邮件解析库的路径	否	内置
THREAT_INTEL_FEED	威胁情报源的 URL	否	无
VT_API_KEY	VirusTotal API 密钥	否	无
SAFE_BROWSE_KEY	Google Safe Browsing API 密钥	否	无

选项

选项	类型	描述
extract_attachments	boolean	自动提取附件
decode_mime	boolean	解码 MIME 编码的内容
parse_html	boolean	解析 HTML 邮件正文
safe_url_check	boolean	安全 URL 验证
parallel	boolean	启用并行处理

示例

示例 1：钓鱼活动调查

场景：调查针对该组织的钓鱼活动

from email_forensics import MailboxParser, PhishingDetector

# Parse quarantined emails
parser = MailboxParser("/evidence/quarantine.pst")
emails = parser.get_all_messages()

# Initialize phishing detector
detector = PhishingDetector()

# Analyze all emails
phishing_emails = []
for email in emails:
    result = detector.analyze(email)
    if result.risk_score > 70:
        phishing_emails.append(result)
        print(f"PHISHING: {email.subject}")
        print(f"  Risk: {result.risk_score}")
        print(f"  Indicators: {len(result.indicators)}")

# Extract IOCs from phishing emails
iocs = detector.extract_iocs(phishing_emails)
print(f"Malicious URLs: {len(iocs.urls)}")
print(f"Sender addresses: {len(iocs.senders)}")

# Generate campaign report
detector.generate_campaign_report(phishing_emails, "/evidence/phishing_campaign.html")

示例 2：BEC 事件调查

场景：调查可能的商业电子邮件入侵

from email_forensics import BECDetector, EmailTimeline, SpoofingDetector

# Analyze the suspicious request email
bec = BECDetector()
result = bec.analyze_email("/evidence/wire_request.eml")

print(f"BEC Score: {result.bec_score}")
print(f"Financial request: {result.request_analysis.wire_transfer}")

# Check for spoofing
spoof = SpoofingDetector()
spoof_result = spoof.analyze_file("/evidence/wire_request.eml")
print(f"Spoofed: {spoof_result.is_spoofed}")

# Build communication timeline
timeline = EmailTimeline()
timeline.add_mailbox("/evidence/cfo_mailbox.pst")
timeline.add_mailbox("/evidence/finance_mailbox.pst")

# Find related emails
related = timeline.filter_by_participant(result.sender_analysis.actual_address)
print(f"Related emails from sender: {len(related)}")

限制

• 大型邮箱可能需要较长的处理时间
• 加密邮件需要解密密钥
• 某些专有格式的支持可能有限
• URL 分析需要网络访问以进行验证
• 附件分析取决于文件类型支持情况
• BEC 检测可能出现误报
• 标头分析的准确性取决于邮件的保留情况

故障排除

常见问题 1：PST 文件损坏

问题：无法解析 PST 文件 解决方案：

• 分析前使用 PST 修复工具
• 尝试使用不同的解析库
• 如有可能，提取单封邮件

常见问题 2：编码内容未解码

问题：邮件正文显示为编码文本 解决方案：

• 启用 MIME 解码
• 检查是否使用了不常见的字符编码
• 尝试不同的解码方法

常见问题 3：附件缺失

问题：未提取附件 解决方案：

• 检查附件大小限制
• 确认是否支持该附件格式
• 查找内嵌附件

相关技能

• 网络取证：分析电子邮件网络流量
• 浏览器取证：调查 Webmail
• 恶意软件取证：分析恶意附件
• 时间线取证：整合电子邮件时间线
• 日志取证：与邮件服务日志进行关联

参考资料

• 电子邮件取证参考资料
• 电子邮件标头分析指南
• 钓鱼检测模式