# 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 the RDAP protocol module."""

import email
import unittest
import urllib.error
import urllib.request

from abusectl import rdap


class RedirectPolicy(unittest.TestCase):
    """A redirect is remote data directing our next request.

    urllib.request.Request is used rather than a hand-rolled fake, because
    HTTPRedirectHandler reads attributes (origin_req_host, unverifiable,
    timeout) that a fake would have to reproduce exactly to prove anything.
    """

    def _request(self):
        return urllib.request.Request("https://rdap.example.invalid/ip/192.0.2.1")

    def test_an_https_to_http_downgrade_is_refused(self):
        handler = rdap._NoDowngradeRedirectHandler()
        with self.assertRaises(urllib.error.HTTPError):
            handler.redirect_request(
                self._request(), None, 302, "Found",
                email.message_from_string(""),
                "http://rdap.example.invalid/ip/192.0.2.1",
            )

    def test_an_https_to_https_redirect_is_allowed(self):
        handler = rdap._NoDowngradeRedirectHandler()
        result = handler.redirect_request(
            self._request(), None, 302, "Found",
            email.message_from_string(""),
            "https://other.example.invalid/ip/192.0.2.1",
        )
        self.assertIsNotNone(result)


import json
import tempfile
import time
from pathlib import Path


class Bootstrap(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.cache = Path(self.tmp.name)
        self.addCleanup(self.tmp.cleanup)

    def test_a_missing_file_is_fetched_and_cached(self):
        calls = []

        def fetch(url):
            calls.append(url)
            return {"services": []}

        data = rdap.bootstrap("ipv4", cache_root=self.cache, fetch=fetch)

        self.assertEqual(data, {"services": []})
        self.assertEqual(calls, ["https://data.iana.org/rdap/ipv4.json"])
        self.assertTrue((self.cache / "ipv4.json").exists())

    def test_a_fresh_cache_is_not_refetched(self):
        (self.cache / "ipv4.json").write_text(json.dumps({"services": ["cached"]}))

        def fetch(url):
            raise AssertionError(f"should not have fetched {url}")

        data = rdap.bootstrap("ipv4", cache_root=self.cache, fetch=fetch)
        self.assertEqual(data, {"services": ["cached"]})

    def test_a_stale_cache_is_refetched(self):
        path = self.cache / "ipv4.json"
        path.write_text(json.dumps({"services": ["old"]}))
        old = time.time() - (rdap._BOOTSTRAP_TTL + 60)
        import os
        os.utime(path, (old, old))

        data = rdap.bootstrap(
            "ipv4", cache_root=self.cache, fetch=lambda url: {"services": ["new"]}
        )
        self.assertEqual(data, {"services": ["new"]})

    def test_a_failed_refetch_falls_back_to_the_stale_copy(self):
        """Losing IANA must not stop the user filing a report.

        Last week's map is almost certainly still correct, and a stale
        bootstrap fails safe: the worst case is querying a server that has
        moved, which misses and reads as no contact.
        """
        path = self.cache / "ipv4.json"
        path.write_text(json.dumps({"services": ["old"]}))
        old = time.time() - (rdap._BOOTSTRAP_TTL + 60)
        import os
        os.utime(path, (old, old))

        def fetch(url):
            raise OSError("network is unreachable")

        data = rdap.bootstrap("ipv4", cache_root=self.cache, fetch=fetch)
        self.assertEqual(data, {"services": ["old"]})

    def test_a_failed_fetch_with_no_cache_raises(self):
        def fetch(url):
            raise OSError("network is unreachable")

        with self.assertRaises(rdap.BootstrapUnavailable):
            rdap.bootstrap("ipv4", cache_root=self.cache, fetch=fetch)


class ServerSelection(unittest.TestCase):
    IPV4 = {
        "services": [
            [["192.0.2.0/24"], ["https://wide.example.invalid/"]],
            [["192.0.2.128/25"], ["https://narrow.example.invalid/"]],
            [["198.51.100.0/24"], ["https://other.example.invalid/"]],
        ]
    }
    DNS = {
        "services": [
            [["invalid"], ["https://registry.example.invalid/"]],
            [["test"], ["https://test.example.invalid/"]],
        ]
    }

    def test_an_address_selects_its_range(self):
        self.assertEqual(
            rdap.server_for_ip("198.51.100.7", self.IPV4),
            "https://other.example.invalid/",
        )

    def test_the_longest_prefix_wins(self):
        """192.0.2.200 is in both /24 and /25; the /25 is more specific.

        Choosing the wider range would ask a registry that has delegated
        the block away, and its answer would name the wrong operator.
        """
        self.assertEqual(
            rdap.server_for_ip("192.0.2.200", self.IPV4),
            "https://narrow.example.invalid/",
        )

    def test_an_unlisted_address_selects_nothing(self):
        self.assertIsNone(rdap.server_for_ip("203.0.113.9", self.IPV4))

    def test_a_tld_selects_its_registry(self):
        self.assertEqual(
            rdap.server_for_tld("invalid", self.DNS),
            "https://registry.example.invalid/",
        )

    def test_tld_matching_ignores_case(self):
        self.assertEqual(
            rdap.server_for_tld("INVALID", self.DNS),
            "https://registry.example.invalid/",
        )

    def test_an_unlisted_tld_selects_nothing(self):
        self.assertIsNone(rdap.server_for_tld("example", self.DNS))


class HostileBootstrap(unittest.TestCase):
    """The bootstrap document comes off the same network as an RDAP response.

    abuse_addresses already treats a response as hostile. These assert the
    same posture for the document that decides WHICH server is asked, which
    is the more valuable one to subvert.
    """

    def test_a_non_string_cidr_does_not_become_a_winning_slash_32(self):
        """ipaddress.ip_network(16909060) returns 1.2.3.4/32 rather than
        raising, and a /32 is the longest possible prefix, so it wins every
        contest. The victim is the user, who would file the phishing report
        to a server the phisher chose."""
        data = {
            "services": [
                [["1.2.3.0/24"], ["https://legit.example.invalid/"]],
                [[16909060], ["https://attacker.example.invalid/"]],
            ]
        }
        self.assertEqual(
            rdap.server_for_ip("1.2.3.4", data),
            "https://legit.example.invalid/",
        )

    def test_a_non_string_tld_name_does_not_crash(self):
        data = {"services": [[[42], ["https://attacker.example.invalid/"]]]}
        self.assertIsNone(rdap.server_for_tld("invalid", data))

    def test_a_malformed_services_value_is_not_fatal(self):
        """One bad response is cached for seven days, so a crash here breaks
        every contacts run until the cache expires."""
        for services in ("abc", [[["192.0.2.0/24"]]], [None], [{"a": 1}],
                         [[5, ["https://u.example.invalid/"]]], 7, None):
            with self.subTest(services=services):
                data = {"services": services}
                self.assertIsNone(rdap.server_for_ip("192.0.2.1", data))
                self.assertIsNone(rdap.server_for_tld("invalid", data))

    def test_a_malformed_entry_does_not_hide_a_good_one(self):
        data = {
            "services": [
                None,
                [[["192.0.2.0/24"]]],
                [["192.0.2.0/24"], ["https://legit.example.invalid/"]],
            ]
        }
        self.assertEqual(
            rdap.server_for_ip("192.0.2.1", data),
            "https://legit.example.invalid/",
        )

    def test_a_plaintext_base_url_is_refused(self):
        """_NoDowngradeRedirectHandler guards redirects only. A bootstrap
        naming an http:// base would send the query in clear text,
        disclosing which netblock the user is investigating."""
        ipv4 = {"services": [[["198.51.100.0/24"],
                              ["http://plaintext.example.invalid/"]]]}
        dns = {"services": [[["invalid"],
                             ["http://plaintext.example.invalid/"]]]}
        self.assertIsNone(rdap.server_for_ip("198.51.100.7", ipv4))
        self.assertIsNone(rdap.server_for_tld("invalid", dns))

    def test_a_non_string_url_is_refused(self):
        ipv4 = {"services": [[["198.51.100.0/24"], [42]]]}
        dns = {"services": [[["invalid"], [42]]]}
        self.assertIsNone(rdap.server_for_ip("198.51.100.7", ipv4))
        self.assertIsNone(rdap.server_for_tld("invalid", dns))

    def test_a_plaintext_base_does_not_hide_an_https_one(self):
        data = {
            "services": [
                [["192.0.2.0/25"], ["http://plaintext.example.invalid/"]],
                [["192.0.2.0/24"], ["https://legit.example.invalid/"]],
            ]
        }
        self.assertEqual(
            rdap.server_for_ip("192.0.2.1", data),
            "https://legit.example.invalid/",
        )

    def test_a_plaintext_base_is_never_queried(self):
        data = {"services": [[["198.51.100.0/24"],
                              ["http://plaintext.example.invalid/"]]]}

        def fetch(url):
            raise AssertionError(f"should not have fetched {url}")

        self.assertIsNone(rdap.query_ip("198.51.100.7", data, fetch=fetch))


def _entity(roles, emails, entities=None):
    """Build an RDAP entity in real jCard shape."""
    properties = [["version", {}, "text", "4.0"]]
    for address in emails:
        properties.append(["email", {}, "text", address])
    entity = {"roles": roles, "vcardArray": ["vcard", properties]}
    if entities:
        entity["entities"] = entities
    return entity


class AbuseExtraction(unittest.TestCase):
    def test_an_abuse_entity_yields_its_address(self):
        response = {"entities": [_entity(["abuse"], ["abuse@example.invalid"])]}
        self.assertEqual(
            rdap.abuse_addresses(response), ["abuse@example.invalid"]
        )

    def test_a_nested_abuse_entity_is_found(self):
        """The abuse entity is usually a child of the organisation entity."""
        response = {
            "entities": [
                _entity(
                    ["registrant"], [],
                    entities=[_entity(["abuse"], ["abuse@example.invalid"])],
                )
            ]
        }
        self.assertEqual(
            rdap.abuse_addresses(response), ["abuse@example.invalid"]
        )

    def test_a_technical_only_response_yields_nothing(self):
        """A technical contact is a named human who never volunteered to
        receive abuse mail. Mailing them is useless and is a small privacy
        harm to an uninvolved third party."""
        response = {"entities": [_entity(["technical"], ["someone@example.invalid"])]}
        self.assertEqual(rdap.abuse_addresses(response), [])

    def test_every_abuse_address_is_kept(self):
        """Some netblocks publish two desks, and picking one arbitrarily
        can drop the one that would have answered."""
        response = {
            "entities": [
                _entity(["abuse"], ["one@example.invalid", "two@example.invalid"])
            ]
        }
        self.assertEqual(
            rdap.abuse_addresses(response),
            ["one@example.invalid", "two@example.invalid"],
        )

    def test_a_newline_in_an_address_is_rejected(self):
        """The address becomes a mail recipient in report and submit, so a
        CRLF here is header injection into mail this tool sends."""
        response = {
            "entities": [
                _entity(["abuse"], ["abuse@example.invalid\r\nBcc: victim@example.org"])
            ]
        }
        self.assertEqual(rdap.abuse_addresses(response), [])

    def test_a_bare_control_character_in_an_address_is_rejected(self):
        """The CRLF case above is rejected by parseaddr itself, so it passes
        even with the control-character check deleted. These two are passed
        straight through by parseaddr and reach a mail header raw, so this
        is the case that actually holds _valid_address's own defence."""
        for raw in ("abuse@example.invalid\x0b", "ab\x0cuse@example.invalid"):
            with self.subTest(raw=raw):
                response = {"entities": [_entity(["abuse"], [raw])]}
                self.assertEqual(rdap.abuse_addresses(response), [])

    def test_a_non_address_is_rejected(self):
        response = {"entities": [_entity(["abuse"], ["not an address"])]}
        self.assertEqual(rdap.abuse_addresses(response), [])

    def test_recursion_is_depth_capped(self):
        """Remote JSON must not be able to hang the tool."""
        deep = _entity(["abuse"], ["deep@example.invalid"])
        for _ in range(10):
            deep = _entity(["registrant"], [], entities=[deep])
        self.assertEqual(rdap.abuse_addresses({"entities": [deep]}), [])

    def test_duplicate_addresses_collapse(self):
        response = {
            "entities": [
                _entity(["abuse"], ["abuse@example.invalid"]),
                _entity(["abuse"], ["abuse@example.invalid"]),
            ]
        }
        self.assertEqual(
            rdap.abuse_addresses(response), ["abuse@example.invalid"]
        )


class Query(unittest.TestCase):
    IPV4 = {"services": [[["198.51.100.0/24"], ["https://rir.example.invalid/"]]]}
    DNS = {"services": [[["invalid"], ["https://registry.example.invalid/"]]]}

    def test_an_ip_query_hits_the_selected_server(self):
        calls = []

        def fetch(url):
            calls.append(url)
            return {"handle": "NET-1", "entities": []}

        result = rdap.query_ip("198.51.100.7", self.IPV4, fetch=fetch)

        self.assertEqual(calls, ["https://rir.example.invalid/ip/198.51.100.7"])
        self.assertEqual(result["handle"], "NET-1")

    def test_an_unlisted_ip_is_not_queried(self):
        def fetch(url):
            raise AssertionError(f"should not have fetched {url}")

        self.assertIsNone(rdap.query_ip("203.0.113.9", self.IPV4, fetch=fetch))

    def test_the_label_walk_stops_at_the_first_answer(self):
        """mail.deep.example.invalid is not registrable; example.invalid is.

        The registry is the authority on what is registrable, which is why
        this walks rather than carrying a Public Suffix List that would go
        stale weekly.
        """
        calls = []

        def fetch(url):
            calls.append(url)
            if url.endswith("/domain/example.invalid"):
                return {"handle": "DOM-1", "entities": []}
            raise urllib.error.HTTPError(url, 404, "Not Found", {}, None)

        result, queried = rdap.query_domain(
            "mail.deep.example.invalid", self.DNS, fetch=fetch
        )

        self.assertEqual(queried, "example.invalid")
        self.assertEqual(result["handle"], "DOM-1")
        self.assertEqual(len(calls), 3)

    def test_the_walk_never_queries_a_bare_tld(self):
        calls = []

        def fetch(url):
            calls.append(url)
            raise urllib.error.HTTPError(url, 404, "Not Found", {}, None)

        result, queried = rdap.query_domain(
            "deep.example.invalid", self.DNS, fetch=fetch
        )

        self.assertIsNone(result)
        self.assertNotIn("https://registry.example.invalid/domain/invalid", calls)

    def test_a_tld_with_no_server_is_not_queried(self):
        def fetch(url):
            raise AssertionError(f"should not have fetched {url}")

        result, queried = rdap.query_domain(
            "example.test", self.DNS, fetch=fetch
        )
        self.assertIsNone(result)

    def test_the_walk_is_capped(self):
        calls = []

        def fetch(url):
            calls.append(url)
            raise urllib.error.HTTPError(url, 404, "Not Found", {}, None)

        # Twelve labels, so the bare-TLD guard alone would allow eleven
        # attempts and only the cap can stop the walk at five. The expected
        # count is hard-coded: reading rdap._MAX_LABEL_WALK here would move
        # the assertion along with the mutation and the test could not fail.
        host = "a.b.c.d.e.f.g.h.i.j.example.invalid"
        rdap.query_domain(host, self.DNS, fetch=fetch)
        self.assertEqual(len(calls), 5)


class QuotedQueryComponent(unittest.TestCase):
    """The interpolated component cannot escape its path segment.

    contacts.is_queryable() is the admission point and validates every
    candidate, but that guarantee has already leaked three times in this
    module's history, each time the same shape: a validator applied to one
    branch and forgotten on its sibling. These call query_ip and
    query_domain DIRECTLY, bypassing contacts entirely, because a future
    branch that skips worklist() is exactly the failure this layer exists
    to survive.
    """

    IPV4 = {"services": [[["198.51.100.0/24"], ["https://rir.example.invalid/"]]]}
    IPV6 = {"services": [[["fe80::/10"], ["https://rir.example.invalid/"]]]}
    DOCV6 = {"services": [[["2001:db8::/32"], ["https://r6.example.invalid/"]]]}
    DNS = {"services": [[["invalid"], ["https://registry.example.invalid/"]]]}

    def _recorder(self):
        calls = []

        def fetch(url):
            calls.append(url)
            return {"handle": "X", "entities": []}

        return calls, fetch

    def test_a_traversal_in_an_ip_stays_under_the_ip_segment(self):
        """server_for_ip refuses this string today, so the base is forced.

        Forcing it is the point: this asserts what query_ip does with a
        component it was handed, not what today's lookup happens to reject.
        """
        calls, fetch = self._recorder()
        original = rdap.server_for_ip
        rdap.server_for_ip = lambda address, data: "https://rir.example.invalid/"
        try:
            rdap.query_ip("198.51.100.7/../../etc", self.IPV4, fetch=fetch)
        finally:
            rdap.server_for_ip = original

        self.assertEqual(
            calls,
            ["https://rir.example.invalid/ip/198.51.100.7%2F..%2F..%2Fetc"],
        )

    def test_a_scope_id_in_an_ip_is_encoded_not_left_malformed(self):
        """A bare % in a URL is a truncated escape, not a literal percent.

        server_for_ip accepts fe80::1%eth0 today, so this one reaches the
        wire through the normal path. The colons stay literal; only the
        scope separator is encoded.
        """
        calls, fetch = self._recorder()
        rdap.query_ip("fe80::1%eth0", self.IPV6, fetch=fetch)

        self.assertEqual(calls, ["https://rir.example.invalid/ip/fe80::1%25eth0"])

    def test_a_normal_ipv6_address_is_not_encoded_at_all(self):
        """RFC 9082 wants the address in its TEXT form, and pchar allows ":".

        Encoding the colon does not fail loudly. A registry that does not
        normalise before matching answers 404, which this tool reads as
        "this netblock publishes no abuse desk" rather than "we asked the
        wrong question", and the abuse contact for an IPv6 hop is lost
        silently. The happy-path guard was IPv4 only, so an over-broad
        safe="" got through once already.
        """
        calls, fetch = self._recorder()
        rdap.query_ip("2001:db8::1", self.DOCV6, fetch=fetch)

        self.assertEqual(calls, ["https://r6.example.invalid/ip/2001:db8::1"])

    def test_a_query_and_fragment_in_a_host_are_not_live(self):
        calls, fetch = self._recorder()
        rdap.query_domain("victim?e=x#frag.example.invalid", self.DNS, fetch=fetch)

        self.assertEqual(len(calls), 1)
        url = calls[0]
        self.assertNotIn("?", url)
        self.assertNotIn("#", url)
        self.assertIn("%3F", url)
        self.assertIn("%23", url)

    def test_a_normal_ip_url_is_unchanged(self):
        calls, fetch = self._recorder()
        rdap.query_ip("198.51.100.7", self.IPV4, fetch=fetch)
        self.assertEqual(calls, ["https://rir.example.invalid/ip/198.51.100.7"])

    def test_a_normal_host_url_is_unchanged(self):
        calls, fetch = self._recorder()
        rdap.query_domain("example.invalid", self.DNS, fetch=fetch)
        self.assertEqual(
            calls, ["https://registry.example.invalid/domain/example.invalid"]
        )

    def test_the_returned_candidate_is_unquoted(self):
        """The manifest and the review dialog show the name, not the URL."""
        calls, fetch = self._recorder()
        _, queried = rdap.query_domain(
            "victim?e=x#frag.example.invalid", self.DNS, fetch=fetch
        )
        self.assertEqual(queried, "victim?e=x#frag.example.invalid")


if __name__ == "__main__":
    unittest.main()