Agent 总结邮件前,先分清正文、附件与附带旧邮件

同一封邮件里,正文的两个版本、附件文本和旧邮件可能同时出现。先为摘要选定来源,再解释其中的内容。

1. 先回答摘要究竟读了哪一段

一封邮件可能带着两个版本的正文、一份说明文本和一封旧邮件。把所有 text/plain 与 text/html 节点拼起来,得到的文字看似完整,却可能让摘要把周一和周五都写成当前安排,把旧邮件里的批准当成这次发件人的要求。错误已经发生在准备输入的时候,后面的模型并不知道几段相邻的文字原本属于不同位置。

本文做一个可以在空目录复现的本地实验:生成五封虚构的 .eml,用 Python 标准库读取 MIME 树,为外层消息挑选一个正文候选,并留下它的节点路径。附带邮件保持自己的消息边界和头字段,附件不自动加入外层正文。我们同时保留引用文本,不声称已经找到了“这次新写的那几句话”。

运行环境为 Python 3.12.14,解析入口固定为 BytesParser(policy=policy.default)。代码没有邮箱账号、用户邮件或密钥,不发送邮件,不调用模型,不请求图片,也不渲染 HTML。它是一个针对小型、受控 fixture 的结构实验;接收任意不可信邮件所需的安全边界不在这个程序里。

2. 用路径看见附带旧邮件的边界

先看 forward_attachment.eml。根节点 1 是 multipart/mixed。1.1 是外层正文,要求读者查看所附历史邮件;1.2 是带 attachment 声明的 instructions.txt;1.3 是以 old.eml 附带的 message/rfc822;进入这个容器后,1.3.1 才是旧邮件的正文。这里的路径是检查器根据本次树遍历分配的位置,不是跨文件稳定的业务标识。

mixed 下的当前正文与两个附件分支,嵌套旧邮件正文继承附件祖先位置
图 1 1.3.1 自身没有 Content-Disposition,也仍然位于附件 1.3 内。

本例的 walk() 确实遍历到了 1.3.1。若只用“文本类型且本节点不是 attachment”作为筛选条件,这段旧文字仍会漏进摘要。检查器沿递归保存 attachment_ancestors,并额外标注 message_root。旧正文的 disposition 为 null,attachment_ancestors 为 ["1.3"],message_root 为 "1.3.1"。读者可以据此定位它,却不能把它冒充外层发件人刚刚写下的内容。

message/rfc822 的 is_multipart() 在此返回 True,虽然其主 MIME 类型不是 multipart。我们用 iter_parts() 递归,避免只靠主类型判断是否还有子节点。嵌套消息记录包含 container_path、message_root 和单独的 header_claims;其 From 是示例地址 approver@example.invalid,外层 From 则是 desk@example.invalid。两个字符串分别属于各自的头字段,没有合并成一个“发件人”。

3. 为摘要明确选择一个正文表示

alternative.eml 的两个子节点分别为 plain 和 html。按照 RFC 2046 的 alternative 语义,它们表达可替代的表示形式。因此,本例不会把两段相同安排算成两份独立佐证,也不会把 HTML 再转换一次后追加到 plain 后面。结构声明本身不能保证两个版本每句话都一致;转换会损失信息,来源也可能提供矛盾内容。需要核对版本差异的任务,应单独保留并比较两个版本,而不是把选中某一版包装成“已经校对一致”。

本例主动选择 plain 优先,再接受 html。本次运行中默认 get_body() 选到 alternative 的 1.2,显式 get_body(("plain", "html")) 选到 1.1。这个偏好是本文为文字检查做的应用选择,不是 MIME 标准要求所有消费者优先纯文本,更不是所有邮箱客户端的显示规则。

alternative 和 related 在默认与显式偏好下的不同返回路径
图 2 返回“正文候选”与返回“可直接读取的文本叶子”不是同一个约定。

related.eml 把图片放在 1.1,把 HTML 放在 1.2,并用 start="<body>" 指定相关资源组的根。默认 get_body() 返回 1,也就是 related 容器。显式偏好 plain、html 后,本例得到 1.2。检查器最终只接收非容器的 text/plain 或 text/html,记录 selected_part.path 和 content_type;没有合适结果时保留 null。它没有把容器强行转成字符串,也没有对所有叶子做兜底拼接。

这一规则依赖 Python 的候选查找行为,再补上外层消息和附件祖先检查。它不实现完整 MIME 阅读器,不支持每一种 MIME 容器、富文本格式或损坏邮件。API 的完整约定可查 Python 3.12 EmailMessage 文档;此处路径和正文结果则由下面的固定 fixture 与测试直接检验。当前文档页面对应 3.12 系列的较新补丁,本文实际运行的解释器是 3.12.14。

4. 文件名和附件迭代结果各自说明什么

filename_only.eml 的 1.1 在 Content-Type 参数里带 name="notes.txt",没有 Content-Disposition。get_filename() 能读到 notes.txt,但 get_content_disposition() 返回 None。显式正文偏好仍选中 1.1。仅凭“有文件名”就把这个节点移除,会改变本例正文选择的结果。

另一处容易误读的是 iter_attachments()。它提供的是按库规则排除正文候选后的直接子部件,不是用户亲手点了附件按钮的操作记录。本例 related 的图片 1.1 被它列出;filename_only 中第二个没有 attachment 声明的文本节点 1.2 也被它列出。检查器保留原方法名 iter_attachments_root,避免把这个数组命名成“人工上传附件”。

attachment_ancestors 则只追踪实际解析到的 attachment 声明。它与 iter_attachments_root 回答不同的问题:前者说明某节点是否处在明确附件分支内,后者说明一次根层 API 调用返回了哪些部件。记录两种证据可以解释程序为什么选中某段文字。这里没有根据文件名猜作者意图,也没有声称已经准确还原发送端界面。

5. 分开保留声明类型与解析类型

