path: root/tests/test_contacts.py

# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# 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), [])

    def test_a_trailing_dot_url_folds_with_the_bare_host(self):
        """_host_of must strip the root dot, not just the domain branch.
        Failing to fold means disclosing the same host to a registry
        twice, which doubles what the user leaks per campaign."""
        iocs = [
            {"id": "ioc-1", "type": "url", "value": "http://a.example.invalid./x"},
            {"id": "ioc-2", "type": "url", "value": "http://a.example.invalid/x"},
        ]
        work = contacts.worklist(iocs)
        self.assertEqual(len(work), 1)
        self.assertEqual(work[0].query, "a.example.invalid")

    def test_an_unparseable_ip_indicator_is_flagged_rather_than_dropped(self):
        """Silent was the bug: an ip indicator the parser mangled must
        still show up in the manifest for the user to see."""
        iocs = [{"id": "ioc-1", "type": "ipv4", "value": "999.999.999.999"}]
        work = contacts.worklist(iocs)
        self.assertEqual([(i.kind, i.query) for i in work],
                         [("unusable", "999.999.999.999")])


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)


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_an_ipv6_scope_id_is_never_queried(self):
        """ipaddress.ip_address accepts a scope id since Python 3.9, and
        everything after the % is free text the attacker chose. rdap
        interpolates the query unquoted, so a scope id carrying the
        recipient's own address would reach a registry, which is the
        disclosure the fourth property exists to prevent."""
        for value in ("fe80::1%eth0", "fe80::1%victim@example.org"):
            with self.subTest(value=value):
                calls, results = self._calls_for(value, ioc_type="ipv6")
                self.assertEqual(calls, [])
                self.assertEqual(results[0]["iocs"], ["ioc-1"])
                self.assertIn("refused as a query", results[0]["error"])

    def test_an_ipv6_scope_id_in_a_url_is_never_queried(self):
        calls, results = self._calls_for(
            "http://[fe80::1%25eth0]/x", ioc_type="url"
        )
        self.assertEqual(calls, [])
        self.assertIn("refused as a query", results[0]["error"])

    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.assertEqual(
            results[0]["error"],
            "refused as a query: not a bare host or IP address, "
            "so it was never sent to a registry",
        )

    def test_a_non_ascii_host_is_rejected_rather_than_guessed_at(self):
        """A non-ASCII name is plausibly a legitimate IDN indicator the
        user may want to handle by hand, and an illegal character is
        hostile. Those imply different actions, so the reasons differ."""
        calls, results = self._calls_for("exämple.invalid")
        self.assertEqual(calls, [])
        self.assertEqual(
            results[0]["error"],
            "refused as a query: not ASCII, and we do not guess at an IDN "
            "encoding, so it was never sent to a registry",
        )

    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_single_label_host_is_refused(self):
        """Asserted on is_queryable rather than through resolve: a
        single-label name matches no tld in the bootstrap, so a resolve
        test issues no query whatever the validator decides and could
        never fail. A bare label is not a registrable name and asking a
        registry about one discloses the investigation for nothing."""
        for value in ("localhost", "invalid", "a"):
            with self.subTest(value=value):
                self.assertFalse(contacts.is_queryable(value))

    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()