diff --git a/CHANGELOG.md b/CHANGELOG.md index 059c477..e3af5f3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## Unreleased + +### Changes + +- Removed dead code found while extending test coverage: the unused `_SMTPTLSReportDoc.add_policy()` helpers in the Elasticsearch and OpenSearch outputs (the save paths construct policy documents directly), a no-op `failure_indexes` loop in both `migrate_indexes()` implementations (the parameter is still accepted; no failure-index migrations are currently needed), an unreachable `importlib.resources` ImportError fallback in `parsedmarc.utils` (it re-imported the same module, and `importlib.resources.files` always exists on the supported Python ≥3.10), and an unreachable "Invalid report content" guard in `extract_report()` (every input branch either assigns the file object or raises first, confirmed by pyright narrowing). + +### Bug fixes + +- **`parse_email()` no longer crashes with `KeyError: 'Headers'` on messages whose `From` header is present but empty/unparseable** (e.g. a bare `From:` line). mailparser omits `"from"` from `mail_json` for such messages, and the fallback read `parsed_email["Headers"]` — a key that is never set; the parsed headers are stored under lowercase `"headers"` (see the assignment at the top of `parse_email()`). The fallback now reads the correct key and treats an empty parsed header list the same as a missing header, yielding `from=None`. +- **A failed IPinfo API token probe no longer logs "IPinfo API configured".** `configure_ipinfo_api(..., probe=True)` documents that non-fatal probe errors are logged as warnings with the token still accepted, but `_ipinfo_api_lookup()` returns `None` on network errors instead of raising, so the probe's exception handler never fired and a probe that couldn't reach the API logged the success message. The probe now checks the lookup result and logs a warning when verification failed. Invalid tokens (401/403) still raise `InvalidIPinfoAPIKey`. + ## 10.2.1 ### Changes diff --git a/parsedmarc/__init__.py b/parsedmarc/__init__.py index cfbfdd9..2607c51 100644 --- a/parsedmarc/__init__.py +++ b/parsedmarc/__init__.py @@ -757,7 +757,7 @@ def parsed_smtp_tls_reports_to_csv( def parse_aggregate_report_xml( - xml: str, + xml: str | bytes, *, ip_db_path: str | None = None, always_use_local_files: bool = False, @@ -773,7 +773,8 @@ def parse_aggregate_report_xml( """Parses a DMARC XML report string and returns a consistent dict Args: - xml (str): A string of DMARC aggregate report XML + xml (str | bytes): DMARC aggregate report XML (bytes are decoded + with errors ignored) ip_db_path (str): Path to a MMDB file from IPinfo, MaxMind, or DBIP always_use_local_files (bool): Do not download files reverse_dns_map_path (str): Path to a reverse DNS map file @@ -1113,9 +1114,6 @@ def extract_report(content: bytes | str | BinaryIO) -> str: remainder = stream.read() file_object = BytesIO(header + bytes(remainder)) - if file_object is None: - raise ParserError("Invalid report content") - if header[: len(MAGIC_ZIP)] == MAGIC_ZIP: _zip = zipfile.ZipFile(file_object) report = _zip.open(_zip.namelist()[0]).read().decode(errors="ignore") diff --git a/parsedmarc/elastic.py b/parsedmarc/elastic.py index 2184bb8..773921a 100644 --- a/parsedmarc/elastic.py +++ b/parsedmarc/elastic.py @@ -290,27 +290,6 @@ class _SMTPTLSReportDoc(Document): report_id = Text() policies = Nested(_SMTPTLSPolicyDoc) - def add_policy( - self, - policy_type: str, - policy_domain: str, - successful_session_count: int, - failed_session_count: int, - *, - policy_string: str | None = None, - mx_host_patterns: list[str] | None = None, - failure_details: str | None = None, - ): - self.policies.append( - policy_type=policy_type, - policy_domain=policy_domain, - successful_session_count=successful_session_count, - failed_session_count=failed_session_count, - policy_string=policy_string, - mx_host_patterns=mx_host_patterns, - failure_details=failure_details, - ) # pyright: ignore[reportCallIssue] - class AlreadySaved(ValueError): """Raised when a report to be saved matches an existing report""" @@ -409,12 +388,12 @@ def migrate_indexes( Args: aggregate_indexes (list): A list of aggregate index names failure_indexes (list): A list of failure index names + (accepted for API compatibility; no migrations are + currently needed for failure indexes) """ version = 2 if aggregate_indexes is None: aggregate_indexes = [] - if failure_indexes is None: - failure_indexes = [] for aggregate_index_name in aggregate_indexes: if not Index(aggregate_index_name).exists(): continue @@ -444,9 +423,6 @@ def migrate_indexes( reindex(connections.get_connection(), aggregate_index_name, new_index_name) # pyright: ignore[reportArgumentType] Index(aggregate_index_name).delete() - for failure_index in failure_indexes: - pass - def save_aggregate_report_to_elasticsearch( aggregate_report: dict[str, Any], diff --git a/parsedmarc/opensearch.py b/parsedmarc/opensearch.py index d99f163..8244271 100644 --- a/parsedmarc/opensearch.py +++ b/parsedmarc/opensearch.py @@ -281,27 +281,6 @@ class _SMTPTLSReportDoc(Document): report_id = Text() policies = Nested(_SMTPTLSPolicyDoc) - def add_policy( - self, - policy_type: str, - policy_domain: str, - successful_session_count: int, - failed_session_count: int, - *, - policy_string: str | None = None, - mx_host_patterns: list[str] | None = None, - failure_details: str | None = None, - ): - self.policies.append( - policy_type=policy_type, - policy_domain=policy_domain, - successful_session_count=successful_session_count, - failed_session_count=failed_session_count, - policy_string=policy_string, - mx_host_patterns=mx_host_patterns, - failure_details=failure_details, - ) - class AlreadySaved(ValueError): """Raised when a report to be saved matches an existing report""" @@ -409,12 +388,12 @@ def migrate_indexes( Args: aggregate_indexes (list): A list of aggregate index names failure_indexes (list): A list of failure index names + (accepted for API compatibility; no migrations are + currently needed for failure indexes) """ version = 2 if aggregate_indexes is None: aggregate_indexes = [] - if failure_indexes is None: - failure_indexes = [] for aggregate_index_name in aggregate_indexes: if not Index(aggregate_index_name).exists(): continue @@ -444,9 +423,6 @@ def migrate_indexes( reindex(connections.get_connection(), aggregate_index_name, new_index_name) Index(aggregate_index_name).delete() - for failure_index in failure_indexes: - pass - def save_aggregate_report_to_opensearch( aggregate_report: dict[str, Any], diff --git a/parsedmarc/utils.py b/parsedmarc/utils.py index 2255187..20481d2 100644 --- a/parsedmarc/utils.py +++ b/parsedmarc/utils.py @@ -22,11 +22,7 @@ from typing import TypedDict, cast import mailparser from expiringdict import ExpiringDict -try: - from importlib.resources import files -except ImportError: - # Try backported to PY<3 `importlib_resources` - from importlib.resources import files +from importlib.resources import files import dns.exception @@ -527,12 +523,11 @@ def configure_ipinfo_api( if not _IPINFO_API_TOKEN or not probe: return - try: - _ipinfo_api_lookup("1.1.1.1") - except InvalidIPinfoAPIKey: - raise - except Exception as e: - logger.warning(f"IPinfo API probe failed (will fall back per-request): {e}") + # _ipinfo_api_lookup() raises InvalidIPinfoAPIKey on 401/403 (which + # must propagate) and returns None on any other failure — network + # errors, non-2xx responses, malformed bodies. + if _ipinfo_api_lookup("1.1.1.1") is None: + logger.warning("IPinfo API probe failed (will fall back per-request)") else: logger.info("IPinfo API configured") @@ -1158,10 +1153,11 @@ def parse_email(data: bytes | str, *, strip_attachment_payloads: bool = False) - received["date_utc"] = received["date_utc"].replace("T", " ") if "from" not in parsed_email: - if "From" in parsed_email["headers"]: - parsed_email["from"] = parsed_email["Headers"]["From"] - else: - parsed_email["from"] = None + # mailparser omits "from" from mail_json when the From header is + # present but unparseable (e.g. an empty "From:"); headers_json may + # still carry a "From" entry, which can be an empty list — treat + # that the same as a missing header. + parsed_email["from"] = parsed_email["headers"].get("From") or None if parsed_email["from"] is not None: parsed_email["from"] = parse_email_address(parsed_email["from"][0]) diff --git a/tests/test_elastic.py b/tests/test_elastic.py index f897cbb..0ddbd3d 100644 --- a/tests/test_elastic.py +++ b/tests/test_elastic.py @@ -579,6 +579,32 @@ class TestSaveAggregateReport(unittest.TestCase): self.assertIn("cust_dmarc_aggregate_tenant_a*", search_index) +class TestAggregateDocPassedDmarc(unittest.TestCase): + """The _AggregateReportDoc.save() override derives passed_dmarc — the + field dashboards filter on for DMARC pass/fail — from SPF/DKIM + alignment. The SDK parent (elasticsearch_dsl.Document.save) is mocked so + no cluster is needed.""" + + def test_passed_dmarc_derived_from_alignment(self): + cases = [ + (True, False, True), + (False, True, True), + (True, True, True), + (False, False, False), + ] + for spf_aligned, dkim_aligned, expected in cases: + with self.subTest(spf=spf_aligned, dkim=dkim_aligned): + with patch.object( + elastic_module.Document, "save", return_value=None + ) as mock_super_save: + doc = elastic_module._AggregateReportDoc( + spf_aligned=spf_aligned, dkim_aligned=dkim_aligned + ) + doc.save() + mock_super_save.assert_called_once() + self.assertEqual(bool(doc.passed_dmarc), expected) + + # --------------------------------------------------------------------------- # save_failure_report_to_elasticsearch # --------------------------------------------------------------------------- diff --git a/tests/test_init.py b/tests/test_init.py index be6559a..6ee8d0b 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -2954,5 +2954,165 @@ class TestAppendCsv(unittest.TestCase): os.remove(path) +def _minimal_aggregate_xml( + policy_published: str = ( + "example.com

