DNS Resolver (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: DNS Resolver

Write a program that performs DNS lookups for domain names, supporting multiple record types, reverse lookups, DNS-over-HTTPS, and batch resolution with response time measurements.

FUNCTIONAL REQUIREMENTS:
- Accept a domain name as a command-line argument
- Support querying multiple DNS record types via --type flag: A, AAAA, MX, NS, TXT, CNAME, SOA, PTR, SRV (default: A)
- Support querying all common record types at once via --all flag
- Support reverse DNS lookups (IP to hostname) via --reverse flag
- Allow specifying a custom DNS server via --server flag (default: system DNS resolver)
- Support DNS-over-HTTPS (DoH) via --doh flag using configurable DoH providers
- Measure and display query response time in milliseconds for each lookup
- Support batch resolution of multiple domains from a file via --file flag (one domain per line)
- Display TTL (Time To Live) values for each record
- Support recursive trace mode via --trace flag: show the full resolution path from root servers to authoritative servers
- Print results to console in a structured format showing: domain, record type, value, TTL, response time, and nameserver used
- Save results as JSON with --output flag (default: dns_results.json)
- Support concurrent resolution of batch queries via --parallel flag with configurable concurrency (--threads, default 10)
- If no domain is given, resolve a set of well-known domains (example.com, google.com, github.com, cloudflare.com) across multiple record types and display a formatted comparison table
- Handle errors: NXDOMAIN (non-existent domain), SERVFAIL, timeout, invalid domain names, and unreachable DNS servers

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# DNS Resolver (Python)

Performs DNS lookups with multi-type support, reverse lookups, optional DNS-over-HTTPS, batch file input, response-time measurement, optional trace output, parallel batch execution, and JSON report export.

## Requirements

- Ubuntu 22.04
- Python 3.10+

## Dependencies (Pinned)

- `dnspython==2.7.0`
- `requests==2.32.3`

## Setup

```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

## Run

```bash
python src/main.py example.com
python src/main.py example.com --type A,AAAA,MX,TXT
python src/main.py --all example.com
python src/main.py 8.8.8.8 --reverse
python src/main.py example.com --server 1.1.1.1
python src/main.py example.com --doh cloudflare --type A,AAAA
python src/main.py --file domains.txt --parallel --threads 20 --type A
python src/main.py example.com --trace --output dns_results.json
python src/main.py
```

## Notes

- Supported types: `A, AAAA, MX, NS, TXT, CNAME, SOA, PTR, SRV`.
- `--all` queries all common record types.
- With no domain input, resolver queries: `example.com`, `google.com`, `github.com`, `cloudflare.com`.

dnspython==2.7.0
requests==2.32.3

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import concurrent.futures
import ipaddress
import json
import time
from pathlib import Path

import dns.exception
import dns.message
import dns.query
import dns.reversename
import dns.resolver
import requests

DEFAULT_DOMAINS = ["example.com", "google.com", "github.com", "cloudflare.com"]
COMMON_TYPES = ["A", "AAAA", "MX", "NS", "TXT", "CNAME", "SOA", "PTR", "SRV"]
DOH_PROVIDERS = {
    "google": "https://dns.google/resolve",
    "cloudflare": "https://cloudflare-dns.com/dns-query",
    "quad9": "https://dns.quad9.net:5053/dns-query",
}


def parse_types(args: argparse.Namespace) -> list[str]:
    if args.all:
        return COMMON_TYPES
    return [x.strip().upper() for x in (args.type or "A").split(",") if x.strip()]


def parse_domains(args: argparse.Namespace) -> list[str]:
    domains = []
    if args.domain:
        domains.append(args.domain)
    if args.file:
        domains.extend([line.strip() for line in Path(args.file).read_text(encoding="utf-8").splitlines() if line.strip()])
    if not domains:
        domains.extend(DEFAULT_DOMAINS)
    return sorted(set(domains))


def ptr_name_from_ip(ip: str) -> str:
    return str(dns.reversename.from_address(ip))


def query_system(resolver: dns.resolver.Resolver, domain: str, rtype: str, reverse_mode: bool) -> list[dict]:
    started = time.perf_counter()
    try:
        if reverse_mode:
            ans = resolver.resolve_address(domain, search=False)
            ms = (time.perf_counter() - started) * 1000.0
            return [{
                "domain": domain,
                "type": "PTR",
                "value": str(x),
                "ttl": getattr(ans.rrset, "ttl", None),
                "response_ms": round(ms, 2),
                "nameserver": ",".join(resolver.nameservers),
                "error": "",
            } for x in ans]

        ans = resolver.resolve(domain, rtype, search=False)
        ms = (time.perf_counter() - started) * 1000.0
        return [{
            "domain": domain,
            "type": rtype,
            "value": str(x),
            "ttl": getattr(ans.rrset, "ttl", None),
            "response_ms": round(ms, 2),
            "nameserver": ",".join(resolver.nameservers),
            "error": "",
        } for x in ans]
    except Exception as exc:
        return [{
            "domain": domain,
            "type": rtype if not reverse_mode else "PTR",
            "value": "",
            "ttl": None,
            "response_ms": round((time.perf_counter() - started) * 1000.0, 2),
            "nameserver": ",".join(resolver.nameservers),
            "error": str(exc),
        }]


def query_doh(provider: str, domain: str, rtype: str) -> list[dict]:
    base = DOH_PROVIDERS.get(provider, provider)
    started = time.perf_counter()
    try:
        res = requests.get(base, params={"name": domain, "type": rtype}, headers={"accept": "application/dns-json"}, timeout=10)
        res.raise_for_status()
        data = res.json()
        ms = (time.perf_counter() - started) * 1000.0
        if data.get("Status", 0) != 0:
            return [{
                "domain": domain, "type": rtype, "value": "", "ttl": None,
                "response_ms": round(ms, 2), "nameserver": f"doh:{provider}", "error": f"DoH status {data.get('Status')}",
            }]
        answers = data.get("Answer", []) or []
        if not answers:
            return [{
                "domain": domain, "type": rtype, "value": "", "ttl": None,
                "response_ms": round(ms, 2), "nameserver": f"doh:{provider}", "error": "NoAnswer",
            }]
        return [{
            "domain": domain,
            "type": rtype,
            "value": str(a.get("data", "")),
            "ttl": a.get("TTL"),
            "response_ms": round(ms, 2),
            "nameserver": f"doh:{provider}",
            "error": "",
        } for a in answers]
    except Exception as exc:
        return [{
            "domain": domain, "type": rtype, "value": "", "ttl": None,
            "response_ms": round((time.perf_counter() - started) * 1000.0, 2),
            "nameserver": f"doh:{provider}", "error": str(exc),
        }]


def trace_domain(resolver: dns.resolver.Resolver, domain: str) -> list[dict]:
    labels = domain.split(".")
    steps = []
    for i in range(len(labels)):
        zone = ".".join(labels[i:])
        try:
            ns = resolver.resolve(zone, "NS", search=False)
            steps.append({"zone": zone, "ns": [str(x) for x in ns]})
        except Exception as exc:
            steps.append({"zone": zone, "error": str(exc)})
    return steps


def print_rows(rows: list[dict]) -> None:
    print("Domain | Type | Value | TTL | Response(ms) | Nameserver | Error")
    print("------ | ---- | ----- | --- | ------------ | ---------- | -----")
    for r in rows:
        print(f"{r['domain']} | {r['type']} | {r['value']} | {r['ttl'] if r['ttl'] is not None else ''} | {r['response_ms']} | {r['nameserver']} | {r['error']}")


def parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(description="DNS Resolver")
    p.add_argument("domain", nargs="?")
    p.add_argument("--type", default="A")
    p.add_argument("--all", action="store_true")
    p.add_argument("--reverse", action="store_true")
    p.add_argument("--server")
    p.add_argument("--doh")
    p.add_argument("--file")
    p.add_argument("--trace", action="store_true")
    p.add_argument("--output", default="dns_results.json")
    p.add_argument("--parallel", action="store_true")
    p.add_argument("--threads", type=int, default=10)
    return p


def main() -> int:
    args = parser().parse_args()
    resolver = dns.resolver.Resolver()
    if args.server:
        resolver.nameservers = [args.server]

    types = parse_types(args)
    domains = parse_domains(args)
    reverse_mode = bool(args.reverse)

    tasks = []
    for d in domains:
        qname = ptr_name_from_ip(d) if (reverse_mode and args.doh) else d
        qtypes = ["PTR"] if reverse_mode and not args.doh else types
        for t in qtypes:
            tasks.append((d, qname, t))

    rows: list[dict] = []
    def run_task(task: tuple[str, str, str]) -> list[dict]:
        original, qname, rtype = task
        records = query_doh(args.doh, qname, rtype) if args.doh else query_system(resolver, qname, rtype, reverse_mode and rtype == "PTR")
        for r in records:
            r["domain"] = original
        return records

    if args.parallel and len(tasks) > 1:
        with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, args.threads)) as ex:
            futs = [ex.submit(run_task, t) for t in tasks]
            for fut in concurrent.futures.as_completed(futs):
                recs = fut.result()
                rows.extend(recs)
                for r in recs:
                    print(f"{r['domain']} {r['type']} -> {r['value'] or r['error']}")
    else:
        for t in tasks:
            recs = run_task(t)
            rows.extend(recs)
            for r in recs:
                print(f"{r['domain']} {r['type']} -> {r['value'] or r['error']}")

    rows.sort(key=lambda r: (r["domain"], r["type"], r["value"]))
    print_rows(rows)

    traces = {}
    if args.trace:
        for d in domains:
            traces[d] = trace_domain(resolver, d)
        print("\nTrace:")
        for d, steps in traces.items():
            print(f"- {d}")
            for s in steps:
                if "ns" in s:
                    print(f"  {s['zone']}: {', '.join(s['ns'])}")
                else:
                    print(f"  {s['zone']}: {s['error']}")

    payload = {
        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "options": vars(args),
        "results": rows,
    }
    if args.trace:
        payload["traces"] = traces
    out = Path(args.output).resolve()
    out.write_text(json.dumps(payload, indent=2), encoding="utf-8")
    print(f"\nSaved: {out}")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as exc:
        print(f"Error: {exc}")
        raise SystemExit(1)