# Copyright (C) 2026 Danilo M. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License version 2 as # published by the Free Software Foundation. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. """Tests for turning indicators into abuse contacts.""" import unittest from abusectl import contacts class Worklist(unittest.TestCase): def test_ips_and_domains_are_resolvable(self): iocs = [ {"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}, {"id": "ioc-2", "type": "domain", "value": "example.invalid"}, ] work = contacts.worklist(iocs) self.assertEqual( {(item.kind, item.query) for item in work}, {("ip", "198.51.100.7"), ("domain", "example.invalid")}, ) def test_hashes_and_observations_are_not_resolvable(self): iocs = [ {"id": "ioc-1", "type": "sha256", "value": "e3b0c442"}, {"id": "ioc-2", "type": "observation", "value": "display-name-carries-address"}, ] self.assertEqual(contacts.worklist(iocs), []) def test_a_url_contributes_only_its_host(self): """THE FOURTH PROPERTY. A query discloses what the user is looking at, and a URL path can carry recipient identity that suspect_path_segments deliberately flags rather than redacts.""" iocs = [{ "id": "ioc-1", "type": "url", "value": "https://login.example.invalid/verify/victim%40example.org?e=REDACTED", }] work = contacts.worklist(iocs) self.assertEqual(len(work), 1) self.assertEqual(work[0].kind, "domain") self.assertEqual(work[0].query, "login.example.invalid") def test_url_userinfo_never_reaches_the_query(self): iocs = [{ "id": "ioc-1", "type": "url", "value": "https://victim%40example.org:secret@login.example.invalid/x", }] work = contacts.worklist(iocs) self.assertEqual(work[0].query, "login.example.invalid") def test_a_url_port_is_stripped(self): iocs = [{"id": "ioc-1", "type": "url", "value": "https://login.example.invalid:8443/x"}] self.assertEqual(contacts.worklist(iocs)[0].query, "login.example.invalid") def test_a_url_host_that_is_an_ip_resolves_as_an_ip(self): iocs = [{"id": "ioc-1", "type": "url", "value": "http://198.51.100.7/login"}] work = contacts.worklist(iocs) self.assertEqual(work[0].kind, "ip") self.assertEqual(work[0].query, "198.51.100.7") def test_a_bracketed_ipv6_url_host_resolves_as_an_ip(self): iocs = [{"id": "ioc-1", "type": "url", "value": "http://[2001:db8::1]/login"}] work = contacts.worklist(iocs) self.assertEqual(work[0].kind, "ip") self.assertEqual(work[0].query, "2001:db8::1") def test_hosts_fold_and_keep_every_contributing_ioc(self): """Twenty URLs on one host must produce one query.""" iocs = [ {"id": "ioc-1", "type": "url", "value": "https://a.example.invalid/one"}, {"id": "ioc-2", "type": "url", "value": "https://a.example.invalid/two"}, {"id": "ioc-3", "type": "domain", "value": "a.example.invalid"}, ] work = contacts.worklist(iocs) self.assertEqual(len(work), 1) self.assertEqual(work[0].iocs, ["ioc-1", "ioc-2", "ioc-3"]) def test_a_trailing_dot_folds_with_the_bare_host(self): iocs = [ {"id": "ioc-1", "type": "domain", "value": "example.invalid."}, {"id": "ioc-2", "type": "domain", "value": "example.invalid"}, ] self.assertEqual(len(contacts.worklist(iocs)), 1) def test_an_untrusted_hop_is_still_resolved(self): """A forged chain's IP may still be the real sender's; the confidence marker stays in the manifest for review.""" iocs = [{"id": "ioc-1", "type": "ipv4", "value": "203.0.113.99", "confidence": "untrusted-hop"}] self.assertEqual(len(contacts.worklist(iocs)), 1) def test_a_malformed_url_contributes_nothing(self): iocs = [{"id": "ioc-1", "type": "url", "value": "not a url"}] self.assertEqual(contacts.worklist(iocs), []) class Resolve(unittest.TestCase): 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 test_an_ip_resolves_to_its_abuse_desk(self): def fetch(url): return { "handle": "NET-1", "entities": [{ "roles": ["abuse"], "vcardArray": ["vcard", [ ["version", {}, "text", "4.0"], ["email", {}, "text", "abuse@example.invalid"], ]], }], } iocs = [{"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}] result = contacts.resolve( iocs, bootstraps=self._bootstraps(), fetch=fetch ) self.assertEqual(len(result), 1) self.assertEqual(result[0]["iocs"], ["ioc-1"]) self.assertEqual(result[0]["query"], "198.51.100.7") self.assertEqual(result[0]["abuse"], ["abuse@example.invalid"]) self.assertEqual(result[0]["handle"], "NET-1") self.assertNotIn("error", result[0]) def test_no_abuse_role_records_a_reason_not_an_error(self): def fetch(url): return {"handle": "NET-2", "entities": []} iocs = [{"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}] result = contacts.resolve( iocs, bootstraps=self._bootstraps(), fetch=fetch ) self.assertEqual(result[0]["abuse"], []) self.assertEqual(result[0]["error"], "no abuse role published") def test_no_rdap_server_records_a_reason(self): def fetch(url): raise AssertionError(f"should not have fetched {url}") iocs = [{"id": "ioc-1", "type": "domain", "value": "example.test"}] result = contacts.resolve( iocs, bootstraps=self._bootstraps(), fetch=fetch ) self.assertEqual(result[0]["abuse"], []) self.assertIn("no rdap server", result[0]["error"]) def test_a_network_failure_is_per_query_and_does_not_stop_the_run(self): def fetch(url): if "198.51.100.7" in url: raise OSError("connection timed out") return { "handle": "DOM-1", "entities": [{ "roles": ["abuse"], "vcardArray": ["vcard", [ ["version", {}, "text", "4.0"], ["email", {}, "text", "abuse@example.invalid"], ]], }], } iocs = [ {"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}, {"id": "ioc-2", "type": "domain", "value": "example.invalid"}, ] result = contacts.resolve( iocs, bootstraps=self._bootstraps(), fetch=fetch ) self.assertEqual(len(result), 2) failed = [r for r in result if r["query"] == "198.51.100.7"][0] worked = [r for r in result if r["query"] == "example.invalid"][0] self.assertIn("connection timed out", failed["error"]) self.assertEqual(worked["abuse"], ["abuse@example.invalid"]) def test_one_query_per_host_however_many_iocs(self): calls = [] def fetch(url): calls.append(url) return {"handle": "DOM-1", "entities": []} iocs = [ {"id": f"ioc-{n}", "type": "url", "value": f"https://a.example.invalid/page{n}"} for n in range(20) ] result = contacts.resolve( iocs, bootstraps=self._bootstraps(), fetch=fetch ) self.assertEqual(len(calls), 1) self.assertEqual(len(result[0]["iocs"]), 20) def test_no_query_ever_carries_a_path(self): """THE FOURTH PROPERTY, asserted at the transport.""" calls = [] def fetch(url): calls.append(url) return {"handle": "DOM-1", "entities": []} iocs = [{ "id": "ioc-1", "type": "url", "value": "https://a.example.invalid/verify/victim%40example.org?e=x", }] contacts.resolve(iocs, bootstraps=self._bootstraps(), fetch=fetch) for url in calls: self.assertNotIn("victim", url) self.assertNotIn("verify", url) self.assertNotIn("%40", url) self.assertNotIn("?", url) if __name__ == "__main__": unittest.main()