none

" + ), + org_name: str = "TestOrg", + email: str = "test@example.com", + reason: str = "", +) -> str: + """A minimal, valid aggregate report with substitutable sections.""" + return f""" + + + {org_name} + {email} + edge-case + 17040672001704153599 + + {policy_published} + + + 192.0.2.1 + 1 + + none + pass + pass + {reason} + + + example.com + + example.compass + + + """ + + +class TestAggregateReportEdgeCases(unittest.TestCase): + """Parsing edge cases for aggregate report XML documents.""" + + def testBytesInputIsDecoded(self): + """parse_aggregate_report_xml accepts bytes input""" + xml = _minimal_aggregate_xml().encode("utf-8") + report = parsedmarc.parse_aggregate_report_xml(xml, offline=True) + self.assertEqual(report["report_metadata"]["report_id"], "edge-case") + + def testPolicyPublishedListUsesFirstEntry(self): + """When a reporter emits multiple policy_published elements, the + first one is used""" + policies = ( + "example.com

reject

" + "" + "other.example

none

" + "" + ) + report = parsedmarc.parse_aggregate_report_xml( + _minimal_aggregate_xml(policy_published=policies), offline=True + ) + self.assertEqual(report["policy_published"]["domain"], "example.com") + self.assertEqual(report["policy_published"]["p"], "reject") + + def testUnknownPolicyOverrideTypeWarnsUnderRFC9990(self): + """An override reason type that RFC 9990 does not define (and RFC + 7489 never defined) logs an 'Unknown policy override reason type' + warning; it is stored as-is. RFC 9990's PolicyOverrideType + enumeration is {local_policy, mailing_list, other, + policy_test_mode, trusted_forwarder}.""" + policies = ( + "example.com

