diff --git a/dns_compare.py b/dns_compare.py index 63813d3..3f3082e 100755 --- a/dns_compare.py +++ b/dns_compare.py @@ -3,6 +3,7 @@ import argparse import json import os +import sys import time from datetime import datetime from concurrent.futures import ThreadPoolExecutor @@ -37,6 +38,7 @@ parser.add_argument('--records', nargs='*', help='Hostname(s) to query. If not p parser.add_argument('--record-type', default='A', help='DNS record type to query. Default: A') parser.add_argument('--threads', type=int, default=None, help='Number of threads to use for queries. Default: CPU count - 1') parser.add_argument('--timeout', type=float, default=5.0, help='DNS query timeout in seconds. Default: 5.0') +parser.add_argument('--verbose', action='store_true', help='Enable verbose debugging output.') parser.add_argument('--list-record-types', action='store_true', help='Print list of supported record types and exit.') args = parser.parse_args() @@ -97,12 +99,36 @@ def perform_query(full_domain, ns, record_type, timeout): qname = name.from_text(full_domain) q = message.make_query(qname, record_type) - # Try UDP first - r = query.udp(q, ns, timeout=timeout) + # Add EDNS0 extension for better compatibility (like dig does) + q.use_edns(edns=0, ednsflags=0, payload=4096) - # If truncated, retry with TCP to get all results - if r.flags & flags.TC: + # Try TCP first for better compatibility with some servers + try: r = query.tcp(q, ns, timeout=timeout) + if args.verbose: + print(f"[DEBUG] TCP Response: Answer={len(r.answer)} RRsets, TC flag: {bool(r.flags & flags.TC)}", file=sys.stderr) + except Exception as e: + # Fall back to UDP if TCP fails + if args.verbose: + print(f"[DEBUG] TCP failed ({e}), trying UDP...", file=sys.stderr) + r = query.udp(q, ns, timeout=timeout) + + # If UDP returns truncated, retry with TCP + if r.flags & flags.TC: + if args.verbose: + print(f"[DEBUG] UDP Response truncated (TC flag set), retrying with TCP...", file=sys.stderr) + r = query.tcp(q, ns, timeout=timeout) + if args.verbose: + print(f"[DEBUG] TCP Retry Response: Answer={len(r.answer)} RRsets, TC flag: {bool(r.flags & flags.TC)}", file=sys.stderr) + + # Verbose debugging + if args.verbose: + print(f"\n[DEBUG] Query: {full_domain} (@{ns}) Type: {rdatatype.to_text(record_type)}", file=sys.stderr) + print(f"[DEBUG] Response code: {rcode.to_text(r.rcode())}", file=sys.stderr) + print(f"[DEBUG] Answer section: {len(r.answer)} RRsets", file=sys.stderr) + print(f"[DEBUG] Authority section: {len(r.authority)} RRsets", file=sys.stderr) + print(f"[DEBUG] Additional section: {len(r.additional)} RRsets", file=sys.stderr) + print(f"[DEBUG] Full message: {r}", file=sys.stderr) # Check response code response_rcode = r.rcode() @@ -111,14 +137,35 @@ def perform_query(full_domain, ns, record_type, timeout): return (full_domain, ns, f"RCODE: {rcode_name}") results = [] + + # Check answer section first answers = r.answer if answers: for rrset in answers: for item in rrset: results.append(str(item)) - else: - # No answer records returned (NODATA) - return (full_domain, ns, "NO RECORDS") + return (full_domain, ns, results) + + # If no answer, check additional section (some servers put records there) + additional = r.additional + if additional: + for rrset in additional: + for item in rrset: + results.append(str(item)) + if results: + return (full_domain, ns, results) + + # If still no results, check authority section + authority = r.authority + if authority: + for rrset in authority: + for item in rrset: + results.append(str(item)) + if results: + return (full_domain, ns, results) + + # Truly no records found + return (full_domain, ns, "NO RECORDS") return (full_domain, ns, results) except (exception.Timeout, TimeoutError):