How to Monitor Domain Expiry and DNS Drift in Python

Losing control of a production domain can happen in two primary ways: failing to renew the domain before it enters redemption, or having records altered without your knowledge due to compromised credentials, human error, or unauthorized registrar transfers.

Traditional domain monitoring often relies on the legacy WHOIS protocol (port 43). WHOIS responses are unstructured plain text with no universal schema; every registrar formats expiration dates and domain statuses differently. When parsers fail on an unexpected date string, expiration dates go untracked. On the DNS side, many teams rely on complex external services or heavyweight third-party libraries just to periodically verify that their nameservers, mail exchangers, and address records point to the correct infrastructure.

The script domainwatch_rdap_expiry_and_dns_monitor.py addresses both problems directly using standard Python libraries. It uses RDAP (Registration Data Access Protocol) over HTTPS to fetch structured JSON expiration dates, constructs RFC 1035 DNS queries over raw UDP sockets to inspect records, and stores records in an SQLite database to detect configuration drift.

Install & Usage

The tool requires Python 3.8 or later and relies entirely on standard Python modules (socket, struct, urllib.request, sqlite3, json, datetime, argparse, and smtplib). You do not need to install any external packages with pip.

Save the code into a single file named domainwatch_rdap_expiry_and_dns_monitor.py.

To view available runtime options, execute the script with the --help flag:

python domainwatch_rdap_expiry_and_dns_monitor.py --help

You can run individual functions directly or integrate the script into existing maintenance workflows. For example, to test DNS and RDAP inspection in an interactive Python shell using the script:

python -c "import domainwatch_rdap_expiry_and_dns_monitor as dw; print(dw.snapshot_dns('example.com'))"

The tool writes all monitored domain records, drift histories, and alerts into a local SQLite database named domainwatch.db by default.

How It Works

The script operates through three core subsystems: a wire-format DNS client, an RDAP JSON parser, and an SQLite state engine.

1. Native RFC 1035 DNS Queries

Instead of importing external DNS resolvers, the script constructs and decodes standard DNS packets directly using Python's struct and socket modules.

The build_dns_query function formats the header and question sections according to RFC 1035:

def build_dns_query(domain: str, qtype: int, tx_id: Optional[int] = None) -> bytes:
    """Builds a raw RFC 1035 UDP DNS standard query packet."""
    if tx_id is None:
        tx_id = random.randint(1, 65535)

    # Flags: 0x0100 (Standard query, Recursion Desired = 1)
    flags = 0x0100
    qdcount = 1
    ancount = 0
    nscount = 0
    arcount = 0

    header = struct.pack("!HHHHHH", tx_id, flags, qdcount, ancount, nscount, arcount)

    # Encode domain labels
    qname = bytearray()
    clean_domain = domain.strip().strip(".")
    for part in clean_domain.split("."):
        encoded = part.encode("utf-8")
        if len(encoded) > 63:
            raise ValueError(f"Label '{part}' exceeds 63 octets")
        qname.append(len(encoded))
        qname.extend(encoded)
    qname.append(0)

    # QTYPE and QCLASS (1 = IN)
    question = bytes(qname) + struct.pack("!HH", qtype, 1)
    return header + question

Once the response returns over UDP port 53, parse_dns_response inspects the byte payload, decompresses pointers (handling offset jumps indicated by the 0xC0 prefix byte), and extracts resource records for four critical types:

• A (IPv4 addresses via socket.inet_ntoa)
• NS (Authoritative nameservers)
• MX (Mail exchangers with priority values)
• TXT (Verification strings and SPF records)

The snapshot_dns function queries all four types in sequence, returning a dictionary containing the active records.

2. RDAP Lookup and Expiration Parsing

Unlike WHOIS, RDAP serves structured JSON responses over HTTPS. The fetch_rdap and parse_rdap_payload functions read registration events to determine when the domain is scheduled to expire:

