desec-stack/test/e2e2/spec/test_replication.py

from base64 import b64decode
import os
import socket

import dns.query
import pytest

from conftest import DeSECAPIV1Client, return_eventually, query_replication, random_domainname, assert_eventually, \
    FaketimeShift


some_ds_records = [
    '60604 8 1 ef66f772935b412376c8445c4442b802b0322814',
    '60604 8 2 c2739629145faaf464ff1bc65612fd1eb5766e80c96932d808edfb55d1e1f2ce',
    '60604 8 4 5943dac4fc4aad637445f483b0f43bd4152fab19250fd26df82bf12020a7f7101caa17e723cf433f43d2bbed11231e03',
]


def test_signature_rotation(api_user_domain: DeSECAPIV1Client):
    name = random_domainname()
    api_user_domain.domain_create(name)
    rrsig = return_eventually(lambda: query_replication(name, "", 'RRSIG', covers='SOA'), timeout=20)
    with FaketimeShift(days=7):
        assert_eventually(lambda: rrsig != query_replication(name, "", 'RRSIG', covers='SOA'), timeout=60)


def test_zone_deletion(api_user_domain: DeSECAPIV1Client):
    name = api_user_domain.domain
    assert_eventually(lambda: query_replication(name, "", 'SOA') is not None, timeout=20)
    api_user_domain.domain_destroy(name)
    assert_eventually(lambda: query_replication(name, "", 'SOA') is None, timeout=20)


@pytest.mark.performance
def test_signature_rotation_performance(api_user_domain: DeSECAPIV1Client):
    root_domain = api_user_domain.domain

    # test configuration
    bulk_block_size = 500
    domain_sizes = {
        # number of delegations: number of zones
        2000: 1,
        1000: 2,
        10: 10,
    }

    # create test domains
    domain_names = {
        num_delegations: [random_domainname() + f'.num-ds-{num_delegations}.' + root_domain for _ in range(num_zones)]
        for num_delegations, num_zones in domain_sizes.items()
    }
    for num_delegations, names in domain_names.items():
        for name in names:
            # create a domain with name `name` and `num_delegations` delegations
            api_user_domain.domain_create(name)
            for a in range(0, num_delegations, bulk_block_size):  # run block-wise to avoid exceeding max request size
                r = api_user_domain.rr_set_create_bulk(
                    name,
                    [
                        {"subname": f'x{i}', "type": "DS", "ttl": 3600, "records": some_ds_records}
                        for i in range(a, a + bulk_block_size)
                    ] + [
                        {"subname": f'x{i}', "type": "NS", "ttl": 3600, "records": ['ns1.test.', 'ns2.test.']}
                        for i in range(a, a + bulk_block_size)
                    ]
                )
                assert r.status_code == 200

    # retrieve all SOA RRSIGs
    soa_rrsig = {}
    for names in domain_names.values():
        for name in names:
            soa_rrsig[name] = return_eventually(lambda: query_replication(name, "", 'RRSIG', covers='SOA'), timeout=20)

    # rotate signatures
    with FaketimeShift(days=7):
        # assert SOA RRSIG has been updated
        for names in domain_names.values():
            for name in names:
                assert_eventually(
                    lambda: soa_rrsig[name] != query_replication(name, "", 'RRSIG', covers='SOA'),
                    timeout=600,  # depending on number of domains in the database, this value requires increase
                )

def test_tsig_axfr(api_user_domain: DeSECAPIV1Client):
    ns_ip = socket.gethostbyname('nsmaster')

    def count_xfr_rrsets(**kwargs):
        xfr = dns.query.xfr(ns_ip, api_user_domain.domain, **kwargs)
        zone = dns.zone.from_xfr(xfr)

        ## from dnspython 2.2.0 on
        #zone = dns.zone.Zone(api_user_domain.domain)
        #query, _ = dns.xfr.make_query(zone, **kwargs)
        #dns.query.inbound_xfr(ns_ip, zone, query)
        return sum(1 for _ in zone.iterate_rdatasets())

    with pytest.raises(dns.xfr.TransferError) as exc_info:
        count_xfr_rrsets()

    assert exc_info.value.rcode == dns.rcode.NOTAUTH

    keyring = {'default.': b64decode('XXXXXXXXXXXXXXXXXXXXXX==')}
    with pytest.raises(dns.xfr.TransferError) as exc_info:
        count_xfr_rrsets(keyring=keyring, keyname=None)

    assert exc_info.value.rcode == dns.rcode.NOTAUTH

    keyring = {'default.': b64decode(os.environ['DESECSTACK_NSMASTER_TSIGKEY'])}
    assert_eventually(lambda: count_xfr_rrsets(keyring=keyring, keyname=None) > 5, timeout=20, retry_on=(Exception,))