过滤 AWS 资源访问验证成功通知

aws-email-check/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: aws-email-check
-description: 读取当天新到的 AWS 账号风控邮件，过滤登录验证码、去重并按中文字段顺序汇报已提取的风险与处理信息。适用于定时检查配置的 IMAP 邮箱。
+description: 读取当天新到的 AWS 账号风控邮件，过滤登录验证码和资源访问请求验证成功通知、去重并按中文字段顺序汇报已提取的风险与处理信息。适用于定时检查配置的 IMAP 邮箱。
 ---
 
 # AWS 账号风控邮件检查
@@ -13,6 +13,8 @@ description: 读取当天新到的 AWS 账号风控邮件，过滤登录验证
 
 登录验证码、OTP 和一次性密码邮件不生成提醒，包括转发邮件、中英文及 HTML 邮件；异常活动提示、建议更换密码或启用 MFA 等验证码安全套话不会将其变成风控提醒。明确说明账号受限、暂停、关闭，或要求提交身份核验材料的真实风控通知仍保留，不能仅凭“验证”、verify 或 MFA 字样过滤。过滤邮件仍记录扫描去重，但不保存到报告历史。
 
+“您访问 AWS 资源的请求已得到验证 / Your Request For Accessing AWS Resources Has Been Validated”这类资源访问请求验证成功通知也不生成提醒，支持中英文、转发及 HTML 邮件。同一邮件仍有明确的账号限制、暂停、关闭、访问密钥泄露、资源滥用或核验材料提交要求时保留提醒；不能泛化过滤所有包含 validated 或“验证”的邮件。
+
 输出每封邮件按以下字段顺序排列，多封之间空一行。只展示有可信非空提取结果的字段；账号 ID、警告信息、涉及资源、要求操作、截止时间等任何缺失字段均整行省略，不输出空标签或 `未提取`，不猜测补齐。没有符合条件的新邮件时只输出固定的“无”行，不列出被过滤邮件或过滤详情。不输出任何访问密钥、授权码、密码或令牌。
 
 ```text
@@ -31,4 +33,4 @@ AWS账号ID：...
 今日无新增AWS账号风控邮件。
 ```
 
-当前工作区有 `参考邮件` 目录时，本地样本验证可用 `python scripts/check_mail.py --eml-dir 参考邮件 --date 2026-09-28`；`过滤邮件` 中的验证码样本也可用同一命令指定目录和对应日期验证。本地模式不读取或改写生产去重状态；这两个目录含真实邮件，不纳入版本控制。离线回归测试使用 `python -m unittest discover -s tests -v`，真实样本目录缺失时相应样本测试跳过，合成邮件测试仍完整运行。
+当前工作区有 `参考邮件` 目录时，本地样本验证可用 `python scripts/check_mail.py --eml-dir 参考邮件 --date 2026-09-28`；`过滤邮件` 中的验证码和资源访问请求验证成功样本也可用同一命令指定目录和对应日期验证。本地模式不读取或改写生产去重状态；这两个目录含真实邮件，不纳入版本控制。离线回归测试使用 `python -m unittest discover -s tests -v`，真实样本目录缺失时相应样本测试跳过，合成邮件测试仍完整运行。

aws-email-check/scripts/check_mail.py