def parse_rdap_payload(data: Dict[str, Any]) -> Dict[str, Any]:
    """Extracts expiration, registrar, status, and event timestamps from RDAP JSON."""
    expires_at = None
    created_at = None
    updated_at = None

    for ev in data.get("events", []):
        action = ev.get("eventAction", "").lower()
        d_str = ev.get("eventDate")
        if "expiration" in action:
            expires_at = d_str
        elif "registration" in action or "transfer" in action:
            if not created_at:
                created_at = d_str
        elif "last changed" in action or "last update" in action:
            updated_at = d_str

    registrar = "Unknown"
    for ent in data.get("entities", []):
        roles = [r.lower() for r in ent.get("roles", [])]
        if "registrar" in roles:
            vcard = ent.get("vcardArray", [])
            if len(vcard) > 1 and isinstance(vcard[1], list):
                for prop in vcard[1]:
                    if prop and prop[0] == "fn" and len(prop) > 3:
                        registrar = str(prop[3])
                        break
            if registrar == "Unknown":
                registrar = ent.get("handle") or ent.get("name") or "Unknown"

    status_list = data.get("status", [])
    status_str = ", ".join(status_list) if status_list else "active"

    days_until_expiry = None
    if expires_at:
        exp_dt = parse_iso_datetime(expires_at)
        if exp_dt:
            now_dt = datetime.datetime.now(datetime.timezone.utc)
            days_until_expiry = (exp_dt - now_dt).days

    return {
        "expires_at": expires_at,
        "created_at": created_at,
        "updated_at": updated_at,
        "registrar": registrar,
        "status": status_str,
        "days_until_expiry": days_until_expiry,
    }

The extracted timestamp is normalized into a UTC datetime object, allowing the script to calculate the exact number of days remaining.

3. State Tracking and Drift Detection

When evaluating DNS records between monitoring intervals, the script compares the latest state against the values stored in SQLite. The diff_dns_snapshots function uses Python set operations to identify items that were added or removed:

def diff_dns_snapshots(
    old_snap: Dict[str, List[str]], new_snap: Dict[str, List[str]]
) -> List[Dict[str, Any]]:
    """Compares two DNS snapshots and produces a structured list of changes."""
    changes = []
    all_keys = sorted(set(list(old_snap.keys()) + list(new_snap.keys())))
    for rtype in all_keys:
        old_vals = set(old_snap.get(rtype, []))
        new_vals = set(new_snap.get(rtype, []))
        added = sorted(new_vals - old_vals)
        removed = sorted(old_vals - new_vals)
        if added or removed:
            changes.append({
                "type": rtype,
                "added": added,
                "removed": removed,
            })
    return changes

If an unauthorized mail exchanger is added or a nameserver entry drops, the change is logged to the events table with the corresponding timestamp and details.

Limitations

While the script avoids third-party package overhead, the implementation has several architectural trade-offs:

1. No Encrypted DNS Transport: Queries use plain DNS on port 53. DNS-over-HTTPS and DNS-over-TLS are not supported, so lookups are visible to anything on the network path to the resolver.
2. Limited Record Coverage: The packet parser explicitly decodes A, NS, MX, and TXT records. It does not parse AAAA (IPv6), CNAME, SRV, CAA, or SOA records.
3. No DNSSEC Validation: The raw query logic sets the recursion desired bit but does not request or validate DNSSEC records (RRSIG, DNSKEY, DS). Responses are parsed as received from the upstream resolver.
4. RDAP Bootstrap Dependency: The default configuration formats queries through a single public RDAP aggregator. If that service experiences an outage or throttles requests, expiry data collection will fail until the endpoint recovers or is updated.

FAQ

Does this script require pip packages like dnspython or requests?

No. The script relies solely on Python's built-in standard library. Network requests are handled by urllib.request, DNS communication runs on standard socket calls, and state management uses the built-in sqlite3 driver.

What happens if a domain registrar does not publish RDAP data?

If the RDAP aggregator or authoritative registry returns an HTTP error (such as a 404 or 500) or fails to provide an expiration event, fetch_rdap returns an error payload. The domain's DNS records will continue to be monitored for drift, but the expires_at column in SQLite remains unset until valid RDAP data is returned.

Can the DNS queries point to an internal or custom resolver?

Yes. The DNS query functions accept an optional dns_server parameter, which defaults to 1.1.1.1. You can pass any reachable IPv4 address, including internal corporate recursive resolvers or local DNS servers listening on port 53.

Full version and the rest of the toolkit: DomainWatch RDAP expiry and DNS monitor on PyForge. Product page: https://py-forge.com/products/domainwatch-rdap-expiry-and-dns-monitor