none

" + "none" + ) + reason = "banana" + with self.assertLogs("parsedmarc.log", level="WARNING") as cm: + report = parsedmarc.parse_aggregate_report_xml( + _minimal_aggregate_xml(policy_published=policies, reason=reason), + offline=True, + ) + self.assertTrue( + any( + "Unknown policy override reason type" in message + for message in cm.output + ) + ) + reasons = report["records"][0]["policy_evaluated"]["policy_override_reasons"] + self.assertEqual(reasons[0]["type"], "banana") + + def testMissingOrgNameAndEmailIsInvalid(self): + """A report with empty org_name and email raises + InvalidAggregateReport, since org_name has no fallback source""" + with self.assertRaises(parsedmarc.InvalidAggregateReport) as ctx: + parsedmarc.parse_aggregate_report_xml( + _minimal_aggregate_xml(org_name="", email=""), offline=True + ) + self.assertIn("Organization name is missing", str(ctx.exception)) + + def testMalformedEmailAttributeOnlyIsDiscarded(self): + """An element that xmltodict turns into an attributes-only + dict (no text) is discarded rather than crashing""" + xml = _minimal_aggregate_xml().replace( + "test@example.com", '' + ) + report = parsedmarc.parse_aggregate_report_xml(xml, offline=True) + self.assertIsNone(report["report_metadata"]["org_email"]) + + +class _NonSeekableStream: + """A minimal non-seekable stream, like sys.stdin / a socket file.""" + + def __init__(self, data): + self._data = data + self._pos = 0 + + def seekable(self): + return False + + def read(self, size=-1): + if size < 0: + result = self._data[self._pos :] + self._pos = len(self._data) + else: + result = self._data[self._pos : self._pos + size] + self._pos += size + return result + + +class _BrokenSeekableStream(_NonSeekableStream): + """A stream whose seekable() itself raises, as some wrapped streams do.""" + + def seekable(self): + raise OSError("stream does not support seekable()") + + +class TestExtractReportStreams(unittest.TestCase): + """extract_report accepts file objects that cannot seek (stdin, pipes, + sockets) and must reject text-mode streams with a clear error.""" + + def testNonSeekableTextStreamRaisesParserError(self): + """A non-seekable text-mode stream raises ParserError instead of + failing later on a bytes/str mismatch""" + with open("samples/extract_report/nice-input.xml") as f: + text = f.read() + with self.assertRaises(parsedmarc.ParserError) as ctx: + parsedmarc.extract_report(cast(BinaryIO, _NonSeekableStream(text))) + self.assertIn("binary", str(ctx.exception)) + + def testNonSeekableBytesStreamIsExtracted(self): + """A non-seekable binary stream is buffered and extracted""" + with open("samples/extract_report/nice-input.xml", "rb") as f: + data = f.read() + result = parsedmarc.extract_report(cast(BinaryIO, _NonSeekableStream(data))) + self.assertIn("", result) + + def testStreamWithBrokenSeekableIsExtracted(self): + """A stream whose seekable() raises is treated as non-seekable""" + with open("samples/extract_report/nice-input.xml", "rb") as f: + data = f.read() + result = parsedmarc.extract_report(cast(BinaryIO, _BrokenSeekableStream(data))) + self.assertIn("", result) + + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/tests/test_opensearch.py b/tests/test_opensearch.py index ef99212..8b05993 100644 --- a/tests/test_opensearch.py +++ b/tests/test_opensearch.py @@ -579,6 +579,32 @@ class TestSaveAggregateReport(unittest.TestCase): self.assertIn("cust_dmarc_aggregate_tenant_a*", search_index) +class TestAggregateDocPassedDmarc(unittest.TestCase): + """The _AggregateReportDoc.save() override derives passed_dmarc — the + field dashboards filter on for DMARC pass/fail — from SPF/DKIM + alignment. The SDK parent (opensearchpy.Document.save) is mocked so + no cluster is needed.""" + + def test_passed_dmarc_derived_from_alignment(self): + cases = [ + (True, False, True), + (False, True, True), + (True, True, True), + (False, False, False), + ] + for spf_aligned, dkim_aligned, expected in cases: + with self.subTest(spf=spf_aligned, dkim=dkim_aligned): + with patch.object( + opensearch_module.Document, "save", return_value=None + ) as mock_super_save: + doc = opensearch_module._AggregateReportDoc( + spf_aligned=spf_aligned, dkim_aligned=dkim_aligned + ) + doc.save() + mock_super_save.assert_called_once() + self.assertEqual(bool(doc.passed_dmarc), expected) + + # --------------------------------------------------------------------------- # save_failure_report_to_opensearch # --------------------------------------------------------------------------- diff --git a/tests/test_utils.py b/tests/test_utils.py index 4bcc6c8..dfcbcae 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -11,6 +11,7 @@ from tempfile import NamedTemporaryFile from unittest.mock import MagicMock, patch import dns.exception +import dns.resolver import requests from expiringdict import ExpiringDict @@ -866,13 +867,82 @@ Body""" for att in result["attachments"]: self.assertNotIn("payload", att) + def testEmptyFromHeaderYieldsNone(self): + """An email whose From header is present but empty parses with + from=None instead of crashing. + + Regression: mailparser omits "from" from mail_json when the From + header value is unparseable, and the headers fallback read + ``parsed_email["Headers"]`` — a key that is never set (the parsed + headers are stored under lowercase "headers", see parse_email) — + so any such message raised KeyError: 'Headers'. + """ + email_str = "From:\r\nTo: a@b.com\r\nSubject: t\r\n\r\nbody\r\n" + result = parsedmarc.utils.parse_email(email_str) + self.assertIsNone(result["from"]) + + def testCcAndBccHeadersAreParsed(self): + """Cc and Bcc headers are parsed into address dicts""" + email_str = ( + "From: a@b.com\r\n" + "To: t@e.com\r\n" + "Cc: c@d.com, C Two \r\n" + "Bcc: e@f.com\r\n" + "Subject: Hi\r\n\r\nBody\r\n" + ) + result = parsedmarc.utils.parse_email(email_str) + self.assertEqual([a["address"] for a in result["cc"]], ["c@d.com", "c2@d.com"]) + self.assertEqual(result["cc"][1]["display_name"], "C Two") + self.assertEqual([a["address"] for a in result["bcc"]], ["e@f.com"]) + + @staticmethod + def _multipart_email(transfer_encoding: str, payload: str) -> str: + return ( + "From: a@b.com\r\nTo: t@e.com\r\nSubject: att\r\n" + "MIME-Version: 1.0\r\n" + 'Content-Type: multipart/mixed; boundary="B"\r\n\r\n' + "--B\r\nContent-Type: text/plain\r\n\r\nbody\r\n" + '--B\r\nContent-Type: application/octet-stream; name="a.bin"\r\n' + f"Content-Transfer-Encoding: {transfer_encoding}\r\n" + 'Content-Disposition: attachment; filename="a.bin"\r\n\r\n' + f"{payload}\r\n" + "--B--\r\n" + ) + + def testNonBase64AttachmentIsHashed(self): + """A non-base64 attachment's sha256 is computed over the encoded + payload text""" + import hashlib + + result = parsedmarc.utils.parse_email( + self._multipart_email("7bit", "hello world") + ) + attachments = result["attachments"] + self.assertEqual(len(attachments), 1) + self.assertEqual( + attachments[0]["sha256"], hashlib.sha256(b"hello world").hexdigest() + ) + + def testUndecodableAttachmentIsKeptWithoutHash(self): + """An attachment whose base64 payload cannot be decoded is kept, + just without a sha256, and parsing does not crash""" + result = parsedmarc.utils.parse_email( + self._multipart_email("base64", "!!!notb64!!!") + ) + attachments = result["attachments"] + self.assertEqual(len(attachments), 1) + self.assertNotIn("sha256", attachments[0]) + self.assertEqual(attachments[0]["payload"], "!!!notb64!!!") + class TestUtilsOutlookMsg(unittest.TestCase): """Tests for Outlook MSG detection and conversion""" + MSG_MAGIC = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + def testIsOutlookMsg(self): """is_outlook_msg detects MSG magic bytes""" - msg_magic = b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + b"\x00" * 100 + msg_magic = self.MSG_MAGIC + b"\x00" * 100 self.assertTrue(parsedmarc.utils.is_outlook_msg(msg_magic)) def testIsNotOutlookMsg(self): @@ -885,6 +955,58 @@ class TestUtilsOutlookMsg(unittest.TestCase): with self.assertRaises(ValueError): parsedmarc.utils.convert_outlook_msg(b"not an msg file") + def testConvertOutlookMsgMissingUtility(self): + """A missing msgconvert utility raises EmailParserError, and the + working directory is restored""" + old_cwd = os.getcwd() + with patch( + "parsedmarc.utils.subprocess.check_call", + side_effect=FileNotFoundError("msgconvert"), + ): + with self.assertRaises(parsedmarc.utils.EmailParserError): + parsedmarc.utils.convert_outlook_msg(self.MSG_MAGIC + b"\x00" * 100) + self.assertEqual(os.getcwd(), old_cwd) + + def testConvertOutlookMsgReadsConvertedFile(self): + """convert_outlook_msg writes the .msg for msgconvert, reads back + the .eml it produces, and restores the working directory. The + subprocess boundary is mocked with a fake msgconvert that converts + the temp .msg into a fixed RFC 822 message.""" + rfc822 = b"From: a@b.com\r\nSubject: converted\r\n\r\nhi\r\n" + + def fake_msgconvert(args, stdout=None, stderr=None): + # msgconvert is invoked in a temp dir containing sample.msg + # and writes sample.eml next to it. + self.assertEqual(args, ["msgconvert", "sample.msg"]) + with open("sample.msg", "rb") as f: + self.assertTrue(parsedmarc.utils.is_outlook_msg(f.read())) + with open("sample.eml", "wb") as f: + f.write(rfc822) + + old_cwd = os.getcwd() + with patch( + "parsedmarc.utils.subprocess.check_call", side_effect=fake_msgconvert + ): + result = parsedmarc.utils.convert_outlook_msg( + self.MSG_MAGIC + b"\x00" * 100 + ) + self.assertEqual(result, rfc822) + self.assertEqual(os.getcwd(), old_cwd) + + def testParseEmailConvertsOutlookMsgBytes(self): + """parse_email detects Outlook MSG bytes and parses the converted + RFC 822 output""" + + def fake_msgconvert(args, stdout=None, stderr=None): + with open("sample.eml", "wb") as f: + f.write(b"From: a@b.com\r\nSubject: from msg\r\n\r\nhi\r\n") + + with patch( + "parsedmarc.utils.subprocess.check_call", side_effect=fake_msgconvert + ): + result = parsedmarc.utils.parse_email(self.MSG_MAGIC + b"\x00" * 100) + self.assertEqual(result["subject"], "from msg") + class TestUtilsReverseDnsMap(unittest.TestCase): """Tests for reverse DNS map loading""" @@ -921,6 +1043,33 @@ class TestUtilsReverseDnsMap(unittest.TestCase): parsedmarc.utils.load_reverse_dns_map(rdns_map) self.assertTrue(len(rdns_map) > 0) + def testLoadReverseDnsMapInvalidCsvFallback(self): + """A fetch that returns a non-map CSV body logs a warning and + falls back to the bundled map""" + response = MagicMock() + response.text = "not,the,map\nfoo,bar,baz\n" + response.raise_for_status.return_value = None + rdns_map = {} + with patch("parsedmarc.utils.requests.get", return_value=response): + with self.assertLogs("parsedmarc.log", level="WARNING") as cm: + parsedmarc.utils.load_reverse_dns_map(rdns_map) + self.assertTrue(any("Not a valid CSV file" in message for message in cm.output)) + self.assertGreater(len(rdns_map), 0) + + def testGetServiceUsesProvidedMap(self): + """get_service_from_reverse_dns_base_domain consults a caller- + provided non-empty map without loading anything""" + provided: parsedmarc.utils.ReverseDNSMap = { + "custom.example": {"name": "Custom Co", "type": "SaaS"} + } + with patch("parsedmarc.utils.load_reverse_dns_map") as mock_load: + service = parsedmarc.utils.get_service_from_reverse_dns_base_domain( + "Custom.Example", reverse_dns_map=provided + ) + mock_load.assert_not_called() + self.assertEqual(service["name"], "Custom Co") + self.assertEqual(service["type"], "SaaS") + class TestPslOverrides(unittest.TestCase): """Tests for PSL override matching""" @@ -961,5 +1110,262 @@ class TestIsMbox(unittest.TestCase): self.assertFalse(parsedmarc.utils.is_mbox("/nonexistent/file.mbox")) +class TestQueryDnsRetries(unittest.TestCase): + """Tests for the query_dns transient-error retry loop, mocking at the + dnspython SDK boundary (Resolver.resolve).""" + + def testTransientErrorIsRetried(self): + """A retryable error (OSError is in _RETRYABLE_DNS_ERRORS) on the + first attempt is retried, and the second attempt's answers are + returned. A single nameserver is passed so the single-nameserver + lifetime branch is exercised too.""" + answer = MagicMock() + answer.to_text.return_value = "mail.example.com." + with patch.object( + dns.resolver.Resolver, + "resolve", + side_effect=[OSError("transient network error"), [answer]], + ) as mock_resolve: + records = parsedmarc.utils.query_dns( + "example.com", + "A", + nameservers=["192.0.2.53"], + timeout=0.1, + retries=1, + ) + self.assertEqual(records, ["mail.example.com"]) + self.assertEqual(mock_resolve.call_count, 2) + + def testErrorRaisedAfterRetriesExhausted(self): + """When every attempt fails, the last error propagates after + retries+1 total attempts.""" + with patch.object( + dns.resolver.Resolver, + "resolve", + side_effect=OSError("persistent network error"), + ) as mock_resolve: + with self.assertRaises(OSError): + parsedmarc.utils.query_dns( + "example.com", + "A", + nameservers=["192.0.2.53"], + timeout=0.1, + retries=2, + ) + self.assertEqual(mock_resolve.call_count, 3) + + +class TestLoadIpDb(unittest.TestCase): + """Tests for the load_ip_db() download/cache/bundled fallback chain, + mocking at the requests SDK boundary.""" + + def setUp(self): + old_ip_db_path = parsedmarc.utils._IP_DB_PATH + parsedmarc.utils._IP_DB_PATH = None + + def restore(): + parsedmarc.utils._IP_DB_PATH = old_ip_db_path + + self.addCleanup(restore) + + # Redirect the download cache into a per-test directory so the + # tests never touch (or depend on) the real tempdir cache. + self.tmp_dir = tempfile.mkdtemp() + self.addCleanup(lambda: shutil.rmtree(self.tmp_dir, ignore_errors=True)) + patcher = patch( + "parsedmarc.utils.tempfile.gettempdir", return_value=self.tmp_dir + ) + patcher.start() + self.addCleanup(patcher.stop) + + def testExistingLocalFileIsUsedDirectly(self): + """An existing local_file_path wins without any network request""" + local_path = os.path.join(self.tmp_dir, "local.mmdb") + with open(local_path, "wb") as f: + f.write(b"local db") + with patch("parsedmarc.utils.requests.get") as mock_get: + parsedmarc.utils.load_ip_db(local_file_path=local_path) + mock_get.assert_not_called() + self.assertEqual(parsedmarc.utils._IP_DB_PATH, local_path) + + def testDownloadSuccessWritesCacheFile(self): + """A successful download is written to the cache path and selected""" + response = MagicMock() + response.content = b"downloaded db bytes" + response.raise_for_status.return_value = None + with patch("parsedmarc.utils.requests.get", return_value=response) as mock_get: + parsedmarc.utils.load_ip_db(url="https://example.com/db.mmdb") + self.assertEqual(mock_get.call_args.args[0], "https://example.com/db.mmdb") + cached_path = os.path.join(self.tmp_dir, "parsedmarc", "ipinfo_lite.mmdb") + self.assertEqual(parsedmarc.utils._IP_DB_PATH, cached_path) + with open(cached_path, "rb") as f: + self.assertEqual(f.read(), b"downloaded db bytes") + + def testDownloadFailureFallsBackToCachedCopy(self): + """On a network error, a previously cached copy is selected""" + cache_dir = os.path.join(self.tmp_dir, "parsedmarc") + os.makedirs(cache_dir) + cached_path = os.path.join(cache_dir, "ipinfo_lite.mmdb") + with open(cached_path, "wb") as f: + f.write(b"stale cached db") + with patch( + "parsedmarc.utils.requests.get", + side_effect=requests.exceptions.ConnectionError("no network"), + ): + with self.assertLogs("parsedmarc.log", level="WARNING") as cm: + parsedmarc.utils.load_ip_db() + self.assertTrue( + any("Failed to fetch IP database" in message for message in cm.output) + ) + self.assertEqual(parsedmarc.utils._IP_DB_PATH, cached_path) + + def testDownloadFailureFallsBackToBundledCopy(self): + """On a network error with no cached copy, the bundled db is used""" + with patch( + "parsedmarc.utils.requests.get", + side_effect=requests.exceptions.ConnectionError("no network"), + ): + parsedmarc.utils.load_ip_db() + bundled = str(files(parsedmarc.resources.ipinfo).joinpath("ipinfo_lite.mmdb")) + self.assertEqual(parsedmarc.utils._IP_DB_PATH, bundled) + + def testSaveFailureFallsBackToBundledCopy(self): + """A download that cannot be written to disk logs a warning and + falls back to the bundled db instead of crashing. The cache dir is + made uncreatable by pointing gettempdir at a regular file.""" + blocker = os.path.join(self.tmp_dir, "blocker") + with open(blocker, "wb") as f: + f.write(b"not a directory") + response = MagicMock() + response.content = b"downloaded db bytes" + response.raise_for_status.return_value = None + with patch("parsedmarc.utils.tempfile.gettempdir", return_value=blocker): + with patch("parsedmarc.utils.requests.get", return_value=response): + with self.assertLogs("parsedmarc.log", level="WARNING") as cm: + parsedmarc.utils.load_ip_db() + self.assertTrue( + any("Failed to save IP database" in message for message in cm.output) + ) + bundled = str(files(parsedmarc.resources.ipinfo).joinpath("ipinfo_lite.mmdb")) + self.assertEqual(parsedmarc.utils._IP_DB_PATH, bundled) + + +class TestConfigureIpinfoApiProbe(unittest.TestCase): + """Tests for the configure_ipinfo_api() token probe.""" + + def setUp(self): + self.addCleanup(parsedmarc.utils.configure_ipinfo_api, None) + + @staticmethod + def _response(status_code, json_body=None): + response = MagicMock() + response.status_code = status_code + response.ok = 200 <= status_code < 300 + response.json.return_value = json_body if json_body is not None else {} + return response + + def testProbeSuccessLogsConfigured(self): + """A successful probe logs that the API is configured""" + api_json = {"ip": "1.1.1.1", "asn": "AS13335", "country_code": "US"} + with patch( + "parsedmarc.utils.requests.get", + return_value=self._response(200, api_json), + ): + with self.assertLogs("parsedmarc.log", level="INFO") as cm: + parsedmarc.utils.configure_ipinfo_api("fake-token", probe=True) + self.assertTrue( + any("IPinfo API configured" in message for message in cm.output) + ) + + def testProbeNetworkErrorKeepsToken(self): + """A probe network error logs a warning but keeps the token so + per-request fallback can take over later""" + with patch( + "parsedmarc.utils.requests.get", + side_effect=requests.exceptions.ConnectionError("no network"), + ): + with self.assertLogs("parsedmarc.log", level="WARNING") as cm: + parsedmarc.utils.configure_ipinfo_api("fake-token", probe=True) + self.assertTrue( + any("IPinfo API probe failed" in message for message in cm.output) + ) + self.assertEqual(parsedmarc.utils._IPINFO_API_TOKEN, "fake-token") + + def testProbeInvalidKeyRaises(self): + """A 401 during the probe raises InvalidIPinfoAPIKey""" + with patch("parsedmarc.utils.requests.get", return_value=self._response(401)): + with self.assertRaises(parsedmarc.utils.InvalidIPinfoAPIKey): + parsedmarc.utils.configure_ipinfo_api("bad-token", probe=True) + + +class TestIpinfoApiLookupFallbacks(unittest.TestCase): + """API lookup failures other than 401/403 must fall back to the MMDB + silently: network errors, non-JSON bodies, and non-dict payloads.""" + + def setUp(self): + parsedmarc.utils.configure_ipinfo_api("fake-token", probe=False) + self.addCleanup(parsedmarc.utils.configure_ipinfo_api, None) + + def _assert_mmdb_fallback(self, response=None, side_effect=None): + with patch( + "parsedmarc.utils.requests.get", + return_value=response, + side_effect=side_effect, + ): + record = parsedmarc.utils.get_ip_address_db_record("8.8.8.8") + # The bundled MMDB attributes 8.8.8.8 to Google's ASN. + self.assertIsNotNone(record) + assert record is not None + self.assertEqual(record["asn"], 15169) + + def testNetworkErrorFallsBackToMmdb(self): + self._assert_mmdb_fallback( + side_effect=requests.exceptions.ConnectionError("no network") + ) + + def testNonJsonBodyFallsBackToMmdb(self): + response = MagicMock() + response.status_code = 200 + response.ok = True + response.json.side_effect = ValueError("not JSON") + self._assert_mmdb_fallback(response=response) + + def testNonDictPayloadFallsBackToMmdb(self): + response = MagicMock() + response.status_code = 200 + response.ok = True + response.json.return_value = ["not", "a", "dict"] + self._assert_mmdb_fallback(response=response) + + +class TestNormalizeIpRecord(unittest.TestCase): + """_normalize_ip_record must produce the same internal shape from both + the IPinfo API schema and the MaxMind MMDB schema.""" + + def testMaxMindSchema(self): + """MaxMind-style records (nested country iso_code, ASN under + autonomous_system_number/organization) normalize correctly""" + record = parsedmarc.utils._normalize_ip_record( + { + "country": {"iso_code": "US"}, + "autonomous_system_number": 15169, + "autonomous_system_organization": "Google LLC", + } + ) + self.assertEqual(record["country"], "US") + self.assertEqual(record["asn"], 15169) + self.assertEqual(record["as_name"], "Google LLC") + self.assertIsNone(record["as_domain"]) + + def testIntegerAsnPassesThrough(self): + """An already-integer asn field is stored as-is, and as_domain is + lowercased on the way in""" + record = parsedmarc.utils._normalize_ip_record( + {"country_code": "US", "asn": 64496, "as_domain": "EXAMPLE.com"} + ) + self.assertEqual(record["asn"], 64496) + self.assertEqual(record["as_domain"], "example.com") + + if __name__ == "__main__": unittest.main(verbosity=2)