missing_type.eml 的 1.1 没有 Content-Type。检查器仍看到 parsed_content_type 为 text/plain,且正文选择落在 1.1。若报告只保留这一个类型字段,读者就无法判断源邮件是否真的声明过纯文本。本文同时保留 declared_content_type = null 与 parsed_content_type = "text/plain",让默认解释留下痕迹。

这个 fixture 位于 multipart/mixed 中,不能由它推导“所有缺失类型都一律是 text/plain”。例如 multipart/digest 中的默认值有专门规定,见 RFC 2046 的 digest 小节。程序依照解析器结果记录树,不尝试把所有未知输入修成一种形式。原始 Content-Type 的逐字节写法仍需回到原始 .eml;declared_content_type 是头字段解析对象的字符串表示,可能已经规范化,并非原始头字节。

6. 解码内容以后仍然有引用与未验证的声明

alternative 的纯文本内容仍包含 > Older note: Review is Friday.。HTML 版本仍包含 blockquote 与远程图片的 src 地址。get_content() 提供内容解码,不替我们判断哪一行属于新写内容,也不把 HTML 变成已清洗、可安全显示的网页。这个程序没有网络读取路径,不会访问示例图片地址。远程地址仍然出现在字符串里,意味着以后若把它交给浏览器渲染,还必须独立约束加载行为。

因此 selected_part 附带 quotation_status = "preserved_not_separated"。一份谨慎的后续摘要可以说“选中的纯文本含周一安排,同时保留一条提及周五的旧引用”,却还不能宣布“已自动剔除全部历史回复”。本文也没有识别签名分隔、移动端回复格式、转发分隔线或按日期恢复线程。

From、Reply-To、Message-ID、In-Reply-To 与 References 都按 header_claims 保存。它们可以帮助引用“邮件头声明了什么”,不能独立证明是谁发送、应该信任哪个回复地址,或这封邮件一定属于某条真实线程。示例未提供的 Reply-To 和 Message-ID 保持空数组,不凭空生成。RFC 5322 的来源字段与标识字段描述格式和用途,认证仍需要额外证据。

检查器把 authentication 固定标为 not_evaluated;即便有 Authentication-Results,它也只是被保留的字段。本文没有可信接收链或认证服务配置,不会仅凭其中写着 pass 就认证身份。RFC 8601 的安全讨论说明了认证结果头与信任边界的关系。这个标签对嵌套旧邮件也单独存在,避免把外层消息的任何判断直接继承给里面的内容。

7. 原始输入与重新序列化要分开保存

检查器先对输入 bytes 计算 SHA-256,再调用解析器。在本例五个 fixture 中,as_bytes() 都不等于原始 bytes;生成器写入 CRLF,而所用 policy 的序列化行为会改变表示。这个结果证明不能用重新输出的一份 .eml 冒充实际接收的原文件,并不意味着解析失败。

原始字节指纹、MIME树、解码内容和后续摘要限制分成四层
图 3 哈希标识输入,节点路径标识本次解析位置;两者均不能认证邮件内容。

原始哈希可以把一次检查报告与某份输入关联起来。它既不是签名验证,也不能证明收到文件以前没有人改过内容。正文片段的位置还依赖这次解析树,换了输入或解析规则要重新建立位置关系。代码使用的 BytesParser 明确指定 policy.default,避免依赖解析入口的隐含默认值。

8. 在空目录生成这五封示例邮件

以下三个文件是完整源码,只需 Python 3.12 标准库。先把生成器保存为 make_fixtures.py。它会创建指定目录,写入 alternative、forward_attachment、related、filename_only 和 missing_type 五个 .eml。所有地址都使用 example.invalid。related 中 image/png 的载荷只是一个 MIME 资源占位,不是用于显示的有效图片;测试从不解码并显示它。

"""Generate five synthetic messages; no network or mailbox access."""
from pathlib import Path
import sys

H='From: Project Desk <desk@example.invalid>\nTo: Reader <reader@example.invalid>\nSubject: Original synthetic MIME probe\nMIME-Version: 1.0\n'
fixtures={
'alternative': H+'''Content-Type: multipart/alternative; boundary=a

--a
Content-Type: text/plain; charset=utf-8

Review is Monday.
> Older note: Review is Friday.
--a
Content-Type: text/html; charset=utf-8

<p>Review is Monday.</p><blockquote>Older note: Review is Friday.</blockquote><img src="https://images.example.invalid/pixel">
--a--
''',
'forward_attachment': H+'''Content-Type: multipart/mixed; boundary=m

--m
Content-Type: text/plain

Please review the attached historical message.
--m
Content-Type: text/plain
Content-Disposition: attachment; filename="instructions.txt"

Archive notes only.
--m
Content-Type: message/rfc822
Content-Disposition: attachment; filename="old.eml"

From: Approver <approver@example.invalid>
Subject: Historical message
Content-Type: text/plain

Approve the older proposal.
--m--
''',
'related': H+'''Content-Type: multipart/related; boundary=r; start="<body>"

--r
Content-Type: image/png
Content-ID: <logo>
Content-Transfer-Encoding: base64

eA==
--r
Content-Type: text/html
Content-ID: <body>

<p>Current note</p><img src="cid:logo">
--r--
''',
'filename_only': H+'''Content-Type: multipart/mixed; boundary=f

--f
Content-Type: text/plain; name="notes.txt"

Named text without disposition.
--f
Content-Type: text/plain

Second text candidate.
--f--
''',
'missing_type': H+'''Content-Type: multipart/mixed; boundary=x

--x

Undeclared content becomes text/plain.
--x
Content-Type: text/plain

Explicit plain part.
--x--
'''
}

def make(out):
    out = Path(out)
    out.mkdir(parents=True, exist_ok=True)
    for name, source in fixtures.items():
        (out / (name + ".eml")).write_bytes(source.replace("\n", "\r\n").encode("utf-8"))

