diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_contacts.py | 98 |
1 files changed, 98 insertions, 0 deletions
diff --git a/tests/test_contacts.py b/tests/test_contacts.py index 0ac197d..0451fe6 100644 --- a/tests/test_contacts.py +++ b/tests/test_contacts.py @@ -235,5 +235,103 @@ class Resolve(unittest.TestCase): self.assertNotIn("?", url) +class HostileDomainIndicator(unittest.TestCase): + """THE FOURTH PROPERTY through the domain branch. + + The url branch is cleaned by _host_of. The domain branch took its value + from parse._domain_of, which is everything after the @ of a From, + Sender or Reply-To addr-spec, a header the attacker owns completely, + and email.utils.parseaddr permits /, ?, # and % there. + """ + + IPV4 = {"services": [[["198.51.100.0/24"], ["https://rir.example.invalid/"]]]} + IPV6 = {"services": []} + DNS = {"services": [[["invalid"], ["https://registry.example.invalid/"]]]} + + def _bootstraps(self): + return {"ipv4": self.IPV4, "ipv6": self.IPV6, "dns": self.DNS} + + def _calls_for(self, value, ioc_type="domain"): + calls = [] + + def fetch(url): + calls.append(url) + return {"handle": "DOM-1", "entities": []} + + iocs = [{"id": "ioc-1", "type": ioc_type, "value": value}] + results = contacts.resolve( + iocs, bootstraps=self._bootstraps(), fetch=fetch + ) + return calls, results + + def test_a_sender_domain_carrying_the_victim_address_is_never_queried(self): + """The attacker writes the recipient's own address into the domain + of the From header, and this tool would send it to a registry. That + is the disclosure the fourth property exists to prevent, reaching a + third party.""" + raw = (b"Received: from relay.example.invalid ([192.0.2.10])\r\n" + b"From: Bank <phish@victim%40example.org.invalid>\r\n" + b"Subject: test\r\n\r\nbody\r\n") + from abusectl import parse + iocs = parse.iocs(raw, trusted=["192.0.2.0/24"]) + calls = [] + + def fetch(url): + calls.append(url) + return {"handle": "DOM-1", "entities": []} + + contacts.resolve(iocs, bootstraps=self._bootstraps(), fetch=fetch) + for url in calls: + self.assertNotIn("victim", url) + self.assertNotIn("%40", url) + + def test_a_traversal_value_is_never_queried(self): + """a/../../x.invalid escapes the /domain/ endpoint altogether.""" + calls, _ = self._calls_for("a/../../x.invalid") + self.assertEqual(calls, []) + + def test_query_fragment_and_space_values_are_never_queried(self): + for value in ("a?e=secret.invalid", "a#frag.invalid", "a b.invalid", + "a@b.invalid", "a:80.invalid", "a\x00b.invalid"): + with self.subTest(value=value): + calls, _ = self._calls_for(value) + self.assertEqual(calls, []) + + def test_a_rejected_value_is_flagged_rather_than_dropped(self): + """Silent was the bug elsewhere too: the user must see it during + review rather than wonder why an indicator vanished.""" + calls, results = self._calls_for("a?e=secret.invalid") + self.assertEqual(len(results), 1) + self.assertEqual(results[0]["iocs"], ["ioc-1"]) + self.assertEqual(results[0]["abuse"], []) + self.assertIn("hostname", results[0]["error"]) + + def test_a_non_ascii_host_is_rejected_rather_than_guessed_at(self): + calls, results = self._calls_for("exämple.invalid") + self.assertEqual(calls, []) + self.assertIn("hostname", results[0]["error"]) + + def test_a_bad_label_is_rejected(self): + for value in ("a..invalid", "-a.invalid", "a-.invalid", + "x" * 64 + ".invalid", ("a." * 130) + "invalid"): + with self.subTest(value=value): + calls, _ = self._calls_for(value) + self.assertEqual(calls, []) + + def test_a_normal_host_still_resolves(self): + calls, results = self._calls_for("mail.example.invalid") + self.assertEqual( + calls, ["https://registry.example.invalid/domain/mail.example.invalid"] + ) + self.assertEqual(results[0]["handle"], "DOM-1") + + def test_an_ip_valued_domain_indicator_still_takes_the_ip_branch(self): + work = contacts.worklist( + [{"id": "ioc-1", "type": "domain", "value": "198.51.100.7"}] + ) + self.assertEqual([(i.kind, i.query) for i in work], + [("ip", "198.51.100.7")]) + + if __name__ == "__main__": unittest.main() |