auth: added a new config for direct queries of dnskey signature

docs/settings.rst

@@ -572,6 +572,17 @@ Configure a delay to send out notifications, no delay by default.
 Read additional DNSKEY, CDS and CDNSKEY records from the records table/your BIND zonefile. If not
 set, DNSKEY, CDS and CDNSKEY records in the zonefiles are ignored.
 
+.. _setting-direct-dnskey-signature:
+
+``direct-dnskey-signature``
+-----------------
+
+-  Boolean
+-  Default: no
+
+Read signatures of DNSKEY records directly from the backend. If not set and the record is not presigned, 
+DNSKEY records will be signed directly by PDNS Authoritative.
+
 .. _setting-disable-axfr:
 
 ``disable-axfr``

pdns/auth-main.cc

@@ -295,6 +295,7 @@ static void declareArguments()
 
   ::arg().setSwitch("traceback-handler", "Enable the traceback handler (Linux only)") = "yes";
   ::arg().setSwitch("direct-dnskey", "Fetch DNSKEY, CDS and CDNSKEY RRs from backend during DNSKEY or CDS/CDNSKEY synthesis") = "no";
+  ::arg().setSwitch("direct-dnskey-signature", "Fetch signature of DNSKEY RRs from backend directly") = "no";
   ::arg().set("default-ksk-algorithm", "Default KSK algorithm") = "ecdsa256";
   ::arg().set("default-ksk-size", "Default KSK size (0 means default)") = "0";
   ::arg().set("default-zsk-algorithm", "Default ZSK algorithm") = "";

pdns/dnssecsigner.cc

@@ -151,7 +151,7 @@ static void addSignature(DNSSECKeeper& dk, UeberBackend& db, const DNSName& sign
   if(toSign.empty())
     return;
   vector<RRSIGRecordContent> rrcs;
-  if(dk.isPresigned(signer)) {
+  if(dk.isPresigned(signer) || (::arg().mustDo("direct-dnskey-signature") && signQType == QType::DNSKEY)) {
     //cerr<<"Doing presignatures"<<endl;
     dk.getPreRRSIGs(db, outsigned, origTTL, packet); // does it all
   }

regression-tests.auth-py/test_DirectDNSKEYSignature.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python
+
+import dns
+import os
+import socket
+from authtests import AuthTest
+
+class TestDirectDNSKEYSignature(AuthTest):
+    _config_template = """
+    launch=bind
+    direct-dnskey=yes
+    direct-dnskey-signature=yes
+    """
+
+    _zones = {
+        'example.org': """
+example.org.                 3600 IN SOA     {soa}
+example.org.                 3600 IN NS      ns1.example.org.
+example.org.                 3600 IN NS      ns2.example.org.
+ns1.example.org.             3600 IN A       192.0.2.1
+ns2.example.org.             3600 IN A       192.0.2.2
+example.org.                 3600 IN DNSKEY  257 3 13 kRMX25/TJovOrsWq9Hv6QEFpzYsxItaOWPduFEwPz+5FM97SEHyCx+fc /XUN9gtktpXx45LAZpg/sFFEQH89og==
+example.org.                 3600 IN DNSKEY  256 3 13 Fy1p5/TTniw9Ukwca3Fnjo4tQk9ZK5zSwX9HZhHC2Tta/+3OZ9+y/Noz G51m/vs/I3oo9OqF+znxOi69yuGZaQ==
+example.org.                 3600 IN RRSIG   DNSKEY 13 2 3600 20250118211239 20241228221941 22273 example.org. 8HNifVnXhm5u+YDL8wWuJou5BWPzRYainXaP45qn2/yoPqBXSwhGFA2a kmh2Lqpj2D7qcs3KJ/QAR1QZ9CUAjw==
+        """
+    }
+
+    @classmethod
+    def setUpClass(cls):
+        cls.setUpSockets()
+        cls.startResponders()
+        confdir = os.path.join('configs', cls._confdir)
+        cls.createConfigDir(confdir)
+        cls.generateAllAuthConfig(confdir)
+        cls.startAuth(confdir, "0.0.0.0")
+        print("Launching tests...")
+
+    @classmethod
+    def setUpSockets(cls):
+        print("Setting up UDP socket...")
+        cls._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        cls._sock.settimeout(2.0)
+        cls._sock.connect((cls._PREFIX + ".2", cls._authPort))
+
+    def testDNSKEYQuery(self):
+        """Test to verify DNSKEY and RRSIG records are served correctly"""
+        query = dns.message.make_query('example.org', 'DNSKEY', use_edns=True, want_dnssec=True)
+        res = self.sendUDPQuery(query)
+
+        # Ensure no error in response
+        self.assertRcodeEqual(res, dns.rcode.NOERROR)
+
+        # Validate DNSKEY record
+        dnskey_found = any(rrset.rdtype == dns.rdatatype.DNSKEY for rrset in res.answer)
+        self.assertTrue(dnskey_found, "DNSKEY record not found in the answer section")
+
+        # Validate RRSIG record for DNSKEY
+        rrsig_found = any(rrset.rdtype == dns.rdatatype.RRSIG and rrset.covers == dns.rdatatype.DNSKEY for rrset in res.answer)
+        self.assertTrue(rrsig_found, "RRSIG for DNSKEY not found in the answer section")
+
+    def testDNSKEYQueryWithoutDNSSEC(self):
+        """Test to ensure no RRSIG records are returned without the DNSSEC flag"""
+        query = dns.message.make_query('example.org', 'DNSKEY', use_edns=True, want_dnssec=False)
+        res = self.sendUDPQuery(query)
+
+        # Ensure no error in response
+        self.assertRcodeEqual(res, dns.rcode.NOERROR)
+
+        # Ensure DNSKEY is present but no RRSIG
+        dnskey_found = any(rrset.rdtype == dns.rdatatype.DNSKEY for rrset in res.answer)
+        self.assertTrue(dnskey_found, "DNSKEY record not found in the answer section")
+
+        rrsig_found = any(rrset.rdtype == dns.rdatatype.RRSIG for rrset in res.answer)
+        self.assertFalse(rrsig_found, "RRSIG records found unexpectedly without DNSSEC flag")