Merge pull request #68 from dnsviz/inconsistent-nxdomain

Check for inconsistent negative responses

3 files changed, 130 insertions, 11 deletions

diff --git a/dnsviz/analysis/errors.py b/dnsviz/analysis/errors.py
index 2e9a5f8..9bc7ac0 100644
--- a/dnsviz/analysis/errors.py
+++ b/dnsviz/analysis/errors.py
@@ -859,6 +859,66 @@ class WildcardCoveredNODATANSEC3(WildcardCoveredNODATA):
     references = ['RFC 5155, Sec. 8.7']
     nsec_type = 'NSEC3'
 
+class ExistingNSECError(NSECError):
+    required_params = ['queries']
+
+    def __init__(self, **kwargs):
+        super(ExistingNSECError, self).__init__(**kwargs)
+        queries_text = ['%s/%s' % (name, rdtype) for name, rdtype in self.template_kwargs['queries']]
+        self.template_kwargs['queries_text'] = ', '.join(queries_text)
+
+class ExistingCovered(ExistingNSECError):
+    description_template = 'The following queries resulted in an answer response, even though the %(nsec_type)s records indicate that the queried names don\'t exist: %(queries_text)s'
+    code = 'EXISTING_NAME_COVERED'
+
+class ExistingCoveredNSEC(ExistingCovered):
+    '''
+    >>> e = ExistingCoveredNSEC(queries=[('www.foo.baz.', 'A'), ('www1.foo.baz.', 'TXT')])
+    >>> e.description
+    "The following queries resulted in an answer response, even though the NSEC records indicate that the queried names don't exist: www.foo.baz./A, www1.foo.baz./TXT"
+    '''
+
+    _abstract = False
+    references = ['RFC 4035, Sec. 3.1.3.2']
+    nsec_type = 'NSEC'
+
+class ExistingCoveredNSEC3(ExistingCovered):
+    '''
+    >>> e = ExistingCoveredNSEC3(queries=[('www.foo.baz.', 'A'), ('www1.foo.baz.', 'TXT')])
+    >>> e.description
+    "The following queries resulted in an answer response, even though the NSEC3 records indicate that the queried names don't exist: www.foo.baz./A, www1.foo.baz./TXT"
+    '''
+
+    _abstract = False
+    references = ['RFC 5155, Sec. 8.4']
+    nsec_type = 'NSEC3'
+
+class ExistingTypeNotInBitmap(ExistingNSECError):
+    description_template = 'The following queries resulted in an answer response, even though the bitmap in the %(nsec_type)s RR indicates that the queried records don\'t exist: %(queries_text)s'
+    code = 'EXISTING_TYPE_NOT_IN_BITMAP'
+
+class ExistingTypeNotInBitmapNSEC(ExistingTypeNotInBitmap):
+    '''
+    >>> e = ExistingTypeNotInBitmapNSEC(queries=[('www.foo.baz.', 'A'), ('www.foo.baz.', 'TXT')])
+    >>> e.description
+    "The following queries resulted in an answer response, even though the bitmap in the NSEC RR indicates that the queried records don't exist: www.foo.baz./A, www.foo.baz./TXT"
+    '''
+
+    _abstract = False
+    references = ['RFC 4035, Sec. 3.1.3.1']
+    nsec_type = 'NSEC'
+
+class ExistingTypeNotInBitmapNSEC3(ExistingTypeNotInBitmap):
+    '''
+    >>> e = ExistingTypeNotInBitmapNSEC3(queries=[('www.foo.baz.', 'A'), ('www.foo.baz.', 'TXT')])
+    >>> e.description
+    "The following queries resulted in an answer response, even though the bitmap in the NSEC3 RR indicates that the queried records don't exist: www.foo.baz./A, www.foo.baz./TXT"
+    '''
+
+    _abstract = False
+    references = ['RFC 5155, Sec. 8.5']
+    nsec_type = 'NSEC3'
+
 class SnameCoveredNODATANSEC(NSECError):
     '''
     >>> e = SnameCoveredNODATANSEC(sname='foo.baz.')
@@ -2115,7 +2175,6 @@ class DNSKEYZeroLength(DNSKEYBadLength):
     references = []
     required_params = []
 
-
 class DNSKEYBadLengthGOST(DNSKEYBadLength):
     '''
     >>> e = DNSKEYBadLengthGOST(length=500)
