Lab 15 — Dangling DNS: Find and Fix an Abandoned Subdomain

"""StashGrid local dangling-DNS teaching lab. Python 3.10+, standard library only.
Minimal UDP DNS, not a general-purpose resolver. No forwarding or public targets.
"""
import argparse
import http.client
import json
from pathlib import Path
import secrets
import socket
import socketserver
import struct
import threading
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

LOOPBACK = '127.0.0.1'
DNS_PORT, WEB_PORT = 1053, 8088
ALIAS, TARGET = 'portal.stashgrid.test', 'retired-app.provider.test'
STATE = Path(__file__).with_name('lab-state.json')
MODES = ('healthy', 'dangling', 'reclaimed', 'fixed')


def mode():
    if not STATE.exists():
        return 'healthy'
    value = json.loads(STATE.read_text(encoding='utf-8'))['mode']
    if value not in MODES:
        raise ValueError('Invalid lab state; run: python lab.py set healthy')
    return value


def set_mode(value):
    temp = STATE.with_suffix('.tmp')
    temp.write_text(json.dumps({'mode': value}), encoding='utf-8')
    temp.replace(STATE)


def encode_name(name):
    return b''.join(bytes([len(part)]) + part.encode('ascii') for part in name.split('.')) + b'\0'


def decode_name(data, offset):
    # This lab emits uncompressed names. Reject pointers rather than following arbitrary data.
    labels = []
    while True:
        size = data[offset]
        offset += 1
        if not size:
            return '.'.join(labels).lower(), offset
        if size > 63 or offset + size > len(data):
            raise ValueError('Invalid or compressed DNS name')
        labels.append(data[offset:offset + size].decode('ascii'))
        offset += size
        if sum(map(len, labels)) + len(labels) > 254:
            raise ValueError('DNS name too long')


def dns_reply(packet, current):
    if len(packet) < 12:
        return b''
    ident, flags, qd, _, _, _ = struct.unpack('!6H', packet[:12])
    if qd != 1 or flags & 0xF800:
        return struct.pack('!6H', ident, 0x8001, 0, 0, 0, 0)
    try:
        name, end = decode_name(packet, 12)
        qtype, qclass = struct.unpack('!HH', packet[end:end + 4])
    except (ValueError, IndexError, UnicodeError, struct.error):
        return struct.pack('!6H', ident, 0x8001, 0, 0, 0, 0)
    question, records, rcode = packet[12:end + 4], [], 0
    if name not in (ALIAS, TARGET) or qclass != 1:
        rcode = 5  # REFUSED. Never recurse, forward, or resolve external names.
    elif name == ALIAS and current == 'fixed':
        rcode = 3
    elif name == ALIAS:
        if qtype in (1, 5):
            records.append((ALIAS, 5, encode_name(TARGET)))
        if qtype == 1:
            if current == 'dangling':
                rcode = 3
            else:
                records.append((TARGET, 1, socket.inet_aton(LOOPBACK)))
    elif current == 'dangling':
        rcode = 3
    elif qtype == 1:
        records.append((TARGET, 1, socket.inet_aton(LOOPBACK)))
    body = b''.join(encode_name(n) + struct.pack('!HHIH', t, 1, 0, len(v)) + v for n, t, v in records)
    # Teaching simplification: TTL 0, no SOA/negative caching, no TCP, DNSSEC or recursion.
    response_flags = 0x8000 | (flags & 0x0100) | rcode | (0 if rcode == 5 else 0x0400)
    return struct.pack('!6H', ident, response_flags, 1, len(records), 0, 0) + question + body


class DNSHandler(socketserver.BaseRequestHandler):
    def handle(self):
        packet, sock = self.request
        answer = dns_reply(packet, mode())
        if answer:
            sock.sendto(answer, self.client_address)


class WebHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.headers.get('Host', '').split(':')[0] not in (ALIAS, TARGET):
            self.send_error(404, 'Use python lab.py visit for the lab hostname')
            return
        current = mode()
        code = 404 if current == 'dangling' else 200
        text = ('NO RESOURCE: the original application has been deleted.' if current == 'dangling'
                else 'OWNER B: harmless replacement content. Simulated resource reassignment.'
                if current in ('reclaimed', 'fixed') else 'OWNER A: original training application.')
        body = (text + '\nLOCAL SIMULATION ONLY - no real cloud resource was claimed.\n').encode()
        self.send_response(code)
        self.send_header('Content-Type', 'text/plain; charset=utf-8')
        self.send_header('Content-Length', str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *_):
        pass


