desec-stack/api/desecapi/tests/test_rrsets.py

from contextlib import nullcontext
from ipaddress import IPv4Network
import re
from itertools import product
from math import ceil, floor

from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.management import call_command
from rest_framework import status

from desecapi.models import BlockedSubnet, Domain, RR, RRset
from desecapi.models.records import RR_SET_TYPES_AUTOMATIC, RR_SET_TYPES_UNSUPPORTED
from desecapi.tests.base import DesecTestCase, AuthenticatedRRSetBaseTestCase


class UnauthenticatedRRSetTestCase(DesecTestCase):
    def test_unauthorized_access(self):
        url = self.reverse("v1:rrsets", name="example.com")
        for method in [
            self.client.get,
            self.client.post,
            self.client.put,
            self.client.delete,
            self.client.patch,
        ]:
            response = method(url)
            self.assertStatus(response, status.HTTP_401_UNAUTHORIZED)


class AuthenticatedRRSetTestCase(AuthenticatedRRSetBaseTestCase):
    def test_subname_validity(self):
        for subname in [
            "aEroport",
            "AEROPORT",
            "aéroport",
            "a" * 64,
        ]:
            with self.assertRaises(ValidationError):
                RRset(domain=self.my_domain, subname=subname, ttl=60, type="A").save()
        for subname in [
            "aeroport",
            "a" * 63,
        ]:
            RRset(domain=self.my_domain, subname=subname, ttl=60, type="A").save()

    def test_retrieve_my_rr_sets(self):
        for response in [
            self.client.get_rr_sets(self.my_domain.name),
            self.client.get_rr_sets(self.my_domain.name, subname=""),
        ]:
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertEqual(len(response.data), 1, response.data)

    def test_retrieve_my_rr_sets_pagination(self):
        def convert_links(links):
            mapping = {}
            for link in links.split(", "):
                _url, label = link.split("; ")
                label = re.search('rel="(.*)"', label).group(1)
                _url = _url[1:-1]
                assert label not in mapping
                mapping[label] = _url
            return mapping

        def assertPaginationResponse(
            response, expected_length, expected_directional_links=[]
        ):
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertEqual(len(response.data), expected_length)

            _links = convert_links(response["Link"])
            self.assertEqual(
                len(_links), len(expected_directional_links) + 1
            )  # directional links, plus "first"
            self.assertTrue(_links["first"].endswith("/?cursor="))
            for directional_link in expected_directional_links:
                self.assertEqual(
                    _links["first"].find("/?cursor="),
                    _links[directional_link].find("/?cursor="),
                )
                self.assertTrue(len(_links[directional_link]) > len(_links["first"]))

        # Prepare extra records so that we get three pages (total: n + 1)
        n = int(settings.REST_FRAMEWORK["PAGE_SIZE"] * 2.5)
        RRset.objects.bulk_create(
            [
                RRset(domain=self.my_domain, subname=str(i), ttl=123, type="A")
                for i in range(n)
            ]
        )

        # No pagination
        response = self.client.get_rr_sets(self.my_domain.name)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(
            response.data["detail"],
            f'Pagination required. You can query up to {settings.REST_FRAMEWORK["PAGE_SIZE"]} items at a time ({n+1} total). '
            "Please use the `first` page link (see Link header).",
        )
        links = convert_links(response["Link"])
        self.assertEqual(len(links), 1)
        self.assertTrue(links["first"].endswith("/?cursor="))

        # First page
        url = links["first"]
        response = self.client.get(url)
        assertPaginationResponse(
            response, settings.REST_FRAMEWORK["PAGE_SIZE"], ["next"]
        )

        # Next
        url = convert_links(response["Link"])["next"]
        response = self.client.get(url)
        assertPaginationResponse(
            response, settings.REST_FRAMEWORK["PAGE_SIZE"], ["next", "prev"]
        )
        data_next = response.data.copy()

        # Next-next (last) page
        url = convert_links(response["Link"])["next"]
        response = self.client.get(url)
        assertPaginationResponse(response, n / 5 + 1, ["prev"])

        # Prev
        url = convert_links(response["Link"])["prev"]
        response = self.client.get(url)
        assertPaginationResponse(
            response, settings.REST_FRAMEWORK["PAGE_SIZE"], ["next", "prev"]
        )

        # Make sure that one step forward equals two steps forward and one step back
        self.assertEqual(response.data, data_next)

    def test_retrieve_other_rr_sets(self):
        self.assertStatus(
            self.client.get_rr_sets(self.other_domain.name), status.HTTP_404_NOT_FOUND
        )
        self.assertStatus(
            self.client.get_rr_sets(self.other_domain.name, subname="test"),
            status.HTTP_404_NOT_FOUND,
        )
        self.assertStatus(
            self.client.get_rr_sets(self.other_domain.name, type="A"),
            status.HTTP_404_NOT_FOUND,
        )

    def test_retrieve_my_rr_sets_filter(self):
        response = self.client.get_rr_sets(self.my_rr_set_domain.name, query="?cursor=")
        self.assertStatus(response, status.HTTP_200_OK)
        expected_number_of_rrsets = min(
            len(self._test_rr_sets()), settings.REST_FRAMEWORK["PAGE_SIZE"]
        )
        self.assertEqual(len(response.data), expected_number_of_rrsets)

        for subname in self.SUBNAMES:
            response = self.client.get_rr_sets(
                self.my_rr_set_domain.name, subname=subname
            )
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertRRSetsCount(
                response.data,
                [dict(subname=subname)],
                count=len(self._test_rr_sets(subname=subname)),
            )

        for type_ in self.ALLOWED_TYPES:
            response = self.client.get_rr_sets(self.my_rr_set_domain.name, type=type_)
            self.assertStatus(response, status.HTTP_200_OK)

    def test_create_my_rr_sets(self):
        for subname in [
            None,
            "create-my-rr-sets",
            "foo.create-my-rr-sets",
            "bar.baz.foo.create-my-rr-sets",
        ]:
            for data in [
                {"subname": subname, "records": ["1.2.3.4"], "ttl": 3660, "type": "A"},
                {
                    "subname": "" if subname is None else subname,
                    "records": ["desec.io."],
                    "ttl": 36900,
                    "type": "PTR",
                },
                {
                    "subname": "" if subname is None else subname,
                    "ttl": 3650,
                    "type": "TXT",
                    "records": ['"foo"'],
                },
                {
                    "subname": f"{subname}.cname".lower(),
                    "ttl": 3600,
                    "type": "CNAME",
                    "records": ["example.com."],
                },
            ]:
                # Try POST with missing subname
                if data["subname"] is None:
                    data.pop("subname")

                with self.assertRequests(
                    self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
                ):
                    response = self.client.post_rr_set(
                        domain_name=self.my_empty_domain.name, **data
                    )
                    self.assertStatus(response, status.HTTP_201_CREATED)
                    self.assertTrue(
                        all(
                            field in response.data
                            for field in [
                                "created",
                                "domain",
                                "subname",
                                "name",
                                "records",
                                "ttl",
                                "type",
                                "touched",
                            ]
                        )
                    )
                    self.assertEqual(
                        self.my_empty_domain.touched,
                        max(
                            rrset.touched
                            for rrset in self.my_empty_domain.rrset_set.all()
                        ),
                    )

                # Check for uniqueness on second attempt
                response = self.client.post_rr_set(
                    domain_name=self.my_empty_domain.name, **data
                )
                self.assertContains(
                    response,
                    "Another RRset with the same subdomain and type exists for this domain.",
                    status_code=status.HTTP_400_BAD_REQUEST,
                )

                response = self.client.get_rr_sets(self.my_empty_domain.name)
                self.assertStatus(response, status.HTTP_200_OK)
                self.assertRRSetsCount(response.data, [data])

                response = self.client.get_rr_set(
                    self.my_empty_domain.name, data.get("subname", ""), data["type"]
                )
                self.assertStatus(response, status.HTTP_200_OK)
                self.assertRRSet(response.data, **data)

    def test_create_my_rr_sets_type_restriction(self):
        for subname in [
            "",
            "create-my-rr-sets",
            "foo.create-my-rr-sets",
            "bar.baz.foo.create-my-rr-sets",
        ]:
            for data in (
                [
                    {"subname": subname, "ttl": 60, "type": "a"},
                    {
                        "subname": subname,
                        "records": ["10 example.com."],
                        "ttl": 60,
                        "type": "txt",
                    },
                ]
                + [
                    {
                        "subname": subname,
                        "records": ["10 example.com."],
                        "ttl": 60,
                        "type": type_,
                    }
                    for type_ in self.UNSUPPORTED_TYPES
                ]
                + [
                    {
                        "subname": subname,
                        "records": [
                            "get.desec.io. get.desec.io. 2584 10800 3600 604800 60"
                        ],
                        "ttl": 60,
                        "type": type_,
                    }
                    for type_ in self.AUTOMATIC_TYPES
                ]
            ):
                response = self.client.post_rr_set(self.my_domain.name, **data)
                self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

                response = self.client.get_rr_sets(self.my_domain.name)
                self.assertStatus(response, status.HTTP_200_OK)
                self.assertRRSetsCount(response.data, [data], count=0)

    def test_create_my_rr_sets_only_at_apex(self):
        for type_, records in {
            "DNSKEY": ["257 3 15 l02Woi0iS8Aa25FQkUd9RMzZHJpBoRQwAQEX1SxZJA4="],
        }.items():
            data = {
                "subname": "non-apex",
                "ttl": 3600,
                "type": type_,
                "records": records,
            }
            r = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertContains(
                r,
                f"{type_} RRset must have empty subname",
                status_code=status.HTTP_400_BAD_REQUEST,
            )

    def test_create_my_rr_sets_restricted_at_apex(self):
        for type_, records in {
            "CNAME": ["foobar.com."],
            "DS": ["45586 5 1 D0FDF996D1AF2CCDBDC942B02CB02D379629E20B"],
        }.items():
            data = {"subname": "", "ttl": 3600, "type": type_, "records": records}
            r = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertContains(
                r,
                f"{type_} RRset cannot have empty subname",
                status_code=status.HTTP_400_BAD_REQUEST,
            )

    def test_create_my_rr_sets_cname_multiple_records(self):
        for records in (["foobar.com.", "foobar.com."], ["foobar.com.", "foobar.org."]):
            data = {"subname": "asdf", "ttl": 3600, "type": "CNAME", "records": records}
            response = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertContains(
                response,
                "CNAME RRset cannot have multiple records",
                status_code=status.HTTP_400_BAD_REQUEST,
            )

    def test_create_my_rr_sets_dname_multiple_records(self):
        for records in (["foobar.com.", "foobar.com."], ["foobar.com.", "foobar.org."]):
            data = {"subname": "asdf", "ttl": 3600, "type": "DNAME", "records": records}
            response = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertContains(
                response,
                "DNAME RRset cannot have multiple records",
                status_code=status.HTTP_400_BAD_REQUEST,
            )

    def test_create_my_rr_sets_cname_exclusivity(self):
        self.create_rr_set(self.my_domain, ["1.2.3.4"], type="A", ttl=3600, subname="a")
        self.create_rr_set(
            self.my_domain, ["example.com."], type="CNAME", ttl=3600, subname="cname"
        )

        # Can't add a CNAME where something else is
        data = {
            "subname": "a",
            "ttl": 3600,
            "type": "CNAME",
            "records": ["foobar.com."],
        }
        response = self.client.post_rr_set(self.my_domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

        # Can't add something else where a CNAME is
        data = {"subname": "cname", "ttl": 3600, "type": "A", "records": ["4.3.2.1"]}
        response = self.client.post_rr_set(self.my_domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_create_my_rr_sets_without_records(self):
        for subname in [
            "",
            "create-my-rr-sets",
            "foo.create-my-rr-sets",
            "bar.baz.foo.create-my-rr-sets",
        ]:
            for data in [
                {"subname": subname, "records": [], "ttl": 60, "type": "A"},
                {"subname": subname, "ttl": 60, "type": "A"},
            ]:
                response = self.client.post_rr_set(self.my_empty_domain.name, **data)
                self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

                response = self.client.get_rr_sets(self.my_empty_domain.name)
                self.assertStatus(response, status.HTTP_200_OK)
                self.assertRRSetsCount(response.data, [], count=0)

    def test_create_other_rr_sets(self):
        data = {"records": ["1.2.3.4"], "ttl": 60, "type": "A"}
        response = self.client.post_rr_set(self.other_domain.name, **data)
        self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    @staticmethod
    def _create_test_txt_record(record, type_="TXT"):
        return {
            "records": [f"{record}"],
            "ttl": 3600,
            "type": type_,
            "subname": f"name{len(record)}",
        }

    def test_create_my_rr_sets_chunk_too_long(self):
        for l, t in product([1, 255, 256, 498], ["TXT", "SPF"]):
            with self.assertRequests(
                self.requests_desec_rr_sets_update(self.my_empty_domain.name)
            ):
                response = self.client.post_rr_set(
                    self.my_empty_domain.name,
                    **self._create_test_txt_record(f'"{"A" * l}"', t),
                )
                self.assertStatus(response, status.HTTP_201_CREATED)
            with self.assertRequests(
                self.requests_desec_rr_sets_update(self.my_empty_domain.name)
            ):
                self.client.delete_rr_set(
                    self.my_empty_domain.name, type_=t, subname=f"name{l+2}"
                )

    def test_create_my_rr_sets_too_long_content(self):
        def token(length):
            if length == 0:
                return ""
            if length <= 255:
                return f'"{"A" * length}"'
            return f"{token(255)} " * (length // 255) + token(length % 255)

        def p2w_length(length):
            return ceil(length / 255 * 256)

        def w2p_length(length):
            return floor(length / 256 * 255)

        max_wirelength = 64000
        max_preslength = w2p_length(max_wirelength)

        assert max_preslength == 63750
        assert p2w_length(max_preslength) == 64000
        assert p2w_length(max_preslength + 1) == 64002

        for t in ["SPF", "TXT"]:
            response = self.client.post_rr_set(
                self.my_empty_domain.name,
                # record of wire length 501 bytes in chunks of max 255 each (RFC 4408)
                **self._create_test_txt_record(token(max_preslength + 1), t),
            )
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
            self.assertIn(
                f"Ensure this value has no more than {max_wirelength} byte in wire format (it has {p2w_length(max_preslength + 1)}).",
                str(response.data),
            )

        with self.assertRequests(
            self.requests_desec_rr_sets_update(self.my_empty_domain.name)
        ):
            response = self.client.post_rr_set(
                self.my_empty_domain.name,
                # record of wire length 500 bytes in chunks of max 255 each (RFC 4408)
                **self._create_test_txt_record(token(max_preslength)),
            )
            self.assertStatus(response, status.HTTP_201_CREATED)

    def test_create_my_rr_sets_too_large_rrset(self):
        network = IPv4Network("127.0.0.0/20")  # size: 4096 IP addresses
        data = {
            "records": [str(ip) for ip in network],
            "ttl": 3600,
            "type": "A",
            "subname": "name",
        }
        response = self.client.post_rr_set(self.my_empty_domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        excess_length = 28743 + len(self.my_empty_domain.name)
        self.assertIn(
            f"Total length of RRset exceeds limit by {excess_length} bytes.",
            str(response.data),
        )

    def test_create_my_rr_sets_twice(self):
        data = {"records": ["1.2.3.4"], "ttl": 3660, "type": "A"}
        with self.assertRequests(
            self.requests_desec_rr_sets_update(self.my_empty_domain.name)
        ):
            response = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertStatus(response, status.HTTP_201_CREATED)

        data["records"][0] = "3.2.2.1"
        response = self.client.post_rr_set(self.my_empty_domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_create_my_rr_sets_duplicate_content(self):
        for records in [
            ["::1", "0::1"],
            # TODO add more examples
        ]:
            data = {"records": records, "ttl": 3660, "type": "AAAA"}
            response = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertContains(
                response, "Duplicate", status_code=status.HTTP_400_BAD_REQUEST
            )

    def test_create_my_rr_sets_upper_case(self):
        for subname in ["asdF", "cAse", "asdf.FOO", "--F", "ALLCAPS"]:
            data = {"records": ["1.2.3.4"], "ttl": 60, "type": "A", "subname": subname}
            response = self.client.post_rr_set(self.my_empty_domain.name, **data)
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
            self.assertIn("Subname can only use (lowercase)", str(response.data))

    def test_create_my_rr_sets_subname_too_long(self):
        name = (
            "9.1.f.1.0.6.1.d.f.2.2.b.9.1.6.3.c.2.f.7.9.5.2.4.7.8.3.2.6.6.c.6.9.3.5.6.9.2.4.d.c.e.a.e.f.1.8"
            ".b.a.c.9.0.6.2.c.b.c.1.6.3.8.2.7.9.0.5.2.a.c.f.f.2.6.a.c.3.c.e.3.0.6.1.8.0.7.4.0.1.0.0.2.ip6.test"
        )
        domain = self.create_domain(name=name, owner=self.owner)

        subname = (
            "e.8.c.f.e.0.9.f.4.9.1.f.1.0.6.1.d.f.2.2.b.9.1.6.3.c.9.3.5.6.9.2.4.d.c.e.a.e.f.1.8.b.a.c.9.0"
            ".6.2.c.b.c.1.6.3.8.2.6.7.9.0.5.2.a.c.f.f.2.6"
        )
        data = {"subname": subname, "records": ["1.2.3.4"], "ttl": 3600, "type": "A"}
        response = self.client.post_rr_set(domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(response.data["subname"][0].code, "name_too_long")

    def test_create_my_rr_sets_subname_too_many_dots(self):
        for subname in ["dottest.", ".dottest", "dot..test"]:
            data = {
                "subname": subname,
                "records": ["10 example.com."],
                "ttl": 3600,
                "type": "MX",
            }
            response = self.client.post_rr_set(self.my_domain.name, **data)
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

        response = self.client.get_rr_sets(self.my_domain.name)
        self.assertStatus(response, status.HTTP_200_OK)
        self.assertRRSetsCount(response.data, [data], count=0)

    def test_create_my_rr_sets_empty_payload(self):
        response = self.client.post_rr_set(self.my_empty_domain.name)
        self.assertContains(
            response, "No data provided", status_code=status.HTTP_400_BAD_REQUEST
        )

    def test_create_my_rr_sets_cname_two_records(self):
        data = {
            "subname": "sub",
            "records": ["example.com.", "example.org."],
            "ttl": 3600,
            "type": "CNAME",
        }
        response = self.client.post_rr_set(self.my_domain.name, **data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_create_my_rr_sets_canonical_content(self):
        # TODO fill in more examples
        datas = [
            # record type: (non-canonical input, canonical output expectation)
            ("A", ("127.0.0.1", "127.0.0.1")),
            ("AAAA", ("0000::0000:0001", "::1")),
            ("AAAA", ("::ffff:127.0.0.1", "::ffff:7f00:1")),
            ("AAAA", ("2001:db8::128.2.129.4", "2001:db8::8002:8104")),
            ("AFSDB", ("02 turquoise.FEMTO.edu.", "2 turquoise.femto.edu.")),
            (
                "APL",
                (
                    "2:FF00:0:0:0:0::/8  !1:192.168.38.0/28",
                    "2:ff00::/8 !1:192.168.38.0/28",
                ),
            ),
            ("CAA", ('0128 "issue" "letsencrypt.org"', '128 issue "letsencrypt.org"')),
            (
                "CDNSKEY",
                (
                    "0256  03  08  AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+ 1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mx t6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLK l3D0L/cD",
                    "256 3 8 AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mxt6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLKl3D0L/cD",
                ),
            ),
            (
                "CDNSKEY",
                (
                    "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/ qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGr CHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll 96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPri ec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAst bxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6as lO7jXv16Gws=",
                    "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGrCHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPriec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAstbxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6aslO7jXv16Gws=",
                ),
            ),
            (
                "CDNSKEY",
                (
                    "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
                    "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryquB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
                ),
            ),
            ("CDNSKEY", ("0 3 0 AA==", "0 3 0 AA==")),
            (
                "CDS",
                (
                    "047883  013  02  43BD262211B2A748335149408F67BC95B9A4A3174FD86E6A83830380 446E7AFD",
                    "47883 13 2 43BD262211B2A748335149408F67BC95B9A4A3174FD86E6A83830380446E7AFD".lower(),
                ),
            ),
            ("CDS", ("0 0 0 00", "0 0 0 00")),
            ("CERT", ("04 257 RSASHA256 sadfdd==", "4 257 8 sadfdQ==")),
            ("CNAME", ("EXAMPLE.COM.", "example.com.")),
            ("CSYNC", ("066 03  NS  AAAA A", "66 3 A NS AAAA")),
            ("DHCID", ("xxxx", "xxxx")),
            (
                "DLV",
                (
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA10DF1F520".lower(),
                ),
            ),
            (
                "DLV",
                (
                    "6454 8 2 5C BA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA10DF1F520".lower(),
                ),
            ),
            ("DNAME", ("EXAMPLE.COM.", "example.com.")),
            (
                "DNSKEY",
                (
                    "0256  03  08  AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+ 1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mx t6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLK l3D0L/cD",
                    "256 3 8 AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mxt6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLKl3D0L/cD",
                ),
            ),
            (
                "DNSKEY",
                (
                    "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/ qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGr CHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll 96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPri ec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAst bxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6as lO7jXv16Gws=",
                    "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGrCHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPriec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAstbxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6aslO7jXv16Gws=",
                ),
            ),
            (
                "DNSKEY",
                (
                    "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
                    "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryquB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
                ),
            ),
            (
                "DS",
                (
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA10DF1F520".lower(),
                ),
            ),
            (
                "DS",
                (
                    "6454 8 2 5C BA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                    "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA10DF1F520".lower(),
                ),
            ),
            ("EUI48", ("AA-BB-CC-DD-EE-FF", "aa-bb-cc-dd-ee-ff")),
            ("EUI64", ("AA-BB-CC-DD-EE-FF-aa-aa", "aa-bb-cc-dd-ee-ff-aa-aa")),
            ("HINFO", ("cpu os", '"cpu" "os"')),
            ("HINFO", ('"cpu" "os"', '"cpu" "os"')),
            (
                "HTTPS",
                ("01 h3POOL.exaMPLe. aLPn=h2,h3", "1 h3POOL.exaMPLe. alpn=h2,h3"),
            ),
            (
                "HTTPS",
                (
                    "01 h3POOL.exaMPLe. aLPn=h2,h3 ECH=MTIzLi4uCg==",
                    '1 h3POOL.exaMPLe. alpn=h2,h3 ech="MTIzLi4uCg=="',
                ),
            ),
            # ('IPSECKEY', ('01 00 02 . ASDFAF==', '1 0 2 . ASDFAA==')),
            # ('IPSECKEY', ('01 00 02 . 000000==', '1 0 2 . 00000w==')),
            ("KX", ("010 example.com.", "10 example.com.")),
            ("L32", ("010  10.1.2.0", "10 10.1.2.0")),
            ("L64", ("010   2001:0Db8:2140:2000", "10 2001:0db8:2140:2000")),
            (
                "LOC",
                (
                    "023 012 59 N 042 022 48.500 W 65.00m 20.00m 10.00m 10.00m",
                    "23 12 59.000 N 42 22 48.500 W 65.00m 20.00m 10.00m 10.00m",
                ),
            ),
            ("LP", ("010   l64-subnet1.example.com.", "10 l64-subnet1.example.com.")),
            ("MX", ("10 010.1.1.1.", "10 010.1.1.1.")),
            ("MX", ("010 010.1.1.2.", "10 010.1.1.2.")),
            ("MX", ("0 .", "0 .")),
            (
                "NAPTR",
                (
                    '100  50  "s"  "z3950+I2L+I2C"     ""  _z3950._tcp.gatech.edu.',
                    '100 50 "s" "z3950+I2L+I2C" "" _z3950._tcp.gatech.edu.',
                ),
            ),
            ("NID", ("010 0014:4fff:ff20:Ee64", "10 0014:4fff:ff20:ee64")),
            ("NS", ("EXaMPLE.COM.", "example.com.")),
            (
                "OPENPGPKEY",
                (
                    "mG8EXtVIsRMFK4EEACIDAwQSZPNqE4tS xLFJYhX+uabSgMrhOqUizJhkLx82",
                    "mG8EXtVIsRMFK4EEACIDAwQSZPNqE4tSxLFJYhX+uabSgMrhOqUizJhkLx82",
                ),
            ),
            ("PTR", ("EXAMPLE.COM.", "example.com.")),
            ("RP", ("hostmaster.EXAMPLE.com. .", "hostmaster.example.com. .")),
            ("SMIMEA", ("3 01 0 aaBBccddeeff", "3 1 0 aabbccddeeff")),
            (
                "SPF",
                (
                    '"v=spf1 ip4:10.1" ".1.1 ip4:127" ".0.0.0/16 ip4:192.168.0.0/27 include:example.com -all"',
                    '"v=spf1 ip4:10.1" ".1.1 ip4:127" ".0.0.0/16 ip4:192.168.0.0/27 include:example.com -all"',
                ),
            ),
            ("SPF", ('"foo" "bar"', '"foo" "bar"')),
            ("SPF", ('"foobar"', '"foobar"')),
            ("SRV", ("0 000 0 .", "0 0 0 .")),
            ("SRV", ("100 1 5061 EXAMPLE.com.", "100 1 5061 example.com.")),
            ("SRV", ("100 1 5061 example.com.", "100 1 5061 example.com.")),
            ("SSHFP", ("2 2 aabbccEEddff", "2 2 aabbcceeddff")),
            (
                "SVCB",
                (
                    "2 sVc2.example.NET. IPV6hint=2001:db8:00:0::2 port=01234",
                    "2 sVc2.example.NET. port=1234 ipv6hint=2001:db8::2",
                ),
            ),
            (
                "SVCB",
                (
                    "2 sVc2.example.NET. ECH=MjIyLi4uCg== IPV6hint=2001:db8:00:0::2 port=01234",
                    '2 sVc2.example.NET. port=1234 ech="MjIyLi4uCg==" ipv6hint=2001:db8::2',
                ),
            ),
            (
                "TLSA",
                (
                    "3 0001 1 000AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
                    "3 1 1 000aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                ),
            ),
            (
                "TLSA",
                (
                    "003 00 002 696B8F6B92A913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
                    "3 0 2 696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
                ),
            ),
            (
                "TLSA",
                (
                    "3 0 2 696B8F6B92A913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd696B8F6B92A913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
                    "3 0 2 696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
                ),
            ),
            ("TXT", ('"foo" "bar"', '"foo" "bar"')),
            ("TXT", ('"foobar"', '"foobar"')),
            ("TXT", ('"foo" "" "bar"', '"foo" "" "bar"')),
            ("TXT", ('"" "" "foo" "" "bar"', '"" "" "foo" "" "bar"')),
            (
                "TXT",
                (
                    r'"\130\164name\164Boss\164type\1611"',
                    r'"\130\164name\164Boss\164type\1611"',
                ),
            ),
            (
                "TXT",
                (
                    '"' + "".join(rf"\{n:03}" for n in range(256)) + '"',  # all bytes
                    r'"\000\001\002\003\004\005\006\007\008\009\010\011\012\013\014\015\016\017\018\019\020\021\022\023\024\025\026\027\028\029\030\031 !\"#$%&'
                    + "'"
                    + r'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~\127\128\129\130\131\132\133\134\135\136\137\138\139\140\141\142\143\144\145\146\147\148\149\150\151\152\153\154\155\156\157\158\159\160\161\162\163\164\165\166\167\168\169\170\171\172\173\174\175\176\177\178\179\180\181\182\183\184\185\186\187\188\189\190\191\192\193\194\195\196\197\198\199\200\201\202\203\204\205\206\207\208\209\210\211\212\213\214\215\216\217\218\219\220\221\222\223\224\225\226\227\228\229\230\231\232\233\234\235\236\237\238\239\240\241\242\243\244\245\246\247\248\249\250\251\252\253\254" "\255"',
                ),
            ),
            (
                "URI",
                (
                    '10 01 "ftp://ftp1.example.com/public"',
                    '10 1 "ftp://ftp1.example.com/public"',
                ),
            ),
        ]
        for t, (record, canonical_record) in datas:
            if not record:
                continue
            subname = "" if t == "DNSKEY" else "test"
            data = {"records": [record], "ttl": 3660, "type": t, "subname": subname}
            with self.assertRequests(
                self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
            ):
                response = self.client.post_rr_set(self.my_empty_domain.name, **data)
                self.assertStatus(response, status.HTTP_201_CREATED)
                self.assertEqual(
                    canonical_record,
                    response.data["records"][0],
                    f"For RR set type {t}, expected '{canonical_record}' to be the canonical form of "
                    f'\'{record}\', but saw \'{response.data["records"][0]}\'.',
                )
            with self.assertRequests(
                self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
            ):
                response = self.client.delete_rr_set(
                    self.my_empty_domain.name, subname=subname, type_=t
                )
                self.assertStatus(response, status.HTTP_204_NO_CONTENT)
        self.assertAllSupportedRRSetTypes(set(t for t, _ in datas))

    def test_create_my_rr_sets_known_type_benign(self):
        # TODO fill in more examples
        datas = {
            "A": ["127.0.0.1", "127.0.0.2"],
            "AAAA": ["::1", "::2"],
            "AFSDB": ["2 turquoise.femto.edu."],
            "APL": [
                # from RFC 3123 Sec. 4
                "1:192.168.32.0/21 !1:192.168.38.0/28",
                "1:192.168.42.0/26 1:192.168.42.64/26 1:192.168.42.128/25",
                "1:127.0.0.1/32 1:172.16.64.0/22",
                "1:224.0.0.0/4  2:FF00:0:0:0:0:0:0:0/8",
                # made-up (not from RFC)
                "1:1.2.3.4/32 2:::/128",
                "2:FF00::/8 !1:192.168.38.0/28",
            ],
            "CAA": [
                '128 issue "letsencrypt.org"',
                '128 iodef "mailto:desec@example.com"',
                '1 issue "letsencrypt.org"',
            ],
            "CERT": [
                "06 0 0 sadfdd==",
                "IPGP 0 0 sadfdd==",
                "4 257 RSASHA256 sadfdd==",
            ],
            "CDNSKEY": [
                "256 3 8 AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+ 1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mx t6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLK l3D0L/cD",
                "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/ qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGr CHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll 96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPri ec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAst bxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6as lO7jXv16Gws=",
                "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
            ],
            "CDS": [
                "6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                "62703 13 2 085BF1EE0ADBBC99D4D9328229EBDCAEC5FAB20E38610072AD055474 4C7AF4A0",
                "61655 13 4 C838A5C66FCBF83B8B6B50C3CEEC3524777FE4AF8A9FE0172ECAD242 48B0CA1A216DD0D538F20C130DD3059538204B04",
                "6454 8 5 24396E17E36D031F71C354B06A979A67A01F503E",
            ],
            "CNAME": ["example.com.", "*._under-score.-foo_bar.example.net.", "."],
            "CSYNC": ["0 0", "66 1 A", "66 2 AAAA", "66 3 A NS AAAA", "66 15 NSEC"],
            "DHCID": ["aaaaaaaaaaaa", "aa aaa  aaaa a a a"],
            "DLV": [
                "6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                "62703 13 2 085BF1EE0ADBBC99D4D9328229EBDCAEC5FAB20E38610072AD055474 4C7AF4A0",
                "61655 13 4 C838A5C66FCBF83B8B6B50C3CEEC3524777FE4AF8A9FE0172ECAD242 48B0CA1A216DD0D538F20C130DD3059538204B04",
                "6454 8 5 24396E17E36D031F71C354B06A979A67A01F503E",
            ],
            "DNAME": ["example.com."],
            "DNSKEY": [
                "256 3 8 AwEAAday3UX323uVzQqtOMQ7EHQYfD5Ofv4akjQGN2zY5AgB/2jmdR/+ 1PvXFqzKCAGJv4wjABEBNWLLFm7ew1hHMDZEKVL17aml0EBKI6Dsz6Mx t6n7ScvLtHaFRKaxT4i2JxiuVhKdQR9XGMiWAPQKrRM5SLG0P+2F+TLK l3D0L/cD",
                "257 3 8 AwEAAcw5QLr0IjC0wKbGoBPQv4qmeqHy9mvL5qGQTuaG5TSrNqEAR6b/ qvxDx6my4JmEmjUPA1JeEI9YfTUieMr2UZflu7aIbZFLw0vqiYrywCGr CHXLalOrEOmrvAxLvq4vHtuTlH7JIszzYBSes8g1vle6KG7xXiP3U5Ll 96Qiu6bZ31rlMQSPB20xbqJJh6psNSrQs41QvdcXAej+K2Hl1Wd8kPri ec4AgiBEh8sk5Pp8W9ROLQ7PcbqqttFaW2m7N/Wy4qcFU13roWKDEAst bxH5CHPoBfZSbIwK4KM6BK/uDHpSPIbiOvOCW+lvu9TAiZPc0oysY6as lO7jXv16Gws=",
                "257 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg==",
            ],
            "DS": [
                "6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 2 5CBA665A006F6487625C6218522F09BD3673C25FA10F25CB18459AA1 0DF1F520",
                "62703 13 2 085BF1EE0ADBBC99D4D9328229EBDCAEC5FAB20E38610072AD055474 4C7AF4A0",
                "61655 13 4 C838A5C66FCBF83B8B6B50C3CEEC3524777FE4AF8A9FE0172ECAD242 48B0CA1A216DD0D538F20C130DD3059538204B04",
                "6454 8 5 24396E17E36D031F71C354B06A979A67A01F503E",
            ],
            "EUI48": ["aa-bb-cc-dd-ee-ff", "AA-BB-CC-DD-EE-FF"],
            "EUI64": ["aa-bb-cc-dd-ee-ff-00-11", "AA-BB-CC-DD-EE-FF-00-11"],
            "HINFO": ['"ARMv8-A" "Linux"'],
            "HTTPS": [
                # from https://www.ietf.org/archive/id/draft-ietf-dnsop-svcb-https-06.html#name-examples, with ech base64'd
                "1 . alpn=h3",
                "0 pool.svc.example.",
                '1 h3pool.example. alpn=h2,h3 ech="MTIzLi4uCg=="',
                '2 .      alpn=h2 ech="YWJjLi4uCg=="',
                # made-up (not from RFC)
                "1 pool.svc.example. no-default-alpn alpn=h2 port=1234 ipv4hint=192.168.123.1",
                "2 . ech=... key65333=ex1 key65444=ex2 mandatory=key65444,ech",  # see #section-7
            ],
            # 'IPSECKEY': [
            #     '12 0 2 . asdfdf==',
            #     '03 1 1 127.0.0.1 asdfdf==',
            #     '10 02 2 bade::affe AQNRU3mG7TVTO2BkR47usntb102uFJtugbo6BSGvgqt4AQ==',
            #     '12 3 01 example.com. asdfdf==',
            # ],
            "KX": ["4 example.com.", "28 io."],
            "L32": ["010   10.1.2.0", "65535 1.2.3.4"],
            "L64": ["010   2001:0DB8:1140:1000", "10 2001:0DB8:1140:1000"],
            "LOC": ["23 12 59.000 N 42 22 48.500 W 65.00m 20.00m 10.00m 10.00m"],
            "LP": ["10 l64-subnet1.example.com.", "65535 ."],
            "MX": ["10 example.com.", "20 1.1.1.1."],
            "NAPTR": ['100  50  "s"  "z3950+I2L+I2C"     ""  _z3950._tcp.gatech.edu.'],
            "NID": ["65535   0014:4fff:ff20:ee64"],
            "NS": ["ns1.example.com."],
            "OPENPGPKEY": [
                "mG8EXtVIsRMFK4EEACIDAwQSZPNqE4tSxLFJYhX+uabSgMrhOqUizJhkLx82",  # key incomplete
                "YWFh\xf0\x9f\x92\xa9YWFh",  # valid as non-alphabet bytes will be ignored
            ],
            "PTR": ["example.com.", "*.example.com."],
            "RP": ["hostmaster.example.com. ."],
            "SMIMEA": ["3 1 0 aabbccddeeff"],
            "SPF": [
                '"v=spf1 include:example.com ~all"',
                '"v=spf1 ip4:10.1.1.1 ip4:127.0.0.0/16 ip4:192.168.0.0/27 include:example.com -all"',
                '"spf2.0/pra,mfrom ip6:2001:558:fe14:76:68:87:28:0/120 -all"',
            ],
            "SRV": ["0 0 0 .", "100 1 5061 example.com."],
            "SSHFP": ["2 2 aabbcceeddff"],
            "SVCB": [
                "0 svc4-baz.example.net.",
                "1 . key65333=...",
                '2 svc2.example.net. ech="MjIyLi4uCg==" ipv6hint=2001:db8::2 port=1234',
            ],
            "TLSA": [
                "3 1 1 AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
                "3 0 2 696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
                "3 0 2 696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd696b8f6b92a913560b23ef5720c378881faffe74432d04eb35db957c0a93987b47adf26abb5dac10ba482597ae16edb069b511bec3e26010d1927bf6392760dd",
            ],
            "TXT": [
                '"foobar"',
                '"foo" "bar"',
                '"“红色联合”对“四·二八兵团”总部大楼的攻击已持续了两天"',
                '"new\\010line"'
                '"🧥 👚 👕 👖 👔 👗 👙 👘 👠 👡 👢 👞 👟 🥾 🥿  🧦 🧤 🧣 🎩 🧢 👒 🎓 ⛑ 👑 👝 👛 👜 💼 🎒 👓 🕶 🥽 🥼 🌂 🧵"',
            ],
            "URI": ['10 1 "ftp://ftp1.example.com/public"'],
        }
        self.assertAllSupportedRRSetTypes(set(datas.keys()))
        for t, records in datas.items():
            subname = "" if t == "DNSKEY" else "test"
            for r in records:
                data = {"records": [r], "ttl": 3660, "type": t, "subname": subname}
                with self.assertRequests(
                    self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
                ):
                    response = self.client.post_rr_set(
                        self.my_empty_domain.name, **data
                    )
                    self.assertStatus(response, status.HTTP_201_CREATED)
                with self.assertRequests(
                    self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
                ):
                    response = self.client.delete_rr_set(
                        self.my_empty_domain.name, subname=subname, type_=t
                    )
                    self.assertStatus(response, status.HTTP_204_NO_CONTENT)

    def test_create_my_rr_sets_known_type_invalid(self):
        # TODO fill in more examples
        datas = {
            # recordtype: [list of examples expected to be rejected, individually]
            "A": [
                "127.0.0.999",
                "127.000.0.01",
                "127.0.0.256",
                "::1",
                "foobar",
                "10.0.1",
                "10!",
            ],
            "AAAA": ["::g", "1:1:1:1:1:1:1:1:", "1:1:1:1:1:1:1:1:1"],
            "AFSDB": ["example.com.", "1 1", "1 de"],
            "APL": [
                "0:192.168.32.0/21 !1:192.168.38.0/28",
                "1:192.168.32.0/21 !!1:192.168.38.0/28",
                "1:192.168.32.0/33",
                "18:12345/2",
                "1:127.0.0.1",
                "2:FF00:0:0:0:0:0:0:0:0/8" "2:::/129",
            ],
            "CAA": ['43235 issue "letsencrypt.org"'],
            "CERT": ["6 0 sadfdd=="],
            "CDNSKEY": [
                "a 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 b 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 3 c aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 3 13 d",
                "0 3 0 0",
            ],
            "CDS": [
                "a 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "-6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 b 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 c 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 d",
                "6454 8 0 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 aabbccddeeff",
                "0 0 0 0",
            ],
            "CNAME": ["example.com", "10 example.com.", "@.", "abcd123." * 32],
            "CSYNC": ["0 -1 A", "444 65536 A", "0 3 AAA"],
            "DHCID": ["x", "xx", "xxx"],
            "DLV": [
                "a 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "-6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 b 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 c 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 d",
                "6454 8 0 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 aabbccddeeff",
            ],
            "DNAME": ["example.com", "10 example.com."],
            "DNSKEY": [
                "a 3 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 b 13 aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 3 c aCoEWYBBVsP9Fek2oC8yqU8ocKmnS1iDSFZNORnQuHKtJ9Wpyz+kNryq uB78Pyk/NTEoai5bxoipVQQXzHlzyg=="
                "257 3 13 d"
            ],
            "DS": [
                "a 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "-6454 8 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 b 1 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 c 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 d",
                "6454 8 0 24396E17E36D031F71C354B06A979A67A01F503E",
                "6454 8 1 aabbccddeeff",
            ],
            "EUI48": ["aa-bb-ccdd-ee-ff", "AA-BB-CC-DD-EE-GG"],
            "EUI64": ["aa-bb-cc-dd-ee-ff-gg-11", "AA-BB-C C-DD-EE-FF-00-11"],
            "HINFO": ['"ARMv8-A"', f'"a" "{"b"*256}"'],
            "HTTPS": [
                # from https://tools.ietf.org/html/draft-ietf-dnsop-svcb-https-02#section-10.3, with ech base64'd
                '1 h3pool alpn=h2,h3 ech="MTIzLi4uCg=="',
                # made-up (not from RFC)
                "0 pool.svc.example. no-default-alpn port=1234 ipv4hint=192.168.123.1",  # no keys in alias mode
                "1 pool.svc.example. no-default-alpn port=1234 ipv4hint=192.168.123.1 ipv4hint=192.168.123.2",  # dup
            ],
            # 'IPSECKEY': [],
            "KX": ["-1 example.com", "10 example.com"],
            "L32": ["65536 10.1.2.0", "5 a.1.2.0", "10 10.1.02.0"],
            "L64": ["65536 2001:0DB8:4140:4000", "5 01:0DB8:4140:4000"],
            "LOC": [
                "23 12 61.000 N 42 22 48.500 W 65.00m 20.00m 10.00m 10.00m",
                "foo",
                "1.1.1.1",
            ],
            "LP": [
                "10 l64-subnet1.example.com",
                "-3 l64-subnet1.example.com.",
                "65536 l64-subnet1.example.com.",
            ],
            "MX": [
                "10 example.com",
                "example.com.",
                "-5 asdf.",
                "65537 asdf.",
                "10 _foo.example.com.",
                "10 $url.",
            ],
            "NAPTR": [
                '100  50  "s"  "z3950+I2L+I2C"     ""  _z3950._tcp.gatech.edu',
                '100  50  "s"     ""  _z3950._tcp.gatech.edu.',
                '100  50  3 2  "z3950+I2L+I2C"     ""  _z3950._tcp.gatech.edu.',
            ],
            "NID": ["010 14:4fff:ff20:Ee64", "d 0014:4fff:ff20:ee64", "20 ::14::ee64"],
            "NS": ["ns1.example.com", "127.0.0.1", "_foobar.example.dedyn.io.", "."],
            "OPENPGPKEY": ["1 2 3"],
            "PTR": ['"example.com."', "10 *.example.com."],
            "RP": ["hostmaster.example.com.", "10 foo."],
            "SMIMEA": ["3 1 0 aGVsbG8gd29ybGQh"],
            "SPF": ['"v=spf1', "v=spf1 include:example.com ~all"],
            "SRV": [
                "0 0 0 0",
                "100 5061 example.com.",
                "0 0 16920 _foo.example.com.",
                "0 0 16920 $url.",
            ],
            "SSHFP": ["aabbcceeddff"],
            "SVCB": [
                "0 svc4-baz.example.net. keys=val",
                "1 not.fully.qualified key65333=...",
                '2 duplicate.key. ech="MjIyLi4uCg==" ech="MjIyLi4uCg=="',
            ],
            "TLSA": ["3 1 1 AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"],
            "TXT": [
                'foob"ar',
                "v=spf1 include:example.com ~all",
                '"foo\nbar"',
                '"\x00" "NUL byte yo"',
                '"'
                + "".join(rf"\{n:03}" for n in range(257))
                + '"',  # \256 does not exist
            ],
            "URI": ['"1" "2" "3"'],
        }
        self.assertAllSupportedRRSetTypes(set(datas.keys()))
        for t, records in datas.items():
            for r in records:
                data = {"records": [r], "ttl": 3660, "type": t, "subname": "subname"}
                response = self.client.post_rr_set(self.my_empty_domain.name, **data)
                self.assertNotContains(
                    response, "Duplicate", status_code=status.HTTP_400_BAD_REQUEST
                )

    def test_create_my_rr_sets_no_ip_block_unless_lps(self):
        # IP block should not be effective unless domain is under Local Public Suffix
        BlockedSubnet.from_ip("3.2.2.3").save()
        with self.assertRequests(
            self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
        ):
            response = self.client.post_rr_set(
                self.my_empty_domain.name,
                records=["3.2.2.5"],
                ttl=3660,
                subname="blocktest",
                type="A",
            )
            self.assertStatus(response, status.HTTP_201_CREATED)

    def test_create_my_rr_sets_txt_splitting(self):
        for t in ["TXT", "SPF"]:
            for l in [200, 255, 256, 300, 400]:
                data = {
                    "records": [f'"{"a"*l}"'],
                    "ttl": 3660,
                    "type": t,
                    "subname": f"x{l}",
                }
                with self.assertRequests(
                    self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
                ):
                    response = self.client.post_rr_set(
                        self.my_empty_domain.name, **data
                    )
                    self.assertStatus(response, status.HTTP_201_CREATED)
                response = self.client.get_rr_set(self.my_empty_domain.name, f"x{l}", t)
                num_tokens = response.data["records"][0].count(" ") + 1
                num_tokens_expected = l // 256 + 1
                self.assertEqual(
                    num_tokens,
                    num_tokens_expected,
                    f"For a {t} record with a token of length of {l}, expected to see "
                    f"{num_tokens_expected} tokens in the canonical format, but saw {num_tokens}.",
                )
                self.assertEqual(
                    "".join(r.strip('" ') for r in response.data["records"][0]), "a" * l
                )

    def test_create_my_rr_sets_unknown_type(self):
        for _type in ["AA", "ASDF"] + list(
            RR_SET_TYPES_AUTOMATIC | RR_SET_TYPES_UNSUPPORTED
        ):
            response = self.client.post_rr_set(
                self.my_domain.name, records=["1234"], ttl=3660, type=_type
            )
            self.assertContains(
                response,
                text=(
                    "managed automatically"
                    if _type in RR_SET_TYPES_AUTOMATIC
                    else "type is currently unsupported"
                ),
                status_code=status.HTTP_400_BAD_REQUEST,
            )

    def test_create_my_rr_sets_ttl_too_small(self):
        ttl = settings.MINIMUM_TTL_DEFAULT - 1
        response = self.client.post_rr_set(
            self.my_empty_domain.name, records=["1.2.3.4"], ttl=ttl, type="A"
        )
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        detail = f"Ensure this value is greater than or equal to {self.my_empty_domain.minimum_ttl}."
        self.assertEqual(response.data["ttl"][0], detail)

        ttl += 1
        with self.assertRequests(
            self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
        ):
            response = self.client.post_rr_set(
                self.my_empty_domain.name, records=["1.2.23.4"], ttl=ttl, type="A"
            )
        self.assertStatus(response, status.HTTP_201_CREATED)

    def test_create_my_rr_sets_ttl_too_large(self):
        max_ttl = 24 * 3600
        response = self.client.post_rr_set(
            self.my_empty_domain.name, records=["1.2.3.4"], ttl=max_ttl + 1, type="A"
        )
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        detail = f"Ensure this value is less than or equal to {max_ttl}."
        self.assertEqual(response.data["ttl"][0], detail)

        with self.assertRequests(
            self.requests_desec_rr_sets_update(name=self.my_empty_domain.name)
        ):
            response = self.client.post_rr_set(
                self.my_empty_domain.name, records=["1.2.23.4"], ttl=max_ttl, type="A"
            )
        self.assertStatus(response, status.HTTP_201_CREATED)

    def test_retrieve_my_rr_sets_apex(self):
        response = self.client.get_rr_set(
            self.my_rr_set_domain.name, subname="", type_="A"
        )
        self.assertStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data["records"][0], "1.2.3.4")
        self.assertEqual(response.data["ttl"], 3620)

    def test_retrieve_my_rr_sets_restricted_types(self):
        for type_ in self.AUTOMATIC_TYPES:
            response = self.client.get_rr_sets(self.my_domain.name, type=type_)
            self.assertStatus(response, status.HTTP_403_FORBIDDEN)
            response = self.client.get_rr_sets(
                self.my_domain.name, type=type_, subname=""
            )
            self.assertStatus(response, status.HTTP_403_FORBIDDEN)

    def test_update_my_rr_sets(self):
        for subname in self.SUBNAMES:
            with self.assertRequests(
                self.requests_desec_rr_sets_update(name=self.my_rr_set_domain.name)
            ):
                data = {
                    "records": ["2.2.3.4"],
                    "ttl": 3630,
                    "type": "A",
                    "subname": subname,
                }
                response = self.client.put_rr_set(
                    self.my_rr_set_domain.name, subname, "A", data
                )
                self.assertStatus(response, status.HTTP_200_OK)

            response = self.client.get_rr_set(self.my_rr_set_domain.name, subname, "A")
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertEqual(response.data["records"], ["2.2.3.4"])
            self.assertEqual(response.data["ttl"], 3630)

            response = self.client.put_rr_set(
                self.my_rr_set_domain.name, subname, "A", {"records": ["2.2.3.5"]}
            )
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

            response = self.client.put_rr_set(
                self.my_rr_set_domain.name, subname, "A", {"ttl": 3637}
            )
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)

    def test_update_my_rr_set_with_invalid_payload_type(self):
        for subname in self.SUBNAMES:
            data = [
                {"records": ["2.2.3.4"], "ttl": 30, "type": "A", "subname": subname}
            ]
            response = self.client.put_rr_set(
                self.my_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
            self.assertEqual(
                response.data["non_field_errors"][0],
                "Invalid data. Expected a dictionary, but got list.",
            )

            data = "foobar"
            response = self.client.put_rr_set(
                self.my_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
            self.assertEqual(
                response.data["non_field_errors"][0],
                "Invalid data. Expected a dictionary, but got str.",
            )

    def test_partially_update_my_rr_sets(self):
        for subname in self.SUBNAMES:
            current_rr_set = self.client.get_rr_set(
                self.my_rr_set_domain.name, subname, "A"
            ).data
            for data in [
                {"records": ["2.2.3.4"], "ttl": 3630},
                {"records": ["3.2.3.4"]},
                {"records": ["3.2.3.4", "9.8.8.7"]},
                {"ttl": 3637},
            ]:
                with self.assertRequests(
                    self.requests_desec_rr_sets_update(name=self.my_rr_set_domain.name)
                ):
                    response = self.client.patch_rr_set(
                        self.my_rr_set_domain.name, subname, "A", data
                    )
                    self.assertStatus(response, status.HTTP_200_OK)

                response = self.client.get_rr_set(
                    self.my_rr_set_domain.name, subname, "A"
                )
                self.assertStatus(response, status.HTTP_200_OK)
                current_rr_set.update(data)
                self.assertEqual(
                    set(response.data["records"]), set(current_rr_set["records"])
                )
                self.assertEqual(response.data["ttl"], current_rr_set["ttl"])

            response = self.client.patch_rr_set(
                self.my_rr_set_domain.name, subname, "A", {}
            )
            self.assertStatus(response, status.HTTP_200_OK)

    def test_rr_sets_touched_if_noop(self):
        for subname in self.SUBNAMES:
            touched_old = RRset.objects.get(
                domain=self.my_rr_set_domain, type="A", subname=subname
            ).touched
            response = self.client.patch_rr_set(
                self.my_rr_set_domain.name, subname, "A", {}
            )
            self.assertStatus(response, status.HTTP_200_OK)

            touched_new = RRset.objects.get(
                domain=self.my_rr_set_domain, type="A", subname=subname
            ).touched
            self.assertGreater(touched_new, touched_old)
            self.assertEqual(
                Domain.objects.get(name=self.my_rr_set_domain.name).touched, touched_new
            )

    def test_partially_update_other_rr_sets(self):
        data = {"records": ["3.2.3.4"], "ttl": 334}
        for subname in self.SUBNAMES:
            response = self.client.patch_rr_set(
                self.other_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    def test_update_other_rr_sets(self):
        data = {"ttl": 305}
        for subname in self.SUBNAMES:
            response = self.client.patch_rr_set(
                self.other_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    def test_update_essential_properties(self):
        # Changing the subname is expected to cause an error
        url = self.reverse(
            "v1:rrset", name=self.my_rr_set_domain.name, subname="test", type="A"
        )
        data = {"records": ["3.2.3.4"], "ttl": 3620, "subname": "test2", "type": "A"}
        response = self.client.patch(url, data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(response.data["subname"][0].code, "read-only-on-update")
        response = self.client.put(url, data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(response.data["subname"][0].code, "read-only-on-update")

        # Changing the type is expected to cause an error
        data = {"records": ["3.2.3.4"], "ttl": 3620, "subname": "test", "type": "TXT"}
        response = self.client.patch(url, data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(response.data["type"][0].code, "read-only-on-update")
        response = self.client.put(url, data)
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertEqual(response.data["type"][0].code, "read-only-on-update")

        # Changing "created" is no-op
        response = self.client.get(url)
        data = response.data
        created = data["created"]
        data["created"] = "2019-07-19T17:22:49.575717Z"
        response = self.client.patch(url, data)
        self.assertStatus(response, status.HTTP_200_OK)
        response = self.client.put(url, data)
        self.assertStatus(response, status.HTTP_200_OK)

        # Check that nothing changed
        response = self.client.get(url)
        self.assertStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data["records"][0], "2.2.3.4")
        self.assertEqual(response.data["ttl"], 3620)
        self.assertEqual(
            response.data["name"], "test." + self.my_rr_set_domain.name + "."
        )
        self.assertEqual(response.data["subname"], "test")
        self.assertEqual(response.data["type"], "A")
        self.assertEqual(response.data["created"], created)

        # This is expected to work, but the fields are ignored
        data = {"records": ["3.2.3.4"], "name": "example.com.", "domain": "example.com"}
        with self.assertRequests(
            self.requests_desec_rr_sets_update(name=self.my_rr_set_domain.name)
        ):
            response = self.client.patch(url, data)
            self.assertStatus(response, status.HTTP_200_OK)

        response = self.client.get(url)
        self.assertStatus(response, status.HTTP_200_OK)
        self.assertEqual(response.data["records"][0], "3.2.3.4")
        self.assertEqual(response.data["domain"], self.my_rr_set_domain.name)
        self.assertEqual(
            response.data["name"], "test." + self.my_rr_set_domain.name + "."
        )

    def test_update_unknown_rrset(self):
        url = self.reverse(
            "v1:rrset",
            name=self.my_rr_set_domain.name,
            subname="doesnotexist",
            type="A",
        )
        data = {"records": ["3.2.3.4"], "ttl": 3620}

        response = self.client.patch(url, data)
        self.assertStatus(response, status.HTTP_404_NOT_FOUND)

        response = self.client.put(url, data)
        self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    def test_delete_my_rr_sets_with_patch(self):
        data = {"records": []}
        for subname in self.SUBNAMES:
            with self.assertRequests(
                self.requests_desec_rr_sets_update(name=self.my_rr_set_domain.name)
            ):
                response = self.client.patch_rr_set(
                    self.my_rr_set_domain.name, subname, "A", data
                )
                self.assertStatus(response, status.HTTP_204_NO_CONTENT)

            # Deletion is only idempotent via DELETE. For PATCH/PUT, the view raises 404 if the instance does not
            # exist. By that time, the view has not parsed the payload yet and does not know it is a deletion.
            response = self.client.patch_rr_set(
                self.my_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

            response = self.client.get_rr_set(self.my_rr_set_domain.name, subname, "A")
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    def test_delete_my_rr_sets_with_delete(self):
        for subname in self.SUBNAMES:
            with self.assertRequests(
                self.requests_desec_rr_sets_update(name=self.my_rr_set_domain.name)
            ):
                response = self.client.delete_rr_set(
                    self.my_rr_set_domain.name, subname=subname, type_="A"
                )
                self.assertStatus(response, status.HTTP_204_NO_CONTENT)
                domain = Domain.objects.get(name=self.my_rr_set_domain.name)
                self.assertEqual(domain.touched, domain.published)

            response = self.client.delete_rr_set(
                self.my_rr_set_domain.name, subname=subname, type_="A"
            )
            self.assertStatus(response, status.HTTP_204_NO_CONTENT)

            response = self.client.get_rr_set(
                self.my_rr_set_domain.name, subname=subname, type_="A"
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

    def test_delete_other_rr_sets(self):
        data = {"records": []}
        for subname in self.SUBNAMES:
            # Try PATCH empty
            response = self.client.patch_rr_set(
                self.other_rr_set_domain.name, subname, "A", data
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

            # Try DELETE
            response = self.client.delete_rr_set(
                self.other_rr_set_domain.name, subname, "A"
            )
            self.assertStatus(response, status.HTTP_404_NOT_FOUND)

            # Make sure it actually is still there
            self.assertGreater(
                len(
                    self.other_rr_set_domain.rrset_set.filter(subname=subname, type="A")
                ),
                0,
            )

    def test_import_rr_sets(self):
        with self.assertRequests(
            self.request_pdns_zone_retrieve(name=self.my_domain.name)
        ):
            call_command("sync-from-pdns", self.my_domain.name)
        for response in [
            self.client.get_rr_sets(self.my_domain.name),
            self.client.get_rr_sets(self.my_domain.name, subname=""),
        ]:
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertEqual(len(response.data), 1, response.data)
            self.assertContainsRRSets(
                response.data,
                [dict(subname="", records=settings.DEFAULT_NS, type="NS")],
            )

    def test_extra_dnskeys(self):
        name = "ietf.org"
        dnskeys = [
            "256 3 5 AwEAAdDECajHaTjfSoNTY58WcBah1BxPKVIHBz4IfLjfqMvium4lgKtKZLe97DgJ5/NQrNEGGQmr6fKvUj67cfrZUojZ2cGRiz"
            "VhgkOqZ9scaTVXNuXLM5Tw7VWOVIceeXAuuH2mPIiEV6MhJYUsW6dvmNsJ4XwCgNgroAmXhoMEiWEjBB+wjYZQ5GtZHBFKVXACSWTiCtdd"
            "HcueOeSVPi5WH94VlubhHfiytNPZLrObhUCHT6k0tNE6phLoHnXWU+6vpsYpz6GhMw/R9BFxW5PdPFIWBgoWk2/XFVRSKG9Lr61b2z1R12"
            "6xeUwvw46RVy3hanV3vNO7LM5HniqaYclBbhk=",
            "257 3 5 AwEAAavjQ1H6pE8FV8LGP0wQBFVL0EM9BRfqxz9p/sZ+8AByqyFHLdZcHoOGF7CgB5OKYMvGOgysuYQloPlwbq7Ws5WywbutbX"
            "yG24lMWy4jijlJUsaFrS5EvUu4ydmuRc/TGnEXnN1XQkO+waIT4cLtrmcWjoY8Oqud6lDaJdj1cKr2nX1NrmMRowIu3DIVtGbQJmzpukpD"
            "VZaYMMAm8M5vz4U2vRCVETLgDoQ7rhsiD127J8gVExjO8B0113jCajbFRcMtUtFTjH4z7jXP2ZzDcXsgpe4LYFuenFQAcRBRlE6oaykHR7"
            "rlPqqmw58nIELJUFoMcb/BdRLgbyTeurFlnxs=",
        ]
        expected_ds = [
            "45586 5 2 67fcd7e0b9e0366309f3b6f7476dff931d5226edc5348cd80fd82a081dfcf6ee",
            "45586 5 4 aee6931c7790c428bca35dab9179cb27f042715e38e5a8adb6bb24c57c21c65dbd02a5b09887787f30128bfac8b6f0b5",
        ]

        domain = Domain.objects.create(name=name, owner=self.owner)
        rrset = domain.rrset_set.create(subname="", type="DNSKEY", ttl=3600)
        rrset.records.bulk_create(
            [RR(rrset=rrset, content=dnskey) for dnskey in dnskeys]
        )

        url = self.reverse("v1:domain-detail", name=domain.name)
        with self.assertRequests(
            self.request_pdns_zone_retrieve_crypto_keys(name=domain.name)
        ):
            response = self.client.get(url)
            self.assertStatus(response, status.HTTP_200_OK)
            self.assertEqual(
                response.data["keys"],
                [
                    {
                        "dnskey": key["dnskey"],
                        "ds": key["cds"] if key["flags"] & 1 else [],
                        "flags": key["flags"],
                        "keytype": key["keytype"],
                        "managed": True,
                    }
                    for key in self.get_body_pdns_zone_retrieve_crypto_keys()
                ]
                + [
                    {
                        "dnskey": dnskeys[0],
                        "ds": [],
                        "flags": 256,
                        "keytype": None,
                        "managed": False,
                    },
                    {
                        "dnskey": dnskeys[1],
                        "ds": expected_ds,
                        "flags": 257,
                        "keytype": None,
                        "managed": False,
                    },
                ],
            )

    def test_rrsets_policies(self):
        domain = self.my_empty_domain

        def assertRequests(*, allowed):
            cm = (
                self.assertRequests(self.requests_desec_rr_sets_update(domain.name))
                if allowed
                else nullcontext()
            )

            data = {"subname": "www", "type": "A", "ttl": 3600, "records": ["1.2.3.4"]}
            with cm:
                self.assertStatus(
                    self.client.post_rr_set(domain_name=domain.name, **data),
                    status.HTTP_201_CREATED if allowed else status.HTTP_403_FORBIDDEN,
                )

            data["records"] = ["4.3.2.1"]
            with cm:
                self.assertStatus(
                    self.client.put_rr_set(domain.name, "www", "A", data),
                    status.HTTP_200_OK if allowed else status.HTTP_404_NOT_FOUND,
                )

            data["records"] = []  # delete
            with cm:
                self.assertStatus(
                    self.client.patch_rr_set(domain.name, "www", "A", data),
                    (
                        status.HTTP_204_NO_CONTENT
                        if allowed
                        else status.HTTP_404_NOT_FOUND
                    ),
                )

            self.assertStatus(
                self.client.patch_rr_set(domain.name, "www", "A", data),
                status.HTTP_404_NOT_FOUND,  # no permission needed to see that
            )

            self.assertStatus(
                self.client.delete_rr_set(domain.name, "www", "A"),
                status.HTTP_204_NO_CONTENT,  # no permission needed for idempotency
            )

            if not allowed:
                # Create RRset manually so we cn try manipulating it
                data["contents"] = data.pop("records")
                self.my_empty_domain.rrset_set.create(**data)
                data["records"] = data.pop("contents")

                for response in [
                    self.client.patch_rr_set(domain.name, "www", "A", data),
                    self.client.put_rr_set(domain.name, "www", "A", data),
                    self.client.delete_rr_set(domain.name, "www", "A"),
                ]:
                    self.assertStatus(response, status.HTTP_403_FORBIDDEN)

            # Clean up
            rrset_qs = domain.rrset_set.filter(subname="www", type="A")
            if not allowed:
                self.assertTrue(rrset_qs.exists())
                rrset_qs.delete()
            self.assertFalse(rrset_qs.exists())

        assertRequests(allowed=True)

        qs = self.token.tokendomainpolicy_set
        qs.create(domain=None, subname=None, type=None)
        assertRequests(allowed=False)

        qs.create(domain=domain, subname=None, type="A", perm_write=True)
        assertRequests(allowed=True)

        qs.create(domain=domain, subname="www", type="A", perm_write=False)
        assertRequests(allowed=False)


class AuthenticatedRRSetLPSTestCase(AuthenticatedRRSetBaseTestCase):
    DYN = True

    def test_create_my_rr_sets_ip_block(self):
        BlockedSubnet.from_ip("3.2.2.3").save()
        response = self.client.post_rr_set(
            self.my_domain.name,
            records=["3.2.2.5"],
            ttl=3660,
            subname="blocktest",
            type="A",
        )
        self.assertStatus(response, status.HTTP_400_BAD_REQUEST)
        self.assertIn("IP address 3.2.2.5 not allowed.", str(response.data))