深入解析:利用DNS协议进行字节传输测试的技术与工具

本文详细介绍了“Bytes over DNS”测试的实现方法,包括服务器端自定义DNS解析脚本、客户端通过操作系统API及Python模块进行测试,以及如何手动构造DNS数据包,深入探讨了网络协议层面的技术细节与工具应用。

Bytes over DNS 工具

以下是进行“Bytes over DNS”测试所使用的工具。

服务器端配置

在服务器端,我使用以下自定义脚本启动 dnsresolver.py 程序:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
LOGFILENAME = 'bod-dnsresolver-test.log'

def BoDTest(request, reply, dCommand):
    if request.q.qtype == dnslib.QTYPE.A:
        if len(request.q.qname.label[2]) == 1 and int(request.q.qname.label[1].decode(), 16) == ord(request.q.qname.label[2]):
            with open(LOGFILENAME, 'a') as fOut:
                print(f'BYTE_EQUAL {request.q.qname.label[1]} {request.q.qname.label[2]}', file=fOut)
            qname = request.q.qname
            answer = '. 60 IN A 127.0.0.1'
            for rr in dnslib.RR.fromZone(answer):
                a = copy.copy(rr)
                a.rname = qname
                reply.add_answer(a)
            return False, None
        else:
            with open(LOGFILENAME, 'a') as fOut:
                print(f'BYTE_DIFFERENT {request.q.qname.label[1]} {request.q.qname.label[2]}', file=fOut)
    return True, None

使用以下命令启动:

1
dnsresolver.py -s bod-dnsresolver-test.py type=resolve,label=bytes,function=BoDTest

同时,确保你的 DNS Glue 记录(例如,针对 mydomain.com)指向你的服务器。

然后,可以进行一个小测试:nslookup bytes.3D.=.mydomain.com。 当请求未经篡改到达时,将返回 127.0.0.1;如果被篡改,则返回 NXDOMAINBoDTest 函数还会将结果记录到文本文件 bod-dnsresolver-test.log 中。

工作站端测试

在你的工作站上,可以运行以下脚本,通过操作系统的 API 测试 DNS 请求中的所有字节值:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
#!/usr/bin/env python3

import socket
import sys

DOMAIN = '.mydomain.com.'

def DNSResolveA(char: int):
    hostname_ascii = 'bytes.%02x.%s' % (char, chr(char)) + DOMAIN
    hostname_ascii = hostname_ascii.replace('\\', '\\\\')
    print(hostname_ascii)
    try:
        results = socket.getaddrinfo(hostname_ascii, None, family=socket.AF_INET, type=0, proto=0, flags=socket.AI_CANONNAME)
    except socket.gaierror as e:
        print(f"Resolution failed: {e}")
        return 1
    except UnicodeError as e:
        print(f"Resolution failed: {e}")
        return 1

    if not results:
        print("No results returned by getaddrinfo.")
        return 0

    # 收集规范名称(可能为空)和地址
    canon_names = set()
    addresses = []
    for res in results:
        family, socktype, proto, canonname, sockaddr = res
        if canonname:
            canon_names.add(canonname)
        # sockaddr 是一个元组;对于 IPv4 是 (addr, port),对于 IPv6 是 (addr, port, flowinfo, scopeid)
        ip = sockaddr[0]
        addresses.append((family, ip))

    if canon_names:
        print("Canonical name(s):")
        for cn in sorted(canon_names):
            print("  -", cn)
        print()

    # 去重并按地址族分组
    unique_ips = {}
    for fam, ip in addresses:
        fam_name = "IPv4" if fam == socket.AF_INET else ("IPv6" if fam == socket.AF_INET6 else str(fam))
        unique_ips.setdefault(fam_name, set()).add(ip)

    for fam_name in sorted(unique_ips.keys()):
        print(f"{fam_name} addresses ({len(unique_ips[fam_name])}):")
        for ip in sorted(unique_ips[fam_name]):
            print("  -", ip)
    print()

    # (可选)尝试对每个 IP 进行反向 DNS 查询(可能较慢/并非始终可用)
    print("Reverse DNS (PTR) lookups:")
    for fam_name, ips in unique_ips.items():
        for ip in sorted(ips):
            try:
                host, aliases, _ = socket.gethostbyaddr(ip)
                print(f"  {ip} -> {host}")
            except Exception as e:
                print(f"  {ip} -> (no PTR)  [{e}]")

    return 0

if __name__ == "__main__":
    for char in range(256):
        DNSResolveA(char)

使用此脚本,通过 dnspython/dns.resolver Python 模块执行测试:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
import dns.resolver

resolver = dns.resolver.Resolver()
DOMAIN = b'.mydomain.com.'

#resolver.nameservers = ['127.0.0.1']
#resolver.nameservers = ['1.1.1.1']
resolver.nameservers = ['8.8.8.8']

for i in range(256):
    if i == 0x2E:
        continue
    if i == 0x5C:
        byte = b'\\\\'
    else:
        byte = bytes([i])
    try:
        answer = resolver.resolve(((b'bytes.%02x.%s' + DOMAIN) % (i, byte)).decode('latin'), "A")
        for rdata in answer:
            print(i, rdata.to_text())
    except (dns.name.LabelTooLong, dns.resolver.NXDOMAIN) as e:
        print(i, e)

使用此脚本,通过手动构造 DNS 数据包执行测试:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
import socket

DOMAIN = b'mydomain.com.'
DNS = '1.1.1.1'
DNS = '8.8.8.8'

def send_udp_payload(data: bytes, target_ip: str, port: int = 53) -> None:
    """
    通过 UDP 将原始二进制数据发送到目标 IP 和端口(默认 53)。

    :param data: 要发送的二进制有效负载(必须是字节)。
    :param target_ip: 目标 IP 地址(字符串)。
    :param port: 目标 UDP 端口(默认 = 53)。
    """
    # 创建 UDP 套接字
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        sock.sendto(data, (target_ip, port))
        print(f"Sent {len(data)} bytes to {target_ip}:{port}")
    except Exception as e:
        print(f"Error sending data: {e}")
    finally:
        sock.close()

def DNSEncodeDomain(domain):
    labels = domain.split(b'.')
    if labels[-1] != b'':
        labels.append(b'')
    data = bytearray()
    for label in labels:
        data += bytes([len(label)])
        data += label
    return data

data = bytearray([0x88, 0xea, 0x01, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, 0x62, 0x79, 0x74, 0x65, 0x73, 0x02, 0x32, 0x65, 0x01, 0x2e]) + DNSEncodeDomain(DOMAIN) + bytearray([0x00, 0x01, 0x00, 0x01])

for i in range(256):
    data[1] = i
    data[22] = i
    hexvalue = b'%02x' % i
    data[19:21] = hexvalue
    print(data)
    send_udp_payload(data, DNS)
comments powered by Disqus
使用 Hugo 构建
主题 StackJimmy 设计