python网络：原始套接字和流量嗅探

发表于 | 分类于 | 本文总阅读量 次
字数统计 1.5k 字 | 阅读时长 7 分钟

开发UDP主机发现工具

嗅探工具的主要目标是基于UDP发现目标网络中的存活主机。攻击者需要了解网络中所有潜在的目标便于开展侦察和漏洞攻击尝试。

当你发送一个UDP数据包到主机的某个关闭的UDP端口上时，目标主机通常会返回一个ICMP包指示目标端口不可达。这样其实就说明了这个ip是存活的，因为ip是不存在的话，是不会有任何响应的。

接下来还有一点要注意，我们必须选择到一个未被使用的UDP端口才行，因为发送到活动端口也不一定会有响应回复给你，为了达到覆盖的范围，我们可以探查多个端口的响应情况，避免正好将数据发送到活动的UDP服务上。

为啥用UDP？因为UDP很简单，不用分析各种上层协议，就发送+等待。

Windows和Linux上的包嗅探

示例代码：

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

host = "192.168.66.190"
# 判断是否为windows，windows可以抓任意协议的包，linux只能抓ICMP
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))
# set socket option 
sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
# 系统为windows，则开启混杂模式
if os.name == "nt":
    sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)
    
print sniffer.recvfrom(65565)
# 系统为windows，则关闭混杂模式
if os.name == "nt":
    sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)

以上代码只是抓一个包，windows可以抓到发送包，而运行在linux上的话，只能抓到ICMP包。

解码IP层

我们接下来要解析一下ip层的报文结构

ip头部格式

我们要解析全部IP头，提取其中的协议类型，源IP地址和目的IP地址。

Python中ctypes模块创建类似于C的结构体，我们可以用这种方式对头部成分进行定义。

IP头部在c语言中的定义：

1
2
3
4
5
6
7
8
9
10
11
12
13
struct ip {
    u_char  ip_v:4;
    u_char  ip_h1:4;
    u_char  ip_tos;
    u_short ip_len;
    u_short ip_id;
    u_short ip_off;
    u_char  ip_ttl;
    u_char  ip_p;
    u_short ip_sum;
    u_long  ip_src;
    u_long  ip_dst;
}

我们将上面c代码带入到上面python代码中（sniffer_ip_header_decode.py）:

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
69
# -*- coding: UTF-8 -*-

import socket
import os

###  新 加 ####
import struct
from ctypes import *
##############

host = "192.168.66.190"
#########################################################
class IP(Structure):
    _fields_ = [
        ("version",      c_ubyte, 4),
        ("ih1",          c_ubyte, 4),
        ("tos",          c_ubyte),
        ("len",          c_ushort),
        ("id",           c_ushort),
        ("offset",       c_ushort),
        ("ttl",          c_ubyte),
        ("protocol_num", c_ubyte),
        ("sum",          c_ushort),
        ("src",          c_ulong),
        ("dst",          c_ulong)
    ]
    
    def __new__(self, socket_buffer = None):
        return self.from_buffer_copy(socket_buffer)
    def __init__(self, socket_buffer = None):
        self.protocol_map = {1:"ICMP", 6:"TCP", 17:"UDP"}
      	self.src_address = socket.inet_ntoa(struct.pack("<L",self.src))
        self.dst_address = socket.inet_ntoa(struct.pack("<L",self.dst))
        try:
            self.protocol = self.protocol_map[self.protocol_num]
        except:
            self.protocol = str(self.protocol_num)
######################################################################


# 判断是否为windows，windows可以抓任意协议的包，linux只能抓ICMP
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))
# set socket option 
sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
# 系统为windows，则开启混杂模式
if os.name == "nt":
    sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)
 
#######################    
try:
    while True:
        raw_buffer = sniffer.recvfrom(65565)[0]
        ip_header = IP(raw_buffer[0:20])
        print "Protocol: %s  *** %s -> %s" % (ip_header.protocol, ip_header.src_address, ip_header.dst_address)
except KeyboardInterrupt, e:
    print e

#############        

#print sniffer.recvfrom(65565)
# 系统为windows，则关闭混杂模式
if os.name == "nt":
    sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)

ICMP_body

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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
# -*- coding: UTF-8 -*-

import socket
import os

###  新 加 ####
import struct
from ctypes import *
##############

host = "192.168.66.190"
#######################  IP CLASS   ##############################
class IP(Structure):
    _fields_ = [
        ("ih1",          c_ubyte, 4),
        ("version",      c_ubyte, 4),
        ("tos",          c_ubyte),
        ("len",          c_ushort),
        ("id",           c_ushort),
        ("offset",       c_ushort),
        ("ttl",          c_ubyte),
        ("protocol_num", c_ubyte),
        ("sum",          c_ushort),
        ("src",          c_ulong),
        ("dst",          c_ulong)
    ]
    
    def __new__(self, socket_buffer = None):
        return self.from_buffer_copy(socket_buffer)
    def __init__(self, socket_buffer = None):
        self.protocol_map = {1:"ICMP", 6:"TCP", 17:"UDP"}
      	self.src_address = socket.inet_ntoa(struct.pack("<L",self.src))
        self.dst_address = socket.inet_ntoa(struct.pack("<L",self.dst))
        try:
            self.protocol = self.protocol_map[self.protocol_num]
        except:
            self.protocol = str(self.protocol_num)