diff --git a/dnsviz/analysis/offline.py b/dnsviz/analysis/offline.py
index 723c33c..72c18d0 100644
--- a/dnsviz/analysis/offline.py
+++ b/dnsviz/analysis/offline.py
@@ -834,6 +834,7 @@ class OfflineDomainNameAnalysis(OnlineDomainNameAnalysis):
         self._populate_rrsig_status_all(supported_algs)
         self._populate_nodata_status(supported_algs)
         self._populate_nxdomain_status(supported_algs)
+        self._populate_inconsistent_negative_dnssec_responses_all()
         self._finalize_key_roles()
         if not is_dlv:
             self._populate_delegation_status(supported_algs, supported_digest_algs)
@@ -870,8 +871,13 @@ class OfflineDomainNameAnalysis(OnlineDomainNameAnalysis):
 
             for rrset_info in query.answer_info:
                 self.yxdomain.add(rrset_info.rrset.name)
+                # for ALL types, add the name and type to yxrrset
                 self.yxrrset.add((rrset_info.rrset.name, rrset_info.rrset.rdtype))
-                self.yxrrset_proper.add((rrset_info.rrset.name, rrset_info.rrset.rdtype))
+                # for all types EXCEPT where the record is a CNAME record
+                # synthesized from a DNAME record, add the name and type to
+                # yxrrset_proper
+                if not (rrset_info.rrset.rdtype == dns.rdatatype.CNAME and rrset_info.cname_info_from_dname):
+                    self.yxrrset_proper.add((rrset_info.rrset.name, rrset_info.rrset.rdtype))
                 if rrset_info.dname_info is not None:
                     self.yxrrset.add((rrset_info.dname_info.rrset.name, rrset_info.dname_info.rrset.rdtype))
                 for cname_rrset_info in rrset_info.cname_info_from_dname:
@@ -2419,6 +2425,55 @@ class OfflineDomainNameAnalysis(OnlineDomainNameAnalysis):
                                 self.nodata_warnings[neg_response_info], self.nodata_errors[neg_response_info], \
                                 supported_algs)
 
+    def _populate_inconsistent_negative_dnssec_responses(self, neg_response_info, neg_status):
+        for nsec_status in neg_status[neg_response_info]:
+            queries_by_error = {
+                    Errors.ExistingTypeNotInBitmapNSEC3: [],
+                    Errors.ExistingTypeNotInBitmapNSEC: [],
+                    Errors.ExistingCoveredNSEC3: [],
+                    Errors.ExistingCoveredNSEC: [],
+            }
+            nsec_set_info = nsec_status.nsec_set_info
+            for (qname, rdtype) in self.yxrrset_proper:
+                if rdtype in (dns.rdatatype.DS, dns.rdatatype.DLV):
+                    continue
+                if nsec_set_info.use_nsec3:
+                    status = Status.NSEC3StatusNXDOMAIN(qname, rdtype, nsec_status.origin, nsec_status.is_zone, nsec_set_info)
+                    err_cls = Errors.ExistingCoveredNSEC3
+                else:
+                    status = Status.NSECStatusNXDOMAIN(qname, rdtype, nsec_status.origin, nsec_status.is_zone, nsec_set_info)
+                    err_cls = Errors.ExistingCoveredNSEC
+
+                if status.validation_status == Status.NSEC_STATUS_VALID and not status.opt_out:
+                    queries_by_error[err_cls].append((qname, rdtype))
+
+                if nsec_set_info.use_nsec3:
+                    status = Status.NSEC3StatusNODATA(qname, rdtype, nsec_status.origin, nsec_status.is_zone, nsec_set_info)
+                    err_cls = Errors.ExistingTypeNotInBitmapNSEC3
+                else:
+                    status = Status.NSECStatusNODATA(qname, rdtype, nsec_status.origin, nsec_status.is_zone, nsec_set_info, sname_must_match=True)
+                    err_cls = Errors.ExistingTypeNotInBitmapNSEC
+
+                if status.validation_status == Status.NSEC_STATUS_VALID and not status.opt_out:
+                    queries_by_error[err_cls].append((qname, rdtype))
+
+            for err_cls in queries_by_error:
+                if not queries_by_error[err_cls]:
+                    continue
+                queries = [(fmt.humanize_name(qname), dns.rdatatype.to_text(rdtype)) for qname, rdtype in queries_by_error[err_cls]]
+                err = Errors.DomainNameAnalysisError.insert_into_list(err_cls(queries=queries), nsec_status.errors, None, None, None)
+
+    def _populate_inconsistent_negative_dnssec_responses_all(self):
+
+        _logger.debug('Looking for negative responses that contradict positive responses (%s)...' % (fmt.humanize_name(self.name)))
+        for (qname, rdtype), query in self.queries.items():
+            if rdtype in (dns.rdatatype.DS, dns.rdatatype.DLV):
+                continue
+            for neg_response_info in query.nodata_info:
+                self._populate_inconsistent_negative_dnssec_responses(neg_response_info, self.nodata_status)
+            for neg_response_info in query.nxdomain_info:
+                self._populate_inconsistent_negative_dnssec_responses(neg_response_info, self.nxdomain_status)
+
     def _populate_dnskey_status(self, trusted_keys):
         if (self.name, dns.rdatatype.DNSKEY) not in self.queries:
             return
