diff options
| -rw-r--r-- | abusectl/parse.py | 36 | ||||
| -rw-r--r-- | tests/test_parse.py | 22 |
2 files changed, 49 insertions, 9 deletions
diff --git a/abusectl/parse.py b/abusectl/parse.py index 468e35e..da99cc5 100644 --- a/abusectl/parse.py +++ b/abusectl/parse.py @@ -84,18 +84,36 @@ def _message(raw: bytes): def _extract_ip(received_value: str) -> str | None: - # The bracketed literal after the connecting hostname is the only part - # of a Received header the accepting server itself wrote; everything - # else (the claimed hostname) is supplied by the connecting client and - # cannot be trusted. Validate through ipaddress so a stray bracketed - # token that isn't actually an IP is skipped rather than misreported. - for candidate in _BRACKETED_IP.findall(received_value): + # Postfix (and others) write two bracketed literals: "from [HELO-CLAIM] + # (rdns-name [OBSERVED-ADDRESS])". The first is whatever the connecting + # client typed as its own HELO/EHLO argument and is entirely + # attacker-chosen; the second is what the accepting server itself saw + # on the connection. The bracketed literal is the only part of the + # header the accepting server wrote AT ALL, but only when there are + # two of them does "which bracket" become a real question, and the + # answer is the one closest to " by " (the accepting server's own + # clause), which is the LAST one appearing before it. A header with + # only one bracketed IP has no such ambiguity, so that single value is + # used regardless of where " by " falls, or whether it appears at all. + candidates = [] + for match in _BRACKETED_IP.finditer(received_value): try: - ipaddress.ip_address(candidate) + ipaddress.ip_address(match.group(1)) except ValueError: continue - return candidate - return None + candidates.append((match.start(), match.group(1))) + + if not candidates: + return None + if len(candidates) == 1: + return candidates[0][1] + + by_pos = received_value.find(" by ") + if by_pos == -1: + return candidates[0][1] + + before_by = [ip for pos, ip in candidates if pos < by_pos] + return before_by[-1] if before_by else candidates[0][1] def received_hops(raw: bytes) -> list[Hop]: diff --git a/tests/test_parse.py b/tests/test_parse.py index a29b7a1..49fc41e 100644 --- a/tests/test_parse.py +++ b/tests/test_parse.py @@ -63,6 +63,28 @@ class TestReceivedChain(unittest.TestCase): ) self.assertIsNone(ip) + def test_a_helo_literal_does_not_beat_the_observed_address(self): + # Postfix writes the client's own HELO string first and the address + # it actually observed second. The first is attacker-chosen. + raw = ( + b"Received: from [198.51.100.7] (unknown [203.0.113.99])" + b" by mx.example.org with ESMTP id X;" + b" Tue, 8 Sep 2026 10:00:00 +0200\r\n" + b"From: <a@evil.invalid>\r\nSubject: t\r\n\r\nbody\r\n" + ) + self.assertEqual(parse.sending_ip(raw, trusted=["192.0.2.0/24"]), + "203.0.113.99") + + def test_a_single_bracketed_address_still_works(self): + raw = ( + b"Received: from x.invalid (x.invalid [203.0.113.5])" + b" by mx.example.org with ESMTP id Y;" + b" Tue, 8 Sep 2026 10:00:00 +0200\r\n" + b"From: <a@evil.invalid>\r\nSubject: t\r\n\r\nbody\r\n" + ) + self.assertEqual(parse.sending_ip(raw, trusted=["192.0.2.0/24"]), + "203.0.113.5") + class TestSenderDomains(unittest.TestCase): def test_the_three_sender_headers_are_collected(self): |