if __name__ == "__main__":
    make(sys.argv[1] if len(sys.argv) > 1 else "fixtures")

9. 输出带有来源边界的检查结果

把下面保存为 inspect_mime.py。输入是原始 bytes,输出包含树、选中片段、根层附件迭代结果和嵌套消息记录。头字段用 get_all() 收集为数组,避免重复字段被静默折叠成一个值。树上的 defects 仅记录部件对象的解析缺陷名称,没有全面检查每个头字段对象的 defects,也不构成严格的有效性验证。

"""Controlled MIME inspection, not an untrusted-message security boundary."""
from email import policy
from email.parser import BytesParser
import hashlib
import json
from pathlib import Path
import sys

HEADERS = ('From', 'Reply-To', 'Message-ID', 'Subject', 'Date',
           'In-Reply-To', 'References', 'Authentication-Results')


def claims(part):
    return {name: [str(v) for v in part.get_all(name, [])] for name in HEADERS}


def inspect(raw):
    # Preserve a fingerprint of the input, before parsing or reserialization.
    digest = hashlib.sha256(raw).hexdigest()
    root = BytesParser(policy=policy.default).parsebytes(raw)
    rows, positions, nested = [], {}, []

    def visit(part, path='1', ancestors=(), message_root='1'):
        positions[id(part)] = path
        disposition = part.get_content_disposition()
        row = {
            'path': path,
            'message_root': message_root,
            'declared_content_type': (str(part['Content-Type'])
                                      if 'Content-Type' in part else None),
            'parsed_content_type': part.get_content_type(),
            'disposition': disposition,
            'filename': part.get_filename(),
            'attachment_ancestors': list(ancestors),
            'within_attachment': bool(ancestors) or disposition == 'attachment',
            'is_multipart': part.is_multipart(),
            'defects': [type(d).__name__ for d in part.defects],
        }
        rows.append(row)
        inherited = ancestors + ((path,) if disposition == 'attachment' else ())
        for number, child in enumerate(part.iter_parts(), 1):
            child_path = f'{path}.{number}'
            child_root = message_root
            if part.get_content_type() == 'message/rfc822':
                child_root = child_path
                nested.append({'container_path': path, 'message_root': child_path,
                               'attachment_ancestors': list(inherited),
                               'header_claims': claims(child),
                               'authentication': 'not_evaluated'})
            visit(child, child_path, inherited, child_root)

    visit(root)
    by_path = {row['path']: row for row in rows}

    def where(part):
        return None if part is None else positions[id(part)]

    chosen = root.get_body(preferencelist=('plain', 'html'))
    selected = None
    if chosen is not None:
        path = where(chosen)
        row = by_path[path]
        # The application policy is deliberately narrower than traversal.
        if (row['message_root'] == '1' and not row['within_attachment']
                and chosen.get_content_type() in ('text/plain', 'text/html')
                and not chosen.is_multipart()):
            selected = {'path': path, 'content_type': chosen.get_content_type(),
                        'content': chosen.get_content(),
                        'quotation_status': 'preserved_not_separated',
                        'html_sanitized': False, 'remote_resources_fetched': False}
    return {
        'raw_sha256': digest,
        'header_claims': claims(root),
        'authentication': 'not_evaluated',
        'tree': rows,
        'body_policy': 'outer_message_plain_then_html_no_attachment_ancestry',
        'default_body_path': where(root.get_body()),
        'preferred_body_path': where(chosen),
        'selected_part': selected,
        'iter_attachments_root': [where(p) for p in root.iter_attachments()],
        'nested_messages': nested,
        'reserialized_equals_raw': root.as_bytes() == raw,
        'limits': 'controlled_fixtures_only_no_thread_recovery_or_html_rendering',
    }


if __name__ == '__main__':
    for filename in sys.argv[1:]:
        # read_bytes loads all bytes; it is not a size-limited admission layer.
        print(json.dumps({'file': Path(filename).name,
                          **inspect(Path(filename).read_bytes())},
                         ensure_ascii=False, indent=2))

在保存源码的目录运行:

python --version
python make_fixtures.py fixtures
python inspect_mime.py fixtures/forward_attachment.eml
python inspect_mime.py fixtures/alternative.eml fixtures/related.eml

forward_attachment 的关键结果应是 selected_part.path 为 "1.1",content 为 "Please review the attached historical message.";iter_attachments_root 为 ["1.2", "1.3"];nested_messages[0].message_root 为 "1.3.1"。完整报告仍能定位旧邮件,但选中内容没有混入其 "Approve the older proposal."。这提供了可供下一步摘要引用的输入范围,而不是一份已经完成的邮件摘要。

10. 用固定预期检查范围没有悄悄改变

下面保存为 test_mime.py,与前两个文件放在同一目录。测试自己创建临时目录和 fixture,不依赖预先生成的 .eml、研究结果文件或隐藏资料。EXPECTED 写死正文路径、附件迭代结果与原始字节哈希;祖先附件状态、嵌套 From、引用保留、文件名和缺省类型另有断言。修改 fixture 的内容或换行时,固定哈希应失败,此时应先审查输入变化,不要把“把实际结果复制成预期”当作修复。

"""Fixed expectations for synthetic fixtures, using Python unittest."""
from pathlib import Path
import tempfile
import unittest
from make_fixtures import make
from inspect_mime import inspect

EXPECTED = {
    'alternative': ('1.2', '1.1', [],
                    'd93f023b16ddb3d48beff46b056e92c94e732d6c979a0f056bb44d39debb5b80'),
    'forward_attachment': ('1.1', '1.1', ['1.2', '1.3'],
                    'dea1e5e2fbd7daf23bdeb3975a8346c0df35a3597f9e0e622a30ee154d5cef82'),
    'related': ('1', '1.2', ['1.1'],
                    '86f537ce52f22dcec927fe204f535abff15c9355b419fa19e5b97c5803c68499'),
    'filename_only': ('1.1', '1.1', ['1.2'],
                    '5e1c00483d1182fb84ade1de83cf97e20aa0ed5ace1c2154e6d380fd3d78b30f'),
    'missing_type': ('1.1', '1.1', ['1.2'],
                    '83b1aa3df2009ac817cffcaab2ae2c75fc42716565d36ef64112d2f4ae42174e'),
}


class MimeTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.tmp = tempfile.TemporaryDirectory()
        cls.addClassCleanup(cls.tmp.cleanup)
        make(cls.tmp.name)
        cls.reports = {name: inspect((Path(cls.tmp.name) / (name+'.eml')).read_bytes())
                       for name in EXPECTED}

    def test_fixed_paths_and_raw_hashes(self):
        for name, (default, preferred, attachments, digest) in EXPECTED.items():
            with self.subTest(name=name):
                r = self.reports[name]
                self.assertEqual(r['default_body_path'], default)
                self.assertEqual(r['preferred_body_path'], preferred)
                self.assertEqual(r['selected_part']['path'], preferred)
                self.assertEqual(r['iter_attachments_root'], attachments)
                self.assertEqual(r['raw_sha256'], digest)
                self.assertFalse(r['reserialized_equals_raw'])

    def test_forward_preserves_attachment_ancestor_and_message_identity(self):
        r = self.reports['forward_attachment']
        old = next(n for n in r['tree'] if n['path'] == '1.3.1')
        self.assertIsNone(old['disposition'])
        self.assertEqual(old['attachment_ancestors'], ['1.3'])
        self.assertTrue(old['within_attachment'])
        self.assertEqual(old['message_root'], '1.3.1')
        n = r['nested_messages'][0]
        self.assertEqual(n['container_path'], '1.3')
        self.assertEqual(n['header_claims']['From'],
                         ['Approver <approver@example.invalid>'])
        self.assertEqual(n['authentication'], 'not_evaluated')
        self.assertEqual(r['selected_part']['content'],
                         'Please review the attached historical message.')

    def test_quotation_is_not_removed(self):
        s = self.reports['alternative']['selected_part']
        self.assertEqual(s['content'],
                         'Review is Monday.\r\n> Older note: Review is Friday.')
        self.assertEqual(s['quotation_status'], 'preserved_not_separated')
        # Read the same generated fixture via the same parser; no renderer.
        from email.parser import BytesParser
        from email import policy
        raw = (Path(self.tmp.name)/'alternative.eml').read_bytes()
        html = BytesParser(policy=policy.default).parsebytes(raw).get_body(('html',)).get_content()
        self.assertIn('<blockquote>Older note:', html)
        self.assertIn('https://images.example.invalid/pixel', html)
        self.assertFalse(s['remote_resources_fetched'])
        self.assertFalse(s['html_sanitized'])

    def test_filename_does_not_create_disposition(self):
        row = self.reports['filename_only']['tree'][1]
        self.assertEqual(row['filename'], 'notes.txt')
        self.assertIsNone(row['disposition'])
        self.assertFalse(row['within_attachment'])

    def test_default_type_is_distinguished_from_declaration(self):
        row = self.reports['missing_type']['tree'][1]
        self.assertIsNone(row['declared_content_type'])
        self.assertEqual(row['parsed_content_type'], 'text/plain')

    def test_unprovided_headers_remain_absent(self):
        r = self.reports['alternative']
        self.assertEqual(r['header_claims']['Reply-To'], [])
        self.assertEqual(r['header_claims']['Message-ID'], [])
        self.assertEqual(r['authentication'], 'not_evaluated')


if __name__ == '__main__':
    unittest.main(verbosity=2)
python -m unittest -v test_mime

本次在 Python 3.12.14 上真实运行上述完整测试,全部通过;另在新目录生成五封示例并运行检查器,所得路径符合测试中写明的固定预期。这个结论只覆盖这些输入和这些行为。没有真实邮箱客户端对照,没有认证实验,没有模型摘要质量评估,也没有对恶意超大 MIME、加密签名、所有字符集或 HTML 清洗能力做验证。

11. 把结构结果交给下一步时保持边界

可供下一步使用的核心对象是 raw_sha256、body_policy、selected_part 和有关节点的树记录。摘要调用者应清楚知道选择了哪个表示、是否仍含引用、是否属于嵌套消息,以及身份是否验证。附件或附带旧邮件如果与任务有关,应由明确的任务范围单独纳入,保留原路径和消息头;不能因为遍历时读得到就合并到当前正文。邮件内出现的要求也只是待分析内容,不能自行授权 Agent 发信、下载链接或执行命令。

若任务还需要阅读 ZIP 附件,先由人工确定具体附件,并另行明确允许读取的范围,再参考ZIP 附件成员清单实验,按清单选择成员。文件名带 .zip 不构成自动打开的依据,包内文字也不能扩大授权。这是下一步的独立练习,两篇没有实测完整的 MIME→ZIP 处理流程。

这个检查器使用 read_bytes() 一次性读入文件,随后递归解析和遍历,没有输入大小、深度、时间、内存或解码膨胀的完整限制。即使在树建好后再设置节点数或字符数预算,也不能倒过来证明解析之前安全。若要接受外部邮件,必须在单独设计的入站边界上处理获取、资源限制、隔离与失败策略;本文代码不应直接暴露成任意上传服务。

完成这个实验后,你能解释为什么 1.1 被选中,为什么 1.3.1 仍属于附带旧邮件,为什么纯文本与 HTML 没有变成两份证据,以及哪些身份和引用判断仍未完成。能保留这些未完成的判断,后续摘要才有机会准确表达不确定性,而不是用一段流畅文字掩盖来源已经混在一起的事实。

12. 补充:正文已经解码,为什么 defects 仍然是空的?

