如何通过命令行或C代码模拟DNS pipelining单次TCP同时查询IPv4/IPv6

glibc resolver默认未开启DNS pipelining支持，每次查询都会新建TCP连接或等待前一个请求返回后再发送下一个，所以无法实现你需要的单TCP会话合并查询的效果，可通过以下两种方案手动构造请求验证：

方案1：Python + Scapy 快速验证（推荐用于给客户演示） #

无需复杂编码，几分钟就能完成验证，还可以配合Wireshark抓包直接展示会话过程：

from scapy.all import DNS, DNSQR
import socket

# 按需修改以下配置
TARGET_DNS_IP = "替换为客户DNS服务器IP"
TEST_DOMAIN = "example.com"
DNS_PORT = 53

# 构造A（IPv4）和AAAA（IPv6）查询报文，注意两个查询ID要不同
query_a = DNS(id=0x1234, rd=1, qd=DNSQR(qname=TEST_DOMAIN, qtype="A"))
query_aaaa = DNS(id=0x1235, rd=1, qd=DNSQR(qname=TEST_DOMAIN, qtype="AAAA"))

# DNS over TCP要求每个报文前加2字节大端序的长度前缀
def wrap_tcp_dns(dns_pkt):
    raw_pkt = bytes(dns_pkt)
    return len(raw_pkt).to_bytes(2, byteorder="big") + raw_pkt

# 单TCP连接内连续发送两个查询，不要等待第一个响应
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((TARGET_DNS_IP, DNS_PORT))
sock.sendall(wrap_tcp_dns(query_a))
sock.sendall(wrap_tcp_dns(query_aaaa))

# 读取响应，拦截pipelining的服务器会直接断开连接或只返回一个响应
resp1 = sock.recv(4096)
resp2 = sock.recv(4096, socket.MSG_DONTWAIT) # 非阻塞读第二个响应
sock.close()

print(f"A记录响应长度：{len(resp1)}，AAAA记录响应长度：{len(resp2) if resp2 else 0}")
if len(resp1) > 0 and len(resp2) > 0:
    print("该DNS服务器支持DNS pipelining")
else:
    print("该DNS服务器拦截了DNS pipelining请求")

运行后如果第二个响应长度为0，或者运行时报连接重置错误，即可证明服务器存在拦截行为，配合Wireshark过滤tcp.port == 53的抓包结果，可直观展示两个查询是在同一个TCP会话内发送，作为给客户的证明材料。

方案2：C语言原生实现（适用于无Python环境的场景） #

如果需要在客户环境用二进制可执行文件验证，可以用原生Socket实现，核心逻辑和Python版本一致：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

// 此处省略DNS报文构造逻辑，和普通UDP DNS查询构造规则完全一致，仅需额外添加2字节大端序长度前缀
// 你可以直接复用现有DNS报文构造代码，或者引入轻量DNS解析库实现

int main(int argc, char *argv[]) {
    if (argc < 2) {
        printf("用法：%s <DNS服务器IP>\n", argv[0]);
        return 1;
    }

    int sock = socket(AF_INET, SOCK_STREAM, 0);
    if (sock < 0) {
        perror("创建socket失败");
        return 1;
    }

    struct sockaddr_in dns_addr = {
        .sin_family = AF_INET,
        .sin_port = htons(53),
        .sin_addr.s_addr = inet_addr(argv[1])
    };

    if (connect(sock, (struct sockaddr *)&dns_addr, sizeof(dns_addr)) < 0) {
        perror("连接DNS服务器失败");
        close(sock);
        return 1;
    }

    // 替换为你构造好的带TCP长度前缀的A、AAAA查询报文
    uint8_t query_a[64] = {0}; // 示例占位，实际替换为真实报文
    uint8_t query_aaaa[64] = {0}; // 示例占位，实际替换为真实报文
    size_t len_a = 35; // 示例长度，实际替换为真实报文长度
    size_t len_aaaa = 37; // 示例长度，实际替换为真实报文长度

    send(sock, query_a, len_a, 0);
    send(sock, query_aaaa, len_aaaa, 0);

    uint8_t buf[4096];
    ssize_t resp1_len = recv(sock, buf, 4096, 0);
    ssize_t resp2_len = recv(sock, buf, 4096, MSG_DONTWAIT);

    printf("第一个响应长度：%zd，第二个响应长度：%zd\n", resp1_len, resp2_len > 0 ? resp2_len : 0);
    close(sock);
    return 0;
}

注意事项 #

• 两个DNS查询的ID必须不同，否则无法区分对应响应
• 必须连续发送两个查询，不能等待第一个响应返回再发第二个，否则不符合RFC 7766定义的pipelining模式
• 抓包验证时请确认两个查询的TCP序列号是连续的，且处于同一个TCP四元组的会话内

内容的提问来源于stack exchange，提问作者viral_mutant