######################################################################

##########   ICMP CLASS ############################################

class ICMP(Structure):
    _fields_ = [
        ("type",          c_ubyte),
        ("code",          c_ubyte),
        ("checksum",      c_ushort),
        ("unused",        c_ushort),
        ("next_hop_mtu",  c_ushort)
    ]
    
    def __new__(self, socket_buffer):
        return self.from_buffer_copy(socket_buffer)
    
    def __init__(self, socket_buffer):
        pass
    
    
####################################################################

# 判断是否为windows，windows可以抓任意协议的包，linux只能抓ICMP
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))
# set socket option 
sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
# 系统为windows，则开启混杂模式
if os.name == "nt":
    sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)
 
#######################    
try:
    while True:
        raw_buffer = sniffer.recvfrom(65565)[0]
        ip_header = IP(raw_buffer[0:20])
        print "Protocol: %s  *** %s -> %s" % (ip_header.protocol, ip_header.src_address, ip_header.dst_address)
        
        if ip_header.protocol == "ICMP":
            offset = ip_header.ih1 * 4
            buf = raw_buffer[offset:offset + sizeof(ICMP)]
            
            icmp_header = ICMP(buf)
            
            print "ICMP —> TYPE: %d CODE: %d" % (icmp_header.type, icmp_header.code)


except KeyboardInterrupt, e:
    print e

#############   

#print sniffer.recvfrom(65565)
# 系统为windows，则关闭混杂模式
if os.name == "nt":
    sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)

计算IP头部长度，直接用报文中的头部长度 乘以 4 个字节。

接下来，我们再把扫描模块给写完：

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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
# -*- coding: UTF-8 -*-
import threading
import socket
import os
import time

from netaddr import IPNetwork, IPAddress



###  新 加 ####
import struct
from ctypes import *
##############

host = "192.168.66.190"

subnet = "192.168.66.0/24"
### include ICMP responce body ###
magic_message = "PYTHONRULES!"

def udp_sender(subnet, magic_message):
    time.sleep(5)
    sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    for ip in IPNetwork(subnet):
        try:
            sender.sendto(magic_message, ("%s" % ip, 65212))
        except:
            pass
        
#######################  IP CLASS   ##############################
class IP(Structure):
    _fields_ = [
        ("ih1",          c_ubyte, 4),
        ("version",      c_ubyte, 4),
        ("tos",          c_ubyte),
        ("len",          c_ushort),
        ("id",           c_ushort),
        ("offset",       c_ushort),
        ("ttl",          c_ubyte),
        ("protocol_num", c_ubyte),
        ("sum",          c_ushort),
        ("src",          c_ulong),
        ("dst",          c_ulong)
    ]
    
    def __new__(self, socket_buffer = None):
        return self.from_buffer_copy(socket_buffer)
    def __init__(self, socket_buffer = None):
        self.protocol_map = {1:"ICMP", 6:"TCP", 17:"UDP"}
      	self.src_address = socket.inet_ntoa(struct.pack("<L",self.src))
        self.dst_address = socket.inet_ntoa(struct.pack("<L",self.dst))
        try:
            self.protocol = self.protocol_map[self.protocol_num]
        except:
            self.protocol = str(self.protocol_num)
######################################################################

##########   ICMP CLASS ############################################

class ICMP(Structure):
    _fields_ = [
        ("type",          c_ubyte),
        ("code",          c_ubyte),
        ("checksum",      c_ushort),
        ("unused",        c_ushort),
        ("next_hop_mtu",  c_ushort)
    ]
    
    def __new__(self, socket_buffer):
        return self.from_buffer_copy(socket_buffer)
    
    def __init__(self, socket_buffer):
        pass
    
    
####################################################################

t = threading.Thread(target = udp_sender, args = (subnet, magic_message))
t.start()



# 判断是否为windows，windows可以抓任意协议的包，linux只能抓ICMP
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))
# set socket option 
sniffer.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
# 系统为windows，则开启混杂模式
if os.name == "nt":
    sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)
 
#######################    
try:
    while True:
        raw_buffer = sniffer.recvfrom(65565)[0]
        ip_header = IP(raw_buffer[0:20])
        print "Protocol: %s  *** %s -> %s" % (ip_header.protocol, ip_header.src_address, ip_header.dst_address)
        
        if ip_header.protocol == "ICMP":
            offset = ip_header.ih1 * 4
            buf = raw_buffer[offset:offset + sizeof(ICMP)]
            
            icmp_header = ICMP(buf)
            
            print "ICMP -> TYPE: %d CODE: %d" % (icmp_header.type, icmp_header.code)
            
            if icmp_header.code == 3 and icmp_header.type == 3:
                if raw_buffer[len(raw_buffer)-len(magic_message):] == magic_message:
                    print "Host Up: %s" % ip_header.src_address

except KeyboardInterrupt, e:
    print e

#############   

#print sniffer.recvfrom(65565)
# 系统为windows，则关闭混杂模式
if os.name == "nt":
    sniffer.ioctl(socket.SIO_RCVALL, socket.RCVALL_OFF)
-----本文结束感谢您的阅读-----
warcup wechat
欢迎扫描二维码关注我的公众号~~~
喜 欢 这 篇 文 章 吗 ？