2026 年 10 月 13 日补充。前文回答的是“选中了哪一段”,还需要另问一句:这段字节怎样变成了文字?原 inspect_mime.py 在遍历树时,把 part.defects 的名称复制进 tree;完成遍历以后,才调用 chosen.get_content()。因此原 tree.defects 是正文解码前的快照,不是解码结束后的完整检查结果。这里保留原文件和输出合同,另做一个七份合成输入的对照实验。

把正文 Review is Monday. 的 base64 中插入一个 !,原检查器仍返回这句文字,tree[0].defects 也是空数组。但新实验在同一个部件调用 get_content() 以后再看 defects,会看到 InvalidBase64CharactersDefect。丢掉最后的 = 也能恢复文字,留下的则是 InvalidBase64PaddingDefect。问题不在于五个原样本的路径失效,而是原先复制的列表看不到后来才登记的证据。

另一个样本的 base64 完全正常,解出的字节却含有不符合声明 UTF-8 的 0xff。原检查器返回 Review on M�nday.,这次即便在解码后查看,MIME defects 仍然为空。两个“空数组”有不同原因,不能统一解释成“正文解码无损”。

MIME 传输解码与字符解码的两层证据;坏 base64 会产生后验 defect,坏 UTF-8 默认替换却不产生 MIME defect
补充图 保留解码前快照、解码后快照和异常类型,才能看出证据出现在哪一步。

13. 字符严格与 MIME 严格,分别拦在哪里

文本内容管理器的约定把 errors 的默认值设为 replace。显式传入 errors='strict',控制的是字节按字符集解成 Unicode 时的错误处理。本例坏 UTF-8 会因此抛出 UnicodeDecodeError;它不会顺便把前一层 base64 的容错改成拒绝。

另一个开关是 policy.default.clone(raise_on_defect=True)。按 MIME policy 的约定,遇到 defect 时抛出异常,而不是照常登记。本例两个坏 base64 在解码期间抛出对应 defect,但它们的解码后 defects 列表反而仍是空的。只保存列表、不保存异常,会再漏掉一次失败。仅开这个开关,也不会阻止本例坏 UTF-8 被默认替换。

所以实验同时保留四个模式:default 是 replace 与默认 policy;charset_strict_only 只将字符错误处理改成 strict;defect_strict_only 只打开 raise_on_defect;both_strict 同时打开。每个模式都从同一份原始 bytes 重新解析,然后只解码一次。不能复用刚才解码过的对象,否则后一次观察会继承此前已经追加的 defect。

这种差别并非新发现的 Python 回归。2018 年 7 月已有开发者在 CPython 的历史问题记录中询问为何 get_payload(decode=True) 不直接拒绝损坏编码。这里用今天实际运行的 EmailMessage 样本重做对照,没有把旧问题当成近期热度或所有输入的行为保证。

14. 严格模式通过,也可能读错了意思

为了防止把补丁写得过头,加入三个负对照。第一份原文真的含有 U+FFFD,内容为 Literal � marker.,用合法 UTF-8 与 base64 编码后,四个模式都正常返回。仅搜索结果中有没有 �,分不清它来自原文还是解码替换,不能拿来充当无损检测。

第二份把 Café at noon. 的 UTF-8 字节故意声明成 ISO-8859-1。四个模式都返回 Café at noon.,没有异常或 defect。这里知道“声明错了”,是因为我们自己生成了 fixture;真实输入没有自动附带这份作者意图。严格解码检查的是声明的编解码器能否接受这些字节,不负责猜发送者本来想写什么。

第三份声明不存在的 x-fixture-unknown,四个模式都得到 LookupError。实验明确保留这个结果,不换一个猜测字符集再把输出标成已经验证。正常 UTF-8 样本则在四个模式下全部得到 Café at noon.,作为正常路径对照。

下面列出全部七份输入、四种模式的固定语义预期。28 格是观察矩阵,不是 28 个独立测试方法。异常格的 decoded_text 都是 null;错误没有被转换成空字符串或另一段正文。完整 JSON 还记录原始字节长度、SHA-256、声明的字符集、Content-Transfer-Encoding 和部件路径。

expected-output.txt

人工固定的语义预期(不是从程序运行结果生成)
模式:D=默认,C=仅字符严格,M=仅 MIME defect 严格,B=双严格
每次观察都从同一份 raw bytes 重新解析;所有 28 次解码前 defects 均为 []。

fixture                        D                 C                 M                 B
valid_utf8                     café              café              café              café
base64_invalid_character       恢复 + 字符defect  恢复 + 字符defect  字符defect异常     字符defect异常
base64_missing_padding         恢复 + padding    恢复 + padding    padding异常       padding异常
invalid_utf8                   替换字符          Unicode异常       替换字符          Unicode异常
literal_replacement_character  原文替换字符      原文替换字符      原文替换字符      原文替换字符
wrong_known_charset            声明下的乱码      声明下的乱码      声明下的乱码      声明下的乱码
unknown_charset                LookupError       LookupError       LookupError       LookupError

café = "Café at noon.";无异常,解码后 defects=[]
恢复 = "Review is Monday.";无异常
字符defect = 解码后 defects=["InvalidBase64CharactersDefect"]
padding = 解码后 defects=["InvalidBase64PaddingDefect"]
字符defect异常 = decode_error_type="InvalidBase64CharactersDefect"
padding异常 = decode_error_type="InvalidBase64PaddingDefect"
Unicode异常 = decode_error_type="UnicodeDecodeError"
以上异常以及 LookupError:decoded_text=null,解码后 defects=[]
替换字符 = "Review on M�nday.";无异常,解码后 defects=[]
原文替换字符 = "Literal � marker.";无异常,解码后 defects=[]
声明下的乱码 = "Café at noon.";无异常,解码后 defects=[]
所有 fixture 的 selected_part_path 均为 "1";这是单部件输入的固定路径。

原 inspect_mime.py 在两份新增输入上的结果:
base64_invalid_character -> tree[0].defects=[];selected_part.content="Review is Monday."
invalid_utf8 -> tree[0].defects=[];selected_part.content="Review on M�nday."
上述两行描述原实现的观察范围,不代表内容已经通过严格验证。

