老李分享：使用 Python 的 Socket 模块开发 UDP 扫描工具

goodmm

　　        poptest是业内唯一的测试开发工程师培训机构，测试开发工程师主要是为测试服务开发测试工具，在工作中要求你做网络级别的安全性测试，但是条件限制你无法用商业工具，所以自己动手要写测试工具，在这里我们在测试开发工程师的就业培训中构建了一个场景，就是自己开发udp扫描工具，我们在现阶段主要是用python为主要开发语言来实现各种场景下的测试，而quicktestprofessional的培训我们已经免费。
　　      首先我们了解下概念：套接字能够访问底层网络信息。如，我们可以用它来检查IP和ICMP报头，他们都是OSI模型中网络层协议。
　　      使用UDP数据包可以当发送信息穿越子网时，不同于TCP，需要三次握手。我们需要做的仅仅是等待ICMP报文回应，来判断对方主机是否可用或不可访问。ICMP协议本质上是一个特殊的控制协议，它可以指示错误报告和控制机器数据传输的的行为。
　　    我们先创建一个socket 并将它绑定到一个外部网卡。这个网卡要启用混淆模式（promiscuous mode），他主要是获取经过这个网卡的所有数据包，包括目标地址不是它的数据包。使用 Windows 时要注意一点：我们需要发送一个 IOCTL 包才能将网卡设置为混淆模式。另外，虽然 linux 需要使用 ICMP，Windows 却可以以一种独立于协议的方式来嗅探收到的数据包。
　　  
　　import socket
import os

# host to listen
HOST = '192.168.1.114'

def sniffing(host, win, socket_prot):
    　　while 1:
        　　　　sniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket_prot)
        　　　　sniffer.bind((host, 0))

        　　　　sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)

        　　　　if win == 1:
            　　　　　　sniffer.ioctl(socket.SIO_RCVALL, socket_RCVALL_ON)

        　　　　
        　　　　print sniffer.recvfrom(65565)

def main(host):
    　　if os.name == 'nt':
        　　　　sniffing(host, 1, socket.IPPROTO_IP)
    　　else:
        　　　　sniffing(host, 0, socket.IPPROTO_ICMP)

if __name__ == '__main__':
    　　main(HOST)
　　在终端中运行如下命令进行测试:   $ sudo python sniffer.py
　　('E\x00\x00T\xb3\xec\x00\x005\x01\xe4\x13J}\xe1\x11\xc0\xa8\x01r\x00\x00v\xdfx\xa2\x00\x01sr\x98T\x00\x00\x00\x008\xe3\r\x00\x00\x00\x00\x00\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f !"#$%&\'()*+,-./01234567', ('74.125.225.17', 0))
('E\x00\x00T\xb4\x1b\x00\x005\x01\xe3\xe4J}\xe1\x11\xc0\xa8\x01r\x00\x00~\xd7x\xa2\x00\x02tr\x98T\x00\x00\x00\x00/\xea\r\x00\x00\x00\x00\x00\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f !"#$%&\'()*+,-./01234567', ('74.125.225.17', 0))
　　
　　我们需要对这些数据包进行解码
下面我们来研究下协议数据包组成，进行IP ICMP层解码
　　
　　IP 头
典型的 IP 头有如下结构，每个字段都对应一个变量
　　
　　
　　 ICMP头
　　ICMP 由于内容的不同其消息类型也不同，但每个消息都包括三个一致的元素：type，code (告知接收主机ICMP消息的解码类型）和 checksum。
　　
　　对于我们的扫描器，如果得到的 type 和 code 值是3，这意味着 Destination Unreachable(目标不可达)和 Port Unreachable (端口不可达) ICMP 消息错误
为描述 ICMP 消息头，用 python 的 ctypes 库来创建一个类
　　import ctypes

class ICMP(ctypes.Structure):
    　　_fields_ = [
    　　('type',        ctypes.c_ubyte),
    　　('code',        ctypes.c_ubyte),
    　　('checksum',    ctypes.c_ushort),
    　　('unused',      ctypes.c_ushort),
    　　('next_hop_mtu',ctypes.c_ushort)
    　　]

    　　def __new__(self, socket_buffer):
        　　　　　return self.from_buffer_copy(socket_buffer)

    　　def __init__(self, socket_buffer):
        　　　　pass
　　编写消息头解码器
　　开发 IP/ICMP 消息头解码器。脚本创建了一个 sniffer socket，然后在循环中持续读取数据包并进行解码。
注意代码中将 IP 头的前20个字节读取到了缓存，然后再打印消息头的变量。
ICMP 头数据如下：
　　import socket
import os
import struct
import ctypes
from ICMPHeader import ICMP

# host to listen on
HOST = '192.168.xxx.xxx'

def main():
    　　socket_protocol = socket.IPPROTO_ICMP
    　　sniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket_protocol)
    　　sniffer.bind(( HOST, 0 ))
    　　sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)

    　　while 1:
        　　　　raw_buffer = sniffer.recvfrom(65565)[0]
        　　　　ip_header = raw_buffer[0:20]
        　　　　iph = struct.unpack('!BBHHHBBH4s4s' , ip_header)

        　　　　version_ihl = iph[0]
        　　　　version = version_ihl >> 4
        　　　　ihl = version_ihl & 0xF
        　　　　iph_length = ihl * 4
        　　　　ttl = iph[5]
        　　　　protocol = iph[6]
        　　　　s_addr = socket.inet_ntoa(iph[8]);
        　　　　d_addr = socket.inet_ntoa(iph[9]);

        　　　　print 'IP -> Version:' + str(version) + ', Header Length:' + str(ihl) + \
        　　　　', TTL:' + str(ttl) + ', Protocol:' + str(protocol) + ', Source:'\
         　　　　+ str(s_addr) + ', Destination:' + str(d_addr)

