biscuits/dns_compare
SHA256
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added more output to track query method used and to support TLS queries.

Browse Source
This commit is contained in:
Discsearcher
2026-08-12 10:32:52 -04:00
parent 08e97a046d
commit 771af73845
1 changed files with 35 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dns_compare.py
+35 -18
View File
@@ -37,7 +37,7 @@ parser.add_argument('--ns-builtin', nargs='+', help=f'Built-in public DNS server
parser.add_argument('--records', nargs='*', help='Hostname(s) to query. If not provided with --domain, queries the domain itself.')
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('--timeout', type=float, default=15.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()
@@ -94,7 +94,7 @@ except rdatatype.UnknownRdatatype:
# Function to perform a single DNS query
def perform_query(full_domain, ns, record_type, timeout):
"""Perform a DNS query and return results"""
"""Perform a DNS query and return results with protocol used"""
try:
qname = name.from_text(full_domain)
q = message.make_query(qname, record_type)
@@ -102,9 +102,12 @@ def perform_query(full_domain, ns, record_type, timeout):
# Add EDNS0 extension for better compatibility (like dig does)
q.use_edns(edns=0, ednsflags=0, payload=4096)
protocol_used = None
# Try UDP first
try:
r = query.udp(q, ns, timeout=timeout)
protocol_used = "UDP"
if args.verbose:
print(f"[DEBUG] UDP Response: Answer={len(r.answer)} RRsets, TC flag: {bool(r.flags & flags.TC)}", file=sys.stderr)
@@ -114,6 +117,7 @@ def perform_query(full_domain, ns, record_type, timeout):
print(f"[DEBUG] UDP Response truncated (TC flag set), retrying with TCP...", file=sys.stderr)
try:
r = query.tcp(q, ns, timeout=timeout)
protocol_used = "TCP"
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 tcp_error:
@@ -121,6 +125,7 @@ def perform_query(full_domain, ns, record_type, timeout):
if args.verbose:
print(f"[DEBUG] TCP failed ({tcp_error}), trying TLS...", file=sys.stderr)
r = query.tls(q, ns, timeout=timeout)
protocol_used = "TLS"
if args.verbose:
print(f"[DEBUG] TLS Response: Answer={len(r.answer)} RRsets, TC flag: {bool(r.flags & flags.TC)}", file=sys.stderr)
except Exception as udp_error:
@@ -129,6 +134,7 @@ def perform_query(full_domain, ns, record_type, timeout):
print(f"[DEBUG] UDP failed ({udp_error}), trying TCP...", file=sys.stderr)
try:
r = query.tcp(q, ns, timeout=timeout)
protocol_used = "TCP"
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 tcp_error:
@@ -136,6 +142,7 @@ def perform_query(full_domain, ns, record_type, timeout):
if args.verbose:
print(f"[DEBUG] TCP failed ({tcp_error}), trying TLS...", file=sys.stderr)
r = query.tls(q, ns, timeout=timeout)
protocol_used = "TLS"
if args.verbose:
print(f"[DEBUG] TLS Response: Answer={len(r.answer)} RRsets, TC flag: {bool(r.flags & flags.TC)}", file=sys.stderr)
@@ -155,7 +162,7 @@ def perform_query(full_domain, ns, record_type, timeout):
response_rcode = r.rcode()
if response_rcode != 0: # 0 = NOERROR
rcode_name = rcode.to_text(response_rcode)
return (full_domain, ns, f"RCODE: {rcode_name}")
return (full_domain, ns, f"RCODE: {rcode_name}", protocol_used)
results = []
@@ -165,7 +172,7 @@ def perform_query(full_domain, ns, record_type, timeout):
for rrset in answers:
for item in rrset:
results.append(str(item))
return (full_domain, ns, results)
return (full_domain, ns, results, protocol_used)
# If no answer, check additional section (some servers put records there)
additional = r.additional
@@ -174,7 +181,7 @@ def perform_query(full_domain, ns, record_type, timeout):
for item in rrset:
results.append(str(item))
if results:
return (full_domain, ns, results)
return (full_domain, ns, results, protocol_used)
# If still no results, check authority section
authority = r.authority
@@ -183,15 +190,15 @@ def perform_query(full_domain, ns, record_type, timeout):
for item in rrset:
results.append(str(item))
if results:
return (full_domain, ns, results)
return (full_domain, ns, results, protocol_used)
# Truly no records found
return (full_domain, ns, "NO RECORDS")
return (full_domain, ns, "NO RECORDS", protocol_used)
except (exception.Timeout, TimeoutError):
return (full_domain, ns, "TIMED OUT")
return (full_domain, ns, "TIMED OUT", None)
except Exception as e:
error_msg = str(e) if str(e) else type(e).__name__
return (full_domain, ns, f"ERROR: {error_msg}")
return (full_domain, ns, f"ERROR: {error_msg}", None)
# Start timer
start_time = time.time()
@@ -226,15 +233,15 @@ for full_domain in full_domains:
thread_count = args.threads if args.threads else max(1, os.cpu_count() - 1)
# Execute queries in parallel
results_map = {} # (full_domain, ns) -> resolved_ip
results_map = {} # (full_domain, ns) -> (resolved_ip, protocol_used)
with ThreadPoolExecutor(max_workers=thread_count) as executor:
futures = [executor.submit(perform_query, full_domain, ns, record_type, args.timeout)
for full_domain, ns, _ in query_tasks]
for future in futures:
full_domain, ns, resolved_ips = future.result()
full_domain, ns, resolved_ips, protocol_used = future.result()
if resolved_ips is not None:
results_map[(full_domain, ns)] = resolved_ips
results_map[(full_domain, ns)] = (resolved_ips, protocol_used)
# Build the output structure
output = {}
@@ -259,26 +266,36 @@ for full_domain in full_domains:
for ns in all_nameservers:
provider = nameserver_map.get(ns)
if (full_domain, ns) in results_map:
results = results_map[(full_domain, ns)]
results, protocol_used = results_map[(full_domain, ns)]
# Create entry with result and protocol
entry = {}
# Handle error/timeout/rcode cases (strings)
if isinstance(results, str):
output[full_domain]["records"][provider][ns] = results
entry["result"] = results
else:
results = sorted(results)
# Store as single value if only one result, otherwise as array
output[full_domain]["records"][provider][ns] = results[0] if len(results) == 1 else results
entry["result"] = results[0] if len(results) == 1 else results
if protocol_used:
entry["protocol"] = protocol_used
output[full_domain]["records"][provider][ns] = entry
else:
# Flat structure for manual nameservers only
for ns in all_nameservers:
if (full_domain, ns) in results_map:
results = results_map[(full_domain, ns)]
results, protocol_used = results_map[(full_domain, ns)]
# Create entry with result and protocol
entry = {}
# Handle error/timeout/rcode cases (strings)
if isinstance(results, str):
output[full_domain]["records"][ns] = results
entry["result"] = results
else:
results = sorted(results)
# Store as single value if only one result, otherwise as array
output[full_domain]["records"][ns] = results[0] if len(results) == 1 else results
entry["result"] = results[0] if len(results) == 1 else results
if protocol_used:
entry["protocol"] = protocol_used
output[full_domain]["records"][ns] = entry
# Calculate and add timing information
end_time = time.time()