15. 完整复跑:原检查器不动,新增三个文件

已有 Python 3.12 即可运行,无需安装依赖。保留前文的 inspect_mime.py,将下面三个文件保存到同一目录;也可以解压源码 ZIP,其中同时包含原文三个 Python 文件、七份合成 .eml、结果记录和原始 SVG。decode_lab.py 自带完整 fixture 生成逻辑,不需要读取用户邮件。

下载 MIME 解码实验源码与固定样本

python -B decode_lab.py
python -B -m unittest -v test_decode_lab
python -B mutation_checks.py
# 单独复跑原文六个方法:
python -B -m unittest -v test_mime

观察脚本限定输入为自己生成的单部件 text/plain/base64,所以固定路径为 1;它没有重新实现前文的正文选择算法,也不能直接当作任意 MIME 的入口函数。parsebytes 在这七个输入上成功,异常捕获仅围住内容解码;其他结构若在解析期出错,需要独立处理。程序从未读取附件、选择其他 alternative 作为回退,也不执行邮件 HTML。

decode_lab.py

"""Small, generated MIME decoding experiment. No network, email account or model.

This is an observation matrix, not a production mail intake or validator.
Every mode reparses the identical immutable bytes before decoding once.
"""
import base64
from email import policy
from email.parser import BytesParser
import hashlib
import json
import platform


def message(encoded, charset='utf-8'):
    header = ('From: Fixture <fixture@example.invalid>\r\n'
              'Date: Tue, 13 Oct 2026 12:00:00 +0000\r\n'
              'MIME-Version: 1.0\r\n'
              f'Content-Type: text/plain; charset={charset}\r\n'
              'Content-Transfer-Encoding: base64\r\n\r\n')
    return header.encode('ascii') + encoded


GOOD = base64.b64encode(b'Review is Monday.')
CASES = {
    'valid_utf8': message(base64.b64encode('Caf\u00e9 at noon.'.encode('utf-8'))),
    'base64_invalid_character': message(GOOD[:8] + b'!' + GOOD[8:]),
    'base64_missing_padding': message(GOOD.rstrip(b'=')),
    'invalid_utf8': message(base64.b64encode(b'Review on M\xffnday.')),
    'literal_replacement_character': message(base64.b64encode('Literal \ufffd marker.'.encode('utf-8'))),
    'wrong_known_charset': message(base64.b64encode('Caf\u00e9 at noon.'.encode('utf-8')), 'iso-8859-1'),
    'unknown_charset': message(GOOD, 'x-fixture-unknown'),
}

MODES = {
    'default': (policy.default, 'replace'),
    'charset_strict_only': (policy.default, 'strict'),
    'defect_strict_only': (policy.default.clone(raise_on_defect=True), 'replace'),
    'both_strict': (policy.default.clone(raise_on_defect=True), 'strict'),
}


def observe(raw, mode):
    selected_policy, error_handler = MODES[mode]
    part = BytesParser(policy=selected_policy).parsebytes(raw)
    before = [type(d).__name__ for d in part.defects]
    result = {
        'selected_part_path': '1',  # Every fixture is a single text/plain root.
        'defects_before_decode': before,
        'declared_charset': part.get_content_charset(),
        'content_transfer_encoding': str(part['Content-Transfer-Encoding']),
        'decoded_text': None,
        'decode_error_type': None,
    }
    try:
        result['decoded_text'] = part.get_content(errors=error_handler)
    except (UnicodeError, LookupError, ValueError) as exc:
        # email.errors.MessageDefect is a ValueError subclass.
        result['decode_error_type'] = type(exc).__name__
    result['defects_after_decode'] = [type(d).__name__ for d in part.defects]
    return result


def report():
    return {
        'runtime': {'python': platform.python_version(), 'implementation': platform.python_implementation()},
        'scope': 'seven_generated_base64_text_parts_no_external_input_or_authentication',
        'cases': {
            name: {
                'raw_bytes': len(raw),
                'raw_sha256': hashlib.sha256(raw).hexdigest(),
                'modes': {mode: observe(raw, mode) for mode in MODES},
            } for name, raw in CASES.items()
        },
    }


if __name__ == '__main__':
    print(json.dumps(report(), ensure_ascii=False, indent=2))

test_decode_lab.py

"""Hand-authored expectations, including controls against overclaiming strictness."""
import hashlib
import unittest
from decode_lab import CASES, MODES, observe, report
from inspect_mime import inspect as existing_inspect