　　　　buf = raw_buffer[iph_length:iph_length + ctypes.sizeof(ICMP)]
        　　　　icmp_header = ICMP(buf)

        　　　　print "ICMP -> Type:%d, Code:%d" %(icmp_header.type, icmp_header.code) + '\n'

if __name__ == '__main__':
    　　main()
　　我们运行 traceroute看下数据包情况：
　　$ traceroute www.google.com
traceroute to www.google.com (74.125.226.50), 30 hops max, 60 byte packets
1  * * *
2  * * *
3  67.59.255.137 (67.59.255.137)  17.183 ms 67.59.255.129 (67.59.255.129)  70.563 ms 67.59.255.137 (67.59.255.137)  21.480 ms
4  451be075.cst.lightpath.net (65.19.99.117)  14.639 ms rtr102.wan.hcvlny.cv.net (65.19.99.205)  24.086 ms 451be075.cst.lightpath.net (65.19.107.117)  24.025 ms
5  64.15.3.246 (64.15.3.246)  24.005 ms 64.15.0.218 (64.15.0.218)  23.961 ms 451be0c2.cst.lightpath.net (65.19.120.194)  23.935 ms
6  72.14.215.203 (72.14.215.203)  23.872 ms  46.943 ms *
7  216.239.50.141 (216.239.50.141)  48.906 ms  46.138 ms  46.122 ms
8  209.85.245.179 (209.85.245.179)  46.108 ms  46.095 ms  46.074 ms
9  lga15s43-in-f18.1e100.net (74.125.226.50)  45.997 ms  19.507 ms  16.607 ms
　　会得到这种输出 (注意 ICMP 的响应类型):
　　sudo python ip_header_decode.py
IP -> Version:4, Header Length:5, TTL:252, Protocol:1, Source:65.19.99.117, Destination:192.168.1.114
ICMP -> Type:11, Code:0(...)IP -> Version:4, Header Length:5, TTL:250, Protocol:1, Source:72.14.215.203, Destination:192.168.1.114
ICMP -> Type:11, Code:0
IP -> Version:4, Header Length:5, TTL:56, Protocol:1, Source:74.125.226.50, Destination:192.168.1.114
ICMP -> Type:3, Code:3
IP -> Version:4, Header Length:5, TTL:249, Protocol:1, Source:216.239.50.141, Destination:192.168.1.114
ICMP -> Type:11, Code:0(...)IP -> Version:4, Header Length:5, TTL:56, Protocol:1, Source:74.125.226.50, Destination:192.168.1.114
ICMP -> Type:3, Code:3
开发扫描器
　　在编写完整的扫描器前首先要安装 netaddr，它是一个用于表示和处理网络地址的 python 库。
Netaddr 提供了操作 IPv4，IPv6 和子网 Mac 等地址的能力。它非常有用，因为我们会用到子网掩码，如192.168.1.0/24
$ sudo pip install netaddr
　　我们可以使用如下的代码段来测试这个库 (成功会打印“OK”)：
　　import netaddr

ip = '192.168.xxx.xxx'
if ip in netaddr.IPNetwork('192.168.xxx.0/24'):
    print('OK!')

进一步强化扫描器
　　我们会将上面所提到的组织在一起来完成我们的扫描器，然后添加一个循环来向目标子网内的所有地址发送 UDP 数据报。
　　import threading

import time
import socket
import os
import struct
from netaddr import IPNetwork, IPAddress
from ICMPHeader import ICMP
import ctypes


HOST = '192.168.xxx.xxx'

SUBNET = '192.168.xxx.0/24'

MESSAGE = 'hellooooo'


def udp_sender(SUBNET, MESSAGE):
    time.sleep(5)
    sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    for ip in IPNetwork(SUBNET):
        try:
            sender.sendto(MESSAGE, ("%s" % ip, 65212))
        except:
            pass

def main():
    t = threading.Thread(target=udp_sender, args=(SUBNET, MESSAGE))
    t.start()

    socket_protocol = socket.IPPROTO_ICMP
    sniffer = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket_protocol)
    sniffer.bind(( HOST, 0 ))
    sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)

   
    while 1:
        raw_buffer = sniffer.recvfrom(65565)[0]
        ip_header = raw_buffer[0:20]
        iph = struct.unpack('!BBHHHBBH4s4s' , ip_header)

   
        version_ihl = iph[0]
        ihl = version_ihl & 0xF
        iph_length = ihl * 4
        src_addr = socket.inet_ntoa(iph[8]);

     
        buf = raw_buffer[iph_length:iph_length + ctypes.sizeof(ICMP)]
        icmp_header = ICMP(buf)

        # check for the type 3 and code and within our target subnet
        if icmp_header.code == 3 and icmp_header.type == 3:
            if IPAddress(src_addr) in IPNetwork(SUBNET):
                if raw_buffer[len(raw_buffer) - len(MESSAGE):] == MESSAGE:
                    print("Host up: %s" % src_addr)

if __name__ == '__main__':
    main()
运行后得到的结果如下:
$ sudo python scanner.py
Host up: 192.168.1.114(...)

      如果感兴趣可以进一步强化代码，进一步加强扫描器的功能，其实大家可以看到这里技术的难点是你对协议的熟悉程度，我们在做性能测试的过程中也会通过loadrunner去模拟协议的请求，所以关键是实现的内容，语言可以任意选择。欢迎大家咨询poptest测试开发工程师培训。