def query(name=ALIAS, qtype=1):
    ident = secrets.randbits(16)
    packet = struct.pack('!6H', ident, 0x0100, 1, 0, 0, 0) + encode_name(name) + struct.pack('!HH', qtype, 1)
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.settimeout(2)
        sock.connect((LOOPBACK, DNS_PORT))
        sock.send(packet)
        data = sock.recv(4096)
    rid, flags, qd, count, _, _ = struct.unpack('!6H', data[:12])
    if rid != ident or not flags & 0x8000 or qd != 1:
        raise ValueError('Unexpected DNS response')
    _, offset = decode_name(data, 12)
    offset += 4
    records = []
    for _ in range(count):
        owner, offset = decode_name(data, offset)
        kind, _, ttl, length = struct.unpack('!HHIH', data[offset:offset + 10])
        offset += 10
        value = socket.inet_ntoa(data[offset:offset + length]) if kind == 1 else decode_name(data, offset)[0]
        records.append((owner, kind, value, ttl))
        offset += length
    return flags & 15, records


def show_query(qtype):
    status, records = query(qtype=qtype)
    print('DNS server: 127.0.0.1:1053; query:', ALIAS, qtype)
    print('Status:', {0: 'NOERROR', 3: 'NXDOMAIN', 5: 'REFUSED'}.get(status, status))
    for owner, kind, value, ttl in records:
        print(owner, 'CNAME' if kind == 5 else 'A', value, 'TTL=' + str(ttl))
    if not records:
        print('(no answer records)')


def visit():
    status, records = query()
    addresses = [value for _, kind, value, _ in records if kind == 1]
    if status or LOOPBACK not in addresses:
        print('No usable DNS address for ' + ALIAS + '; no HTTP request sent.')
        return
    # Use the explicit lab DNS result; never change the OS resolver or use a proxy.
    connection = http.client.HTTPConnection(LOOPBACK, WEB_PORT, timeout=2)
    try:
        connection.request('GET', '/', headers={'Host': ALIAS})
        response = connection.getresponse()
        print('HTTP', response.status, 'for', ALIAS)
        print(response.read().decode())
    finally:
        connection.close()


def serve():
    # Bind both ports before starting either listener. Never fall back to 0.0.0.0.
    with socketserver.UDPServer((LOOPBACK, DNS_PORT), DNSHandler) as dns:
        with HTTPServer((LOOPBACK, WEB_PORT), WebHandler) as web:
            for server in (dns, web):
                threading.Thread(target=server.serve_forever, daemon=True).start()
            print('LOCAL LAB: DNS 127.0.0.1:1053 UDP; HTTP 127.0.0.1:8088', flush=True)
            print('State:', mode(), '| Keep this terminal open. Ctrl+C stops both listeners.', flush=True)
            try:
                while True:
                    time.sleep(0.25)
            except KeyboardInterrupt:
                pass
            finally:
                dns.shutdown()
                web.shutdown()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    commands = parser.add_subparsers(dest='command', required=True)
    commands.add_parser('serve', help='Start loopback-only DNS and HTTP servers')
    commands.add_parser('visit', help='Resolve through lab DNS, then request the local site')
    commands.add_parser('status', help='Show the current local simulation state')
    q = commands.add_parser('query', help='Send a real UDP DNS question to the local lab')
    q.add_argument('type', choices=('A', 'CNAME'), nargs='?', default='A')
    commands.add_parser('set', help='Change only this local simulation').add_argument('mode', choices=MODES)
    args = parser.parse_args()
    if args.command == 'serve':
        serve()
    elif args.command == 'set':
        set_mode(args.mode)
        print('Local simulation state:', args.mode)
    elif args.command == 'status':
        print(mode())
    elif args.command == 'query':
        show_query(1 if args.type == 'A' else 5)
    else:
        visit()


if __name__ == '__main__':
    try:
        main()
    except (OSError, ValueError, KeyError) as error:
        raise SystemExit('Lab error: ' + str(error) + '\nCheck that python lab.py serve is running and ports 1053/8088 are free.')