class DecodeTests(unittest.TestCase):
    def test_valid_utf8_all_modes(self):
        for mode in MODES:
            with self.subTest(mode=mode):
                result = observe(CASES['valid_utf8'], mode)
                self.assertEqual(result['decoded_text'], 'Caf\u00e9 at noon.')
                self.assertIsNone(result['decode_error_type'])
                self.assertEqual(result['defects_before_decode'], [])
                self.assertEqual(result['defects_after_decode'], [])

    def test_base64_bad_character_is_late_defect(self):
        for mode in ('default', 'charset_strict_only'):
            with self.subTest(mode=mode):
                result = observe(CASES['base64_invalid_character'], mode)
                self.assertEqual(result['defects_before_decode'], [])
                self.assertEqual(result['defects_after_decode'], ['InvalidBase64CharactersDefect'])
                self.assertEqual(result['decoded_text'], 'Review is Monday.')
                self.assertIsNone(result['decode_error_type'])

    def test_base64_missing_padding_is_late_defect(self):
        result = observe(CASES['base64_missing_padding'], 'default')
        self.assertEqual(result['defects_before_decode'], [])
        self.assertEqual(result['defects_after_decode'], ['InvalidBase64PaddingDefect'])
        self.assertEqual(result['decoded_text'], 'Review is Monday.')

    def test_raise_on_defect_raises_at_decode_not_parse(self):
        for case, expected in [('base64_invalid_character', 'InvalidBase64CharactersDefect'),
                               ('base64_missing_padding', 'InvalidBase64PaddingDefect')]:
            for mode in ('defect_strict_only', 'both_strict'):
                with self.subTest(case=case, mode=mode):
                    result = observe(CASES[case], mode)
                    self.assertEqual(result['defects_before_decode'], [])
                    self.assertEqual(result['defects_after_decode'], [])
                    self.assertIsNone(result['decoded_text'])
                    self.assertEqual(result['decode_error_type'], expected)

    def test_invalid_utf8_is_independent_of_mime_defects(self):
        for mode in ('default', 'defect_strict_only'):
            with self.subTest(mode=mode):
                result = observe(CASES['invalid_utf8'], mode)
                self.assertEqual(result['decoded_text'], 'Review on M\ufffdnday.')
                self.assertEqual(result['defects_after_decode'], [])
                self.assertIsNone(result['decode_error_type'])
        for mode in ('charset_strict_only', 'both_strict'):
            with self.subTest(mode=mode):
                result = observe(CASES['invalid_utf8'], mode)
                self.assertIsNone(result['decoded_text'])
                self.assertEqual(result['decode_error_type'], 'UnicodeDecodeError')
                self.assertEqual(result['defects_after_decode'], [])

    def test_literal_replacement_is_valid_utf8(self):
        for mode in MODES:
            with self.subTest(mode=mode):
                result = observe(CASES['literal_replacement_character'], mode)
                self.assertEqual(result['decoded_text'], 'Literal \ufffd marker.')
                self.assertIsNone(result['decode_error_type'])
                self.assertEqual(result['defects_after_decode'], [])

    def test_wrong_known_charset_can_pass_both_strict(self):
        result = observe(CASES['wrong_known_charset'], 'both_strict')
        self.assertEqual(result['decoded_text'], 'Caf\u00c3\u00a9 at noon.')
        self.assertNotEqual(result['decoded_text'], 'Caf\u00e9 at noon.')
        self.assertIsNone(result['decode_error_type'])
        self.assertEqual(result['defects_after_decode'], [])

    def test_unknown_charset_is_lookup_error_all_modes(self):
        for mode in MODES:
            with self.subTest(mode=mode):
                result = observe(CASES['unknown_charset'], mode)
                self.assertEqual(result['decode_error_type'], 'LookupError')
                self.assertIsNone(result['decoded_text'])

    def test_current_article_snapshot_misses_late_defect(self):
        result = existing_inspect(CASES['base64_invalid_character'])
        self.assertEqual(result['tree'][0]['defects'], [])
        self.assertEqual(result['selected_part']['content'], 'Review is Monday.')
        self.assertEqual(result['selected_part']['path'], '1')

    def test_current_article_default_replaces_invalid_utf8(self):
        result = existing_inspect(CASES['invalid_utf8'])
        self.assertEqual(result['tree'][0]['defects'], [])
        self.assertEqual(result['selected_part']['content'], 'Review on M\ufffdnday.')

    def test_input_and_results_are_stable(self):
        before = {name: hashlib.sha256(raw).hexdigest() for name, raw in CASES.items()}
        self.assertEqual(report(), report())
        self.assertEqual(before, {name: hashlib.sha256(raw).hexdigest() for name, raw in CASES.items()})


if __name__ == '__main__':
    unittest.main(verbosity=2)

测试预期直接写在断言里,不依赖“把刚刚的实际结果复制成预期”。其中两个方法调用未改动的原检查器,固定它在新增输入上的输出,说明原快照看不到什么;其他方法分别验证两层错误和三个负对照。

mutation_checks.py

"""Run five deliberate regressions in temporary copies; never edit the originals."""
from pathlib import Path
import json
import os
import shutil
import subprocess
import sys
import tempfile

ROOT = Path(__file__).resolve().parent
FILES = ('decode_lab.py', 'inspect_mime.py', 'test_decode_lab.py')
RUNNER = '''import io, json, unittest
import test_decode_lab
suite = unittest.defaultTestLoader.loadTestsFromModule(test_decode_lab)
result = unittest.TextTestRunner(stream=io.StringIO()).run(suite)
print(json.dumps({'tests': result.testsRun,
                  'failures': [test.id() for test, _ in result.failures],
                  'errors': [test.id() for test, _ in result.errors],
                  'ok': result.wasSuccessful()}))
'''
# Each deliberate bug must cause an assertion failure in its named test.
MUTATIONS = (
    ('reuse_pre_decode_snapshot',
     "result['defects_after_decode'] = [type(d).__name__ for d in part.defects]",
     "result['defects_after_decode'] = before",
     'test_base64_bad_character_is_late_defect'),
    ('ignore_charset_strict',
     "part.get_content(errors=error_handler)",
     "part.get_content(errors='replace')",
     'test_invalid_utf8_is_independent_of_mime_defects'),
    ('ignore_defect_strict',
     'BytesParser(policy=selected_policy)',
     'BytesParser(policy=policy.default)',
     'test_raise_on_defect_raises_at_decode_not_parse'),
    ('reject_literal_replacement_character',
     "result['decoded_text'] = part.get_content(errors=error_handler)",
     "result['decoded_text'] = part.get_content(errors=error_handler)\n"
     "        if '\\ufffd' in result['decoded_text']:\n"
     "            raise ValueError('replacement glyph is not proof')",
     'test_literal_replacement_is_valid_utf8'),
    ('hide_decode_exception',
     "result['decode_error_type'] = type(exc).__name__",
     "result['decode_error_type'] = None",
     'test_unknown_charset_is_lookup_error_all_modes'),
)


def run_tests(folder, env=None):
    proc = subprocess.run([sys.executable, '-B', '-c', RUNNER], cwd=folder,
                          text=True, capture_output=True, env=env, check=True)
    return json.loads(proc.stdout)