@@ -34,6 +34,15 @@ WARNING_TERMS = re.compile(
 EXPLICIT_CLOSURE_SUBJECT = re.compile(
     r"permanent closure confirmation|永久关闭确认", re.I
 )
+ACCESS_VALIDATED = re.compile(
+    r"your\s+request\s+for\s+accessing\s+AWS\s+resources\s+has\s+been\s+"
+    r"(?:successfully\s+)?validated\b|"
+    r"您访问\s*AWS\s*资源的请求已(?:得到验证|通过验证|验证成功)", re.I
+)
+RISK_ALERT_SUBJECT = re.compile(
+    r"suspicious activity|security (?:alert|warning)|abuse report|phishing|compromis|"
+    r"可疑活动|异常活动|滥用|钓鱼|泄露|安全警告", re.I
+)
 SECRET = re.compile(
     r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b|"
     r"\bBearer\s+[A-Za-z0-9._~+/-]+|"
@@ -76,6 +85,15 @@ DOCUMENT_REQUEST = re.compile(
     r"(?:请|需要|必须).{0,30}(?:上传|提交|提供).{0,40}(?:材料|证件|身份证|护照|账单|证明)", re.I
 )
 RISK_EVIDENCE = re.compile(ACCOUNT_RISK.pattern + "|" + DOCUMENT_REQUEST.pattern, re.I)
+RESOURCE_RISK_ACTION = re.compile(
+    r"(?:please|must|required to|take action).{0,80}(?:remove|delete|stop|disable|revoke|rotate)"
+    r".{0,100}(?:phishing|abus(?:e|ive)|malicious|compromised|unauthorized)|"
+    r"(?:请|必须|需要|立即).{0,40}(?:删除|移除|停止|停用|撤销|轮换|更换)"
+    r".{0,60}(?:钓鱼|滥用|恶意|泄露|未授权)", re.I
+)
+ACCESS_NOTICE_RISK_EVIDENCE = re.compile(
+    RISK_EVIDENCE.pattern + "|" + RESOURCE_RISK_ACTION.pattern, re.I
+)
 CONDITIONAL = re.compile(r"\bif\b|如果|假如|倘若|若|如(?:您|你)", re.I)
 NON_RESPONSE = re.compile(
     r"\b(?:do not|don't|fail to|failed to|unless)\b|"
@@ -84,7 +102,7 @@ NON_RESPONSE = re.compile(
 )
 
 
-def has_explicit_risk(text):
+def has_explicit_risk(text, evidence=RISK_EVIDENCE):
     # Evaluate each clause, so a conditional security tip cannot hide an actual
     # restriction elsewhere in the same paragraph (or make a code into an alert).
     for sentence in re.split(r"[\n.!?。！？;；]+|\b(?:but|however)\b|但是|然而", text, flags=re.I):
@@ -94,11 +112,12 @@ def has_explicit_risk(text):
         for clause in re.split(r"[,，]+", sentence):
             if CONDITIONAL.search(clause):
                 conditional_help = not bool(NON_RESPONSE.search(clause))
-            for match in RISK_EVIDENCE.finditer(clause):
+            for match in evidence.finditer(clause):
                 prefix = clause[:match.start()]
                 if CONDITIONAL.search(prefix) and not NON_RESPONSE.search(prefix):
                     continue
-                if conditional_help and DOCUMENT_REQUEST.match(match.group()):
+                if conditional_help and (DOCUMENT_REQUEST.match(match.group())
+                                         or RESOURCE_RISK_ACTION.match(match.group())):
                     continue
                 return True
     return False
@@ -114,6 +133,17 @@ def is_verification_code(subject, body):
     )
 
 
+def is_access_validated_notice(subject, body):
+    """Ignore completed resource-access verification, while keeping risk notices."""
+    combined = subject + "\n" + body
+    return bool(
+        ACCESS_VALIDATED.search(combined)
+        and not (has_explicit_risk(combined, ACCESS_NOTICE_RISK_EVIDENCE)
+                 or EXPLICIT_CLOSURE_SUBJECT.search(subject)
+                 or RISK_ALERT_SUBJECT.search(subject))
+    )
+
+
 class PlainHTML(HTMLParser):
     def __init__(self):
         super().__init__()
@@ -238,7 +268,7 @@ def extract(message, received):
     subject = header_value(message, "Subject")
     body = body_text(message)
     combined = subject + "\n" + body
-    if is_verification_code(subject, body):
+    if is_verification_code(subject, body) or is_access_validated_notice(subject, body):
         return None
     if not RISK_TERMS.search(combined) or not (
         WARNING_TERMS.search(combined) or EXPLICIT_CLOSURE_SUBJECT.search(subject)

aws-email-check/tests/test_extraction.py

@@ -83,8 +83,91 @@ class ExtractionTests(unittest.TestCase):
         )
         self.assertEqual(record["type"], "账户验证或暂停")
 
-    def test_local_forwarded_login_code_is_filtered(self):
-        self.assertIsNone(sample("*.eml", "过滤邮件"))
+    def test_local_forwarded_login_code_is_filtered(self):
+        self.assertIsNone(sample("*FW__*.eml", "过滤邮件"))
+
+    def test_local_forwarded_access_validated_notice_is_filtered(self):
+        self.assertIsNone(sample("*Resources_Has_Been_Validated.eml", "过滤邮件"))
+
+    def test_access_validated_notices_are_filtered(self):
+        cases = [
+            ("您访问 AWS 资源的请求已得到验证", "您的 AWS 资源访问请求已完成核验，现在可以访问资源。", False),
+            ("AWS: Your Request For Accessing AWS Resources Has Been Validated",
+             "The request is complete. You may access AWS resources.", False),
+            ("Fw: 您访问 AWS 资源的请求已得到验证 / "
+             "Your Request For Accessing AWS Resources Has Been Validated",
+             "AWS 资源访问现已恢复。如果您有任何问题，请联系支持。", False),
+            ("Fwd: AWS notification", "Forwarded message:\nSubject: Your Request For Accessing "
+             "AWS Resources Has Been Validated\nYou may access resources again.", False),
+            ("FW: AWS notification", "<div>Your Request For Accessing</div><p>AWS Resources "
+             "Has Been Successfully Validated</p><p>You may access AWS resources.</p>", True),
+            ("Fw: AWS 通知", "<p>您访问 AWS 资源的请求已得到验证</p><p>现在可以访问资源。</p>", True),
+        ]
+        for subject, body, rich in cases:
+            with self.subTest(subject=subject, rich=rich):
+                self.assertIsNone(synthetic(subject, body, rich))
+
+    def test_success_notice_does_not_hide_actual_risk_or_required_documents(self):
+        success = "Your Request For Accessing AWS Resources Has Been Validated"
+        risks = [
+            "Your AWS account has been suspended.",
+            "Your AWS account is restricted.",
+            "Your AWS account is disabled.",
+            "Your AWS account has been terminated.",
+            "Your AWS account is closed.",
+            "Your access key has been compromised.",
+            "Your AWS resource has been implicated in hosting a phishing website.",
+            "We detected potentially unwanted activity in your AWS account.",
+            "Please upload your identity documents.",
+            "If you fail to respond, your account will be suspended.",
+            "您的 AWS 账户已受限。",
+            "您的账户已暂停。请提交身份证明材料。",
+        ]
+        for body in risks:
+            with self.subTest(body=body):
+                self.assertIsNotNone(synthetic(success, body))
+        for subject in ("AWS Abuse Report", "AWS Suspicious Activity", "AWS Account Permanent Closure Confirmation"):
+            with self.subTest(subject=subject):
+                self.assertIsNotNone(synthetic(subject, success))
+
+    def test_other_validation_notices_remain_reported(self):
+        cases = [
+            ("AWS Identity Validated", "Contact support regarding your identity verification."),
+            ("AWS 验证您的身份", "请上传身份证核验材料。"),
+            ("Your Request For Accessing AWS Resources Has Not Been Validated",
+             "We could not validate important details. Please upload proof of identity."),
+            ("AWS notification", "Your Request For Accessing AWS Resources Has Been Validated? "
+             "Your account has been placed on hold."),
+        ]
+        for subject, body in cases:
+            with self.subTest(subject=subject):
+                self.assertIsNotNone(synthetic(subject, body))
+
+    def test_access_success_keeps_explicit_resource_risk_actions(self):
+        subject = "Your Request For Accessing AWS Resources Has Been Validated"
+        cases = [
+            "Please take action immediately to remove phishing content from your AWS resource.",
+            "Please delete malicious content hosted on your AWS resource.",
+            "You must stop abusive traffic from your AWS resource.",
+            "请立即删除您的 AWS 资源上的钓鱼内容。",
+            "必须停止 AWS 资源上的滥用行为。",
+        ]
+        for body in cases:
+            with self.subTest(body=body):
+                self.assertIsNotNone(synthetic(subject, body))
+
+    def test_access_success_security_tips_are_still_filtered(self):
+        subject = "Your Request For Accessing AWS Resources Has Been Validated"
+        cases = [
+            "If you find phishing content, please remove phishing content from your AWS resource.",
+            "If you have questions, please take action to contact AWS support.",
+            "如果您发现钓鱼内容，请立即删除 AWS 资源上的钓鱼内容。",
+            "如果您有任何问题，请联系支持团队。",
+            "Please contact AWS support if you need help accessing resources.",
+        ]
+        for body in cases:
+            with self.subTest(body=body):
+                self.assertIsNone(synthetic(subject, body))
 
     def test_login_codes_with_security_boilerplate_are_filtered(self):
         cases = [

aws-email-check/tests/test_replay.py

@@ -204,6 +204,38 @@ class ReplayTests(unittest.TestCase):
         self.assertEqual(after["report_history"], before["report_history"])
         self.assertEqual(self.run_main("--replay-last"), first)
 
+    def test_filtered_access_validated_notice_is_deduplicated_without_history(self):
+        mail = EmailMessage()
+        mail["Subject"] = "Fw: Your Request For Accessing AWS Resources Has Been Validated"
+        mail["Message-ID"] = "<validated-42@example.com>"
+        mail.set_content("Your AWS resource-access request has completed verification.")
+        fake = FakeIMAP(mail.as_bytes())
+        with patch.object(check_mail.imaplib, "IMAP4_SSL", return_value=fake):
+            self.assertEqual(self.run_main(), check_mail.NO_NEW + "\n")
+            self.assertEqual(self.run_main(), check_mail.NO_NEW + "\n")
+        self.assertEqual(fake.fetches, 1)
+        state = json.loads(self.state.read_text(encoding="utf-8"))
+        entry = next(iter(state["mailboxes"].values()))
+        self.assertEqual(entry["uids"], ["42"])
+        self.assertEqual(entry["message_ids"], ["<validated-42@example.com>"])
+        self.assertNotIn("report_history", state)
+        self.assertNotIn("last_report", state)
+
+    def test_filtered_access_validated_notice_preserves_previous_report(self):
+        with patch.object(check_mail.imaplib, "IMAP4_SSL", return_value=self.fake):
+            first = self.run_main()
+        before = json.loads(self.state.read_text(encoding="utf-8"))
+        mail = EmailMessage()
+        mail["Subject"] = "Fw: 您访问 AWS 资源的请求已得到验证"
+        mail["Message-ID"] = "<validated-43@example.com>"
+        mail.set_content("<p>现在可以访问 AWS 资源。</p>", subtype="html")
+        with patch.object(check_mail.imaplib, "IMAP4_SSL", return_value=FakeIMAP(mail.as_bytes(), uid=b"43")):
+            self.assertEqual(self.run_main(), check_mail.NO_NEW + "\n")
+        after = json.loads(self.state.read_text(encoding="utf-8"))
+        self.assertEqual(after["report_history"], before["report_history"])
+        self.assertEqual(after["last_report"], before["last_report"])
+        self.assertEqual(self.run_main("--replay-last"), first)
+
 
 if __name__ == "__main__":
     unittest.main()