diff options
| -rw-r--r-- | abusectl/parse.py | 120 | ||||
| -rw-r--r-- | tests/test_parse.py | 61 |
2 files changed, 181 insertions, 0 deletions
diff --git a/abusectl/parse.py b/abusectl/parse.py index 33e1226..6d6789a 100644 --- a/abusectl/parse.py +++ b/abusectl/parse.py @@ -32,16 +32,31 @@ address the sender controls confirms to them that the address is live and can fire a tracker embedded in the DNS or HTTP response. """ +import hashlib import ipaddress import re from dataclasses import dataclass from email import policy from email.parser import BytesParser +from abusectl import redact + _BRACKETED_IP = re.compile(r"\[([0-9a-fA-F:.]+)\]") _ADDR_DOMAIN = re.compile(r"@([A-Za-z0-9.-]+)") _AUTH_VERDICT = re.compile(r"(spf|dkim|dmarc)=(\w+)", re.IGNORECASE) +# Deliberately permissive about where a URL stops: mail wraps URLs across +# lines and butts them up against punctuation, so a strict grammar would +# under-match and silently lose an indicator. Over-matching a trailing +# character is cheap to correct (stripped below); losing the URL entirely +# is not correctable at all. +_URL = re.compile(r"https?://[^\s<>\"')]+", re.IGNORECASE) + +# The nested value inside a redirect_chains() step is attacker-supplied, so +# recursion depth is bounded against a hostile chain designed to blow the +# stack or spin forever, independent of the `seen` cycle guard. +_MAX_REDIRECT_DEPTH = 5 + @dataclass(frozen=True) class Hop: @@ -179,3 +194,108 @@ def auth_results(raw: bytes) -> dict[str, str]: verdicts.setdefault(mechanism.lower(), verdict.lower()) return verdicts + + +@dataclass(frozen=True) +class Attachment: + filename: str + sha256: str + + +def _text_parts(message) -> list[str]: + """Return the decoded text of every non-attachment text part. + + An HTML body and a plain-text body are both `text/*`; an attached + text file is not walked here because its disposition is `attachment`, + not because of its type. + """ + parts = [] + for part in message.walk(): + if part.get_content_maintype() != "text": + continue + if part.get_content_disposition() == "attachment": + continue + try: + parts.append(part.get_content()) + except (LookupError, UnicodeDecodeError): + # An unknown or wrong charset label is not a reason to lose the + # rest of the message; decode what bytes there are. + payload = part.get_payload(decode=True) or b"" + parts.append(payload.decode("utf-8", "replace")) + return parts + + +def urls(raw: bytes) -> list[str]: + """Return every http(s) URL found in the message's text parts, redacted. + + Nothing here is fetched or resolved: URLs are found by matching the + message's own text, never by requesting anything. Each match is + redacted with redact.url() before being returned, deduplicated and + sorted so the result is stable regardless of where in the message a + URL happened to repeat. + """ + message = _message(raw) + found = set() + for text in _text_parts(message): + for match in _URL.findall(text): + found.add(redact.url(match.rstrip(".,;:!?"))) + return sorted(found) + + +def _redirect_targets(url_text: str, depth: int, seen: set[str]) -> list[tuple[str, str]]: + if depth >= _MAX_REDIRECT_DEPTH or url_text in seen: + return [] + seen.add(url_text) + + chains = [] + for target in redact.url_valued_parameters(url_text): + chains.append((redact.url(url_text), redact.url(target))) + chains.extend(_redirect_targets(target, depth + 1, seen)) + return chains + + +def redirect_chains(raw: bytes) -> list[tuple[str, str]]: + """Return redirect hops the message DECLARES, as (from, to) pairs. + + Nothing is followed. A redirector names its own destination in a query + parameter, so the chain is read out of that parameter's decoded value, + never by requesting the URL. The target is reported as an indicator in + its own right, not as a recipient identifier, but it is still passed + through redact.url(): recovering it as an indicator does not license + leaving a recipient token in the destination's own query string. + + Recursion follows a target that is itself a redirector, bounded by + _MAX_REDIRECT_DEPTH and a `seen` set, because the nested value is + attacker-supplied and an unbounded walk over hostile input is a hazard + in itself. + """ + message = _message(raw) + seen: set[str] = set() + chains = [] + for text in _text_parts(message): + for match in _URL.findall(text): + chains.extend(_redirect_targets(match.rstrip(".,;:!?"), 0, seen)) + return chains + + +def attachments(raw: bytes) -> list[Attachment]: + """Return filename and SHA-256 for every attachment part. + + The filename is untrusted text supplied by the sender: it is recorded + for the report, never used to build or open a path. + """ + message = _message(raw) + found = [] + for part in message.walk(): + if part.get_content_disposition() != "attachment": + continue + payload = part.get_payload(decode=True) + if payload is None: + continue + found.append( + Attachment( + filename=part.get_filename() or "", + sha256=hashlib.sha256(payload).hexdigest(), + ) + ) + return found diff --git a/tests/test_parse.py b/tests/test_parse.py index 1083f1d..0fac304 100644 --- a/tests/test_parse.py +++ b/tests/test_parse.py @@ -96,5 +96,66 @@ class TestAuthResults(unittest.TestCase): self.assertEqual(parse.auth_results(load("with-attachment.eml")), {}) +class TestUrls(unittest.TestCase): + def test_an_href_is_found_and_redacted(self): + urls = parse.urls(load("simple.eml")) + self.assertEqual( + urls, + ["http://login.bank-verify.example.invalid/verify?id=REDACTED"], + ) + + def test_a_plain_text_url_is_found_and_redacted(self): + urls = parse.urls(load("forged-chain.eml")) + self.assertEqual(urls, ["http://evil.example.invalid/go?u=REDACTED"]) + + def test_urls_are_deduplicated_and_ordered(self): + raw = ( + b"From: <a@b.example.invalid>\r\n" + b"Subject: t\r\n" + b"Content-Type: text/plain\r\n\r\n" + b"http://z.example.invalid/ and http://a.example.invalid/ and " + b"http://z.example.invalid/ again\r\n" + ) + self.assertEqual( + parse.urls(raw), + ["http://a.example.invalid/", "http://z.example.invalid/"], + ) + + +class TestRedirectChains(unittest.TestCase): + def test_a_declared_target_is_recovered_as_a_hop(self): + chains = parse.redirect_chains(load("redirector.eml")) + self.assertEqual(len(chains), 1) + source, target = chains[0] + self.assertTrue(source.startswith("http://t.example.invalid/c")) + self.assertTrue(target.startswith("http://evil.example.invalid/pay")) + + def test_the_recovered_target_is_itself_redacted(self): + _, target = parse.redirect_chains(load("redirector.eml"))[0] + self.assertEqual(target, "http://evil.example.invalid/pay?ref=REDACTED") + + def test_the_recipient_token_does_not_survive(self): + # The whole point: the destination is an indicator, the token is not. + chains = parse.redirect_chains(load("redirector.eml")) + self.assertNotIn("dGVzdEBleGFtcGxlLm9yZw", repr(chains)) + + def test_a_message_with_no_redirector_reports_none(self): + self.assertEqual(parse.redirect_chains(load("simple.eml")), []) + + +class TestAttachments(unittest.TestCase): + def test_filename_and_sha256_are_recorded(self): + found = parse.attachments(load("with-attachment.eml")) + self.assertEqual(len(found), 1) + self.assertEqual(found[0].filename, "invoice.pdf") + self.assertEqual( + found[0].sha256, + "315f5bdb76d078c43b8ac0064e4a0164612b1fce77c869345bfc94c75894edd3", + ) + + def test_a_message_with_no_attachment_reports_none(self): + self.assertEqual(parse.attachments(load("simple.eml")), []) + + if __name__ == "__main__": unittest.main() |