1.程式碼及程式碼講解。
程式碼編寫工具：VsCode
（1）socket嗅探器
首先第一個指令碼是最簡單的原始socket嗅探器，它只會讀一個封包，然後直接退出：
import socket import os #host to listen on HOST = ''
#這裡輸入自己的IP地址
def main(): #create raw socket, bin to public interface if os.name == 'nt': socket_protocol = socket.IPPROTO_IP else: socket_protocol = socket.IPPROTO_ICMP sniffer = socket.socket(socket.AF_INET,socket.SOCK_RAW.socket_protocol) sniffer.bind((HOST,0)) #include the IP header in the capture sniffer.setsockopt(socket.IPPROTO_IP,socket.IP_HDRINCL,1) if os.name =='nt': sniffer.ioctl(socket.SIO_RCVALL,socket.RCVALL_ON) #read one packet print(sniffer.recvfrom(65565)) if os.name == 'nt': sniffer.ioctl(socket.SIO_RCVALL,socket.RCVALL_OFF) if __name__ == '__main__': main()
注意：這裡Windows和Linux的區別是，前者允許我們嗅探任何協定的所有流入資料，而後者強制我們指定一個協定來嗅探，這裡指定的是ICMP。
上面這只是一個非常簡單的嗅探器，那我們將對它的功能進行進一步的拓展。
（2）ip解碼器
import ipaddress import os import socket import struct import sys #定義了一個Python結構，把封包的前20個位元組對映到IP頭物件中。展示目前的通訊協定和通訊雙方的IP地址 class IP: def __init__(self, buff=None): header = struct.unpack('<BHHHBBH4s4s', buff) self.ver = header[0] >> 4 self.ihl = header[0] & 0xF self.tos = header[1] self.len = header[2] self.id = header[3] self.offset = header[4] self.ttl = header[5] self.protocol = header[6] self.sum = header[7] self.src = header[8] self.dst = header[9] # human readable IP addresses #使用新打造的IP頭結構，將抓包邏輯改成持續抓包和解析 self.src_address = ipaddress.ip_address(self.src) self.dst_address = ipaddress.ip_address(self.dst) # map protocol constants to their names self.protocol_map = {1: "ICMP", 6: "TCP", 17: "UDP"} try: self.protocol = self.protocol_map[self.protocol_num] except Exception as e: print('%s No protocol for %s' % (e, self.protocol_num)) self.protocol = str(self.protocol_num) def sniff(host): # should look familiar from previous example if os.name == 'nt': socket_protocol = socket.IPPROTO_IP else: 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) if os.name == 'nt': sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON) try: while True: # read a packet #將前20位元組轉換成IP頭物件 raw_buffere = sniffer.recvfrom(65535)[0] # create an IP header from the first 20 bytes ip_header = IP(raw_buffere[0:20]) # print the detected protocol and hosts #列印抓取資訊 print('Protocol: %s %s -> %s' % (ip_header.protocol, ip_header.src_address, ip_header.dst_address)) except KeyboardInterrupt: # if we're on Windows, turn off promiscuous mode if os.name == 'nt': sniffer.ioctl(socket.SIO_RCVALL,socket.RCVALL_OFF) sys.exit() if __name__ == '__main': if len(sys.argv) == 2: host = sys.argv[1] else: host = '' sniffer(host)
在理解這一部分的內容之前，我們首先要了解一個python庫。struct庫：struct庫提供了一些格式字元，用來定義二進位制資料的結構。這個庫在解碼，密碼學方面經常會用到。
注：在struct庫裡，不存在對應於nybble格式（即4個二進位制位組成的資料塊，也叫作nibble）的格式字元。
self.ver = header[0] >> 4
對於IP頭的第一個位元組，我們只想取高位nybble（整個位元組裡的第一個nybble）作為ver的值。取某位元組高位nybble的常規方法是將其向右位移4位元，相當於在該位元組的開頭填4個0，把其尾部的4位元擠出去。這樣我們就得到了原位元組的第一個nybble。
self.ihl = header[0] & 0xF
我們想把低位nybble（或者說原位元組的最後4個二進位制位）填進hdrlen裡，取某個位元組低位nybble的常規方法是將其與數位0xF（00001111）進行按位元與運算。它利用了0 AND 1 = 0的特性（0代表假，1代表真）。想要AND表示式為真，表示式兩邊都必須為真。所以這個操作相當於刪除前4個二進位制位，因為任何數AND 0都得0；它保持了最後4個二進位制位不變，因為任何數AND 1還是原數位。
（3）解碼ICMP
import ipaddress import os import socket import struct import sys #定義了一個Python結構，把封包的前20個位元組對映到IP頭物件中。展示目前的通訊協定和通訊雙方的IP地址 class IP: def __init__(self, buff=None): header = struct.unpack('<BHHHBBH4s4s', buff) self.ver = header[0] >> 4 self.ihl = header[0] & 0xF self.tos = header[1] self.len = header[2] self.id = header[3] self.offset = header[4] self.ttl = header[5] self.protocol = header[6] self.sum = header[7] self.src = header[8] self.dst = header[9] # human readable IP addresses #使用新打造的IP頭結構，將抓包邏輯改成持續抓包和解析 self.src_address = ipaddress.ip_address(self.src) self.dst_address = ipaddress.ip_address(self.dst) # map protocol constants to their names self.protocol_map = {1: "ICMP", 6: "TCP", 17: "UDP"} try: self.protocol = self.protocol_map[self.protocol_num] except Exception as e: print('%s No protocol for %s' % (e, self.protocol_num)) self.protocol = str(self.protocol_num) class ICMP: def __init__(self, buff): header = struct.unpack('<BBHHH', buff) self.type = header[0] self.code = header[1] self.sum = header[2] self.id = header[3] self.seq = header[4] def sniff(host): # should look familiar from previous example if os.name == 'nt': socket_protocol = socket.IPPROTO_IP else: 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) if os.name == 'nt': sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON) try: while True: # read a packet #將前20位元組轉換成IP頭物件 raw_buffere = sniffer.recvfrom(65535)[0] # create an IP header from the first 20 bytes ip_header = IP(raw_buffere[0:20]) # if it's ICMP, we want it if ip_header.protocol == 'ICMP': print('Protocol: %s %s -> %s' % (ip_header.protocol,ip_header.src_address, ip_header.dst_address)) print(f'Version: {ip_header.ver}') print(f'Header Length; {ip_header.ihl} TTL: {ip_header.ttl}') # calculate where our ICMP packet starts offset = ip_header.ihl * 4 buf = raw_buffere[offset:offset + 8] # create our ICMP structure icmp_header = ICMP(buf) print('ICMP -> Yype: %s Code: %s\n' % (icmp_header.type, icmp_header.code)) except KeyboardInterrupt: if os.name == 'nt': sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF) sys.exit() if __name__ == '__main': if len(sys.argv) == 2: host = sys.argv[1] else: host = '10.74.212.22' sniff(host)
這段程式碼在之前的IP結構下方又建立了一個ICMP結構。在負責接收封包的主迴圈中，我們會判斷接收到的封包是否為ICMP封包，然後計算出ICMP資料在原始封包中的偏移，最後將資料按照ICMP結構進行解析。
（4）scanner.py
那前面我們已經完成了大部分工作，只剩下了一個任務——群發UDP封包，並解析結果
import ipaddress
import os
import socket import struct import sys import threading import time #subnet to target SUBNET = '192.168.0.1/24' # magic string we'll check ICMP responses for
#新增的這點程式碼應該很好理解。我們定義了一個簡單的字串作為「簽名」，用於確認收到的ICMP響應是否是由我們傳送的UDP包所觸發的
MESSAGE = 'PYTHONRULES!' #定義了一個Python結構，把封包的前20個位元組對映到IP頭物件中。展示目前的通訊協定和通訊雙方的IP地址 class IP: def __init__(self, buff=None): header = struct.unpack('<BHHHBBH4s4s', buff) self.ver = header[0] >> 4 self.ihl = header[0] & 0xF self.tos = header[1] self.len = header[2] self.id = header[3] self.offset = header[4] self.ttl = header[5] self.protocol = header[6] self.sum = header[7] self.src = header[8] self.dst = header[9] # human readable IP addresses #使用新打造的IP頭結構，將抓包邏輯改成持續抓包和解析 self.src_address = ipaddress.ip_address(self.src) self.dst_address = ipaddress.ip_address(self.dst) # map protocol constants to their names self.protocol_map = {1: "ICMP", 6: "TCP", 17: "UDP"} try: self.protocol = self.protocol_map[self.protocol_num] except Exception as e: print('%s No protocol for %s' % (e, self.protocol_num)) self.protocol = str(self.protocol_num) class ICMP: def __init__(self, buff): header = struct.unpack('<BBHHH', buff) self.type = header[0] self.code = header[1] self.sum = header[2] self.id = header[3] self.seq = header[4] def udp_sender(): with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sender: for ip in ipaddress.ip_network(SUBNET).hosts(): sender.sendto(bytes(MESSAGE, 'utf8'), (str(ip), 65212)) class Scanner: def __init__(self, host): self.host = host if os.name == 'nt': socket_protocol = socket.IPPROTO_IP else: socket_protocol = socket.IPPROTO_ICMP self.socket = socket.socket(socket.AF_INET,socket.SOCK_RAW, socket_protocol) self.socket.bind((host, 0)) self.socket.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1) if os.name == 'nt': self.socket.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON) def sniff(self): hosts_up = set([f'{str(self.host)} *']) try: while True: # read a packet raw_buffer = self.socket.recvfrom(65535)[0] # create an IP header from the first 20 bytes ip_header = IP(raw_buffer[0:20]) # if it's ICMP, we went it if ip_header.protocol == "ICMP": offset = ip_header.ihl1 * 4 buf = raw_buffer[offset:offset + 8] icmp_header = ICMP(buf) if icmp_header.code == 3 and icmp_header.type == 3: if ipaddress.ip_address(ip_header.src_address) in ipaddress.IPv4Network(SUBNET): # make sure it has our magic message if raw_buffer[len(raw_buffer) - len(MESSAGE):] == bytes(MESSAGE, 'utf8'): tgt = str(ip_header.src_address) if tgt !=self.host and tgt not in hosts_up: hosts_up.add(str(ip_header.src_address)) print(f'Host Up: {tgt}') except KeyboardInterrupt: if os.name == 'nt': self.socket.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF) print('\nUser interrupted') if hosts_up: print(f'\n\nSummary: Hosts up on {SUBNET}') for host in sorted(hosts_up): print(f'{host}') print('') sys.exit() if __name__ == '__main': if len(sys.argv) == 2: host = sys.argv[1] else: host = '' s = Scanner(host) time.sleep(5) t = threading.Thread(target=udp_sender) t.start() s.sniff()
其中我們引入ipaddress庫，這樣就能對整個子網進行主機發現掃描。
sniff函數會嗅探網路上的資料，步驟跟之前的例子差不多，唯一的區別就是這次它會把線上的主機記錄下來。接收到預期的ICMP訊息時，我們首先檢查這個響應是不是來自我們正在掃描的子網，然後檢查ICMP訊息裡有沒有我們自定義的簽名。如果所有檢查都通過了，就把傳送這條ICMP訊息的主機IP地址列印出來。如果使用Ctrl+C組合鍵中斷掃描過程的話，程式就會關閉網路卡混雜模式（如果是Windows平臺），並且把迄今為止掃描出來的主機都列印到螢幕上。
那到這裡我們的流量嗅探工具已經算編寫完成了，希望大家有所收穫。