diff --git a/dnsviz/analysis/status.py b/dnsviz/analysis/status.py
index d349ded..621c438 100644
--- a/dnsviz/analysis/status.py
+++ b/dnsviz/analysis/status.py
@@ -487,6 +487,8 @@ class NSECStatusNXDOMAIN(NSECStatus):
 
         self.nsec_names_covering_qname = {}
         covering_names = nsec_set_info.nsec_covering_name(self.qname)
+        self.opt_out = None
+
         if covering_names:
             self.nsec_names_covering_qname[self.qname] = covering_names
 
@@ -676,7 +678,7 @@ class NSECStatusWildcard(NSECStatusNXDOMAIN):
         return d
 
 class NSECStatusNODATA(NSECStatus):
-    def __init__(self, qname, rdtype, origin, is_zone, nsec_set_info):
+    def __init__(self, qname, rdtype, origin, is_zone, nsec_set_info, sname_must_match=False):
         self.qname = qname
         self.rdtype = rdtype
         self.origin = origin
@@ -700,13 +702,14 @@ class NSECStatusNODATA(NSECStatus):
             self.has_ds = False
             self.has_soa = False
 
-            # If no NSEC exists for the name itself, then look for an NSEC with
-            # an (empty non-terminal) ancestor
-            for nsec_name in nsec_set_info.rrsets:
-                next_name = nsec_set_info.rrsets[nsec_name].rrset[0].next
-                if next_name.is_subdomain(self.qname) and next_name != self.qname:
-                    self.nsec_for_qname = nsec_set_info.rrsets[nsec_name]
-                    break
+            if not sname_must_match:
+                # If no NSEC exists for the name itself, then look for an NSEC with
+                # an (empty non-terminal) ancestor
+                for nsec_name in nsec_set_info.rrsets:
+                    next_name = nsec_set_info.rrsets[nsec_name].rrset[0].next
+                    if next_name.is_subdomain(self.qname) and next_name != self.qname:
+                        self.nsec_for_qname = nsec_set_info.rrsets[nsec_name]
+                        break
 
         self.nsec_names_covering_qname = {}
         covering_names = nsec_set_info.nsec_covering_name(self.qname)
@@ -731,6 +734,8 @@ class NSECStatusNODATA(NSECStatus):
         if covering_names:
             self.nsec_names_covering_origin[self.origin] = covering_names
 
+        self.opt_out = None
+
         self._set_validation_status(nsec_set_info)
 
     def __str__(self):
@@ -1309,7 +1314,7 @@ class NSEC3StatusNODATA(NSEC3Status):
                     self.errors.append(invalid_alg_err)
             if self.wildcard_has_rdtype:
                 self.validation_status = NSEC_STATUS_INVALID
-                self.errors.append(Errors.StypeInBitmapWildcardNODATANSEC3(sname=fmt.humanize_name(self.wildcard_name), stype=dns.rdatatype.to_text(self.rdtype)))
+                self.errors.append(Errors.StypeInBitmapWildcardNODATANSEC3(sname=fmt.humanize_name(self.get_wildcard()), stype=dns.rdatatype.to_text(self.rdtype)))
         elif self.nsec_names_covering_qname:
             if not self.opt_out:
                 self.validation_status = NSEC_STATUS_INVALID