Home | History | Annotate | Line # | Download | only in nsec_synthesis 
      1 #!/usr/bin/python3
      2 
      3 # Copyright (C) Internet Systems Consortium, Inc. ("ISC")
      4 #
      5 # SPDX-License-Identifier: MPL-2.0
      6 
      7 from pathlib import Path
      8 
      9 import dns.flags
     10 import dns.name
     11 import dns.rdataclass
     12 import dns.rdatatype
     13 import pytest
     14 
     15 from isctest.run import EnvCmd
     16 
     17 import isctest
     18 
     19 PARENT = "f007.test."
     20 CHILD = "!.f007.test."
     21 PROBE = f"probe.{CHILD}"
     22 VICTIM = f"victim.{PARENT}"
     23 VICTIM_A = "203.0.113.7"
     24 AUTH = "10.53.0.3"
     25 RESOLVER = "10.53.0.2"
     26 
     27 pytestmark = pytest.mark.extra_artifacts(
     28     [
     29         "ans1/ans.run",
     30         "ns3/K*",
     31         "ns3/child.f007.test.db",
     32         "ns3/child.f007.test.db.signed",
     33         "ns3/dsset-*",
     34         "ns3/f007.test.db",
     35         "ns3/f007.test.db.signed",
     36     ]
     37 )
     38 
     39 
     40 def _write_zone(path, text):
     41     Path(path).write_text(text, encoding="ascii")
     42 
     43 
     44 def bootstrap():
     45     keygen = EnvCmd("KEYGEN", "-a ECDSA256 -q")
     46     dsfromkey = EnvCmd("DSFROMKEY", "-2")
     47     signer = EnvCmd("SIGNER", "-S -g")
     48 
     49     parent_ksk = keygen(f"-f KSK {PARENT}", cwd="ns3").out.strip()
     50     keygen(PARENT, cwd="ns3")
     51     child_ksk = keygen(f"-f KSK {CHILD}", cwd="ns3").out.strip()
     52     keygen(CHILD, cwd="ns3")
     53 
     54     child_ds = dsfromkey(f"{child_ksk}.key", cwd="ns3").out
     55 
     56     _write_zone(
     57         "ns3/child.f007.test.db",
     58         f"""$TTL 300
     59 $ORIGIN {CHILD}
     60 @ SOA ns.f007.test. hostmaster.f007.test. 1 3600 600 86400 300
     61 @ NS ns.f007.test.
     62 """,
     63     )
     64     _write_zone(
     65         "ns3/f007.test.db",
     66         f"""$TTL 300
     67 $ORIGIN {PARENT}
     68 @ SOA ns hostmaster 1 3600 600 86400 300
     69 @ NS ns
     70 ns A 10.53.0.3
     71 victim A {VICTIM_A}
     72 ! NS ns.f007.test.
     73 {child_ds}""",
     74     )
     75 
     76     signer(f"-o {CHILD} -f child.f007.test.db.signed child.f007.test.db", cwd="ns3")
     77     signer(f"-o {PARENT} -f f007.test.db.signed f007.test.db", cwd="ns3")
     78 
     79     dnskey = str(isctest.kasp.Key(parent_ksk, keydir="ns3").dnskey).split()
     80     ta = "".join(dnskey[7:])
     81 
     82     return {
     83         "DNSKEY": ta,
     84     }
     85 
     86 
     87 def _query(server, qname, qtype):
     88     query = isctest.query.create(qname, qtype)
     89     return isctest.query.tcp(query, server)
     90 
     91 
     92 def _rrset(response, section, owner, rdtype, covers=None):
     93     if covers is None:
     94         return response.get_rrset(
     95             section,
     96             dns.name.from_text(owner),
     97             dns.rdataclass.IN,
     98             rdtype,
     99         )
    100     return response.get_rrset(
    101         section,
    102         dns.name.from_text(owner),
    103         dns.rdataclass.IN,
    104         rdtype,
    105         covers=covers,
    106     )
    107 
    108 
    109 def _has_a(response, owner, address):
    110     rrset = _rrset(response, response.answer, owner, dns.rdatatype.A)
    111     return rrset is not None and any(rdata.address == address for rdata in rrset)
    112 
    113 
    114 def _check_child_self_loop_nsec(response):
    115     nsec = _rrset(response, response.authority, CHILD, dns.rdatatype.NSEC)
    116     assert nsec is not None, response.to_text()
    117     assert nsec[0].next == dns.name.from_text(CHILD), response.to_text()
    118 
    119     rrsig = _rrset(
    120         response,
    121         response.authority,
    122         CHILD,
    123         dns.rdatatype.RRSIG,
    124         covers=dns.rdatatype.NSEC,
    125     )
    126     assert rrsig is not None, response.to_text()
    127     assert rrsig[0].signer == dns.name.from_text(CHILD), response.to_text()
    128 
    129 
    130 def test_direct_child_self_loop_nsec_fixture():
    131     response = _query(AUTH, PROBE, "A")
    132     isctest.check.nxdomain(response)
    133     assert response.flags & dns.flags.AA
    134     _check_child_self_loop_nsec(response)
    135 
    136 
    137 def test_resolver_does_not_synth_parent_nxdomain_from_child_nsec():
    138     poison = _query(RESOLVER, PROBE, "A")
    139     isctest.check.nxdomain(poison)
    140     isctest.check.adflag(poison)
    141     _check_child_self_loop_nsec(poison)
    142 
    143     response = _query(RESOLVER, VICTIM, "A")
    144     isctest.check.noerror(response)
    145     assert _has_a(response, VICTIM, VICTIM_A), response.to_text()
    146     isctest.check.adflag(response)
    147     assert _rrset(response, response.authority, CHILD, dns.rdatatype.NSEC) is None
    148