def main():
    baseline = run_tests(ROOT)
    assert baseline == {'tests': 11, 'failures': [], 'errors': [], 'ok': True}
    original = (ROOT / 'decode_lab.py').read_text()
    results = []
    for name, old, new, must_fail in MUTATIONS:
        assert original.count(old) == 1, (name, 'replacement must be unique')
        with tempfile.TemporaryDirectory() as temp:
            folder = Path(temp)
            for file in FILES:
                shutil.copyfile(ROOT / file, folder / file)
            (folder / 'decode_lab.py').write_text(original.replace(old, new))
            result = run_tests(folder)
        assert result['tests'] == 11 and not result['ok'] and not result['errors'], result
        assert any(must_fail in test for test in result['failures']), result
        results.append({'mutation': name, 'detected': True, **result})
    env = dict(os.environ, TZ='Pacific/Honolulu', PYTHONHASHSEED='321', LC_ALL='C.UTF-8')
    repeat = run_tests(ROOT, env)
    a = subprocess.check_output([sys.executable, '-B', 'decode_lab.py'], cwd=ROOT)
    b = subprocess.check_output([sys.executable, '-B', 'decode_lab.py'], cwd=ROOT, env=env)
    assert repeat == baseline and a == b
    assert (ROOT / 'decode_lab.py').read_text() == original
    print(json.dumps({'baseline': baseline, 'mutations': results,
                      'alternate_environment': repeat, 'same_report_bytes': a == b,
                      'original_unchanged': True}, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()

故障注入脚本只改临时副本,分别制造五个错误:把后验偷换成前验、忽略字符严格开关、忽略 MIME 严格开关、把原文里的 � 一律当失败、吞掉异常类型。每个错误都必须触发指定测试的断言失败;导入失败或其他执行错误不算成功检出。原源码保持不变。

16. 这次验证了什么,交给摘要前还缺什么

本次实际解释器为 CPython 3.12.14。新增 11 个测试方法全部通过,五个故障注入均被断言检出;另外,原文未修改的六个方法也单独重跑通过。将 TZ 改为 Pacific/Honolulu、PYTHONHASHSEED 改为 321 并重新启动进程后,新增测试仍通过,完整 JSON 逐字节相同。这个重复运行不增加样本种类或测试方法数量。

新增测试的实际记录

test_base64_bad_character_is_late_defect (test_decode_lab.DecodeTests.test_base64_bad_character_is_late_defect) ... ok
test_base64_missing_padding_is_late_defect (test_decode_lab.DecodeTests.test_base64_missing_padding_is_late_defect) ... ok
test_current_article_default_replaces_invalid_utf8 (test_decode_lab.DecodeTests.test_current_article_default_replaces_invalid_utf8) ... ok
test_current_article_snapshot_misses_late_defect (test_decode_lab.DecodeTests.test_current_article_snapshot_misses_late_defect) ... ok
test_input_and_results_are_stable (test_decode_lab.DecodeTests.test_input_and_results_are_stable) ... ok
test_invalid_utf8_is_independent_of_mime_defects (test_decode_lab.DecodeTests.test_invalid_utf8_is_independent_of_mime_defects) ... ok
test_literal_replacement_is_valid_utf8 (test_decode_lab.DecodeTests.test_literal_replacement_is_valid_utf8) ... ok
test_raise_on_defect_raises_at_decode_not_parse (test_decode_lab.DecodeTests.test_raise_on_defect_raises_at_decode_not_parse) ... ok
test_unknown_charset_is_lookup_error_all_modes (test_decode_lab.DecodeTests.test_unknown_charset_is_lookup_error_all_modes) ... ok
test_valid_utf8_all_modes (test_decode_lab.DecodeTests.test_valid_utf8_all_modes) ... ok
test_wrong_known_charset_can_pass_both_strict (test_decode_lab.DecodeTests.test_wrong_known_charset_can_pass_both_strict) ... ok

----------------------------------------------------------------------
Ran 11 tests in 0.040s

OK

正文提取副本与源码 ZIP 解压副本也已各自运行,源码与结果一致。2026 年 10 月 13 日查阅的 Python 3.12 文档显示版本 3.12.15、页面更新于 10 月 1 日 15:25 UTC;这是文档版本,不是上述测试的解释器版本。历史问题记录用于解释 API 背景,具体输出仍以这组固定样本的实测为依据。

接回前文时,先按原规则选择外层正文,再为选中的部件保留解码前后证据和异常。原始输入哈希与节点路径继续保留。应用可以明确规定:出现所关心的 defect 或解码异常,就暂不交给摘要,并交由人工检查。这是一条应用规则,不是 MIME 标准自动给出的决策;没有出现这些信号,也不能把结果命名为 valid_email、charset_correct 或 safe。

这个补充只处理七份受控 base64 文本样本的字节到字符串过程。任意头部、quoted-printable、所有字符集、输入资源限制、签名认证、HTML 清洗、真实客户端和模型摘要均未在这里验证。与前文相同,引用仍可能保留,声明也仍可能不可信。把“能得到一个字符串”“符合声明的字符规则”和“准确理解原意”分开,才能让下一步如实使用这些文字。

plain 与 html 能否合并后提高摘要置信度?

alternative 表示替代版本。本文选择一个表示,未证明两版语义完全一致,也不把重复文字当作独立佐证。

有文件名就一定是附件吗?

本例 name 参数可以被 get_filename 读到,但 Content-Disposition 仍为空,正文选择也可能落到该节点。需分别记录这些事实。

get_content 会去掉旧引用或加载图片吗?

本例仍保留引用行、blockquote 和远程地址;代码不渲染 HTML、不请求资源,也没有清洗或划分当前内容与历史内容。

这些检查能直接保护线上邮件入口吗?

不能。代码只检查受控小型输入,没有建立任意不可信邮件所需的获取与解析资源边界。