首页 > 编程语言 >python渗透测试入门——流量嗅探器

python渗透测试入门——流量嗅探器

时间：2023-03-06 16:46:08浏览次数：40  
标签：protocol 入门 python IP self 嗅探器 header ICMP socket

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平台），并且把迄今为止扫描出来的主机都打印到屏幕上。

那到这里我们的流量嗅探工具已经算编写完成了，希望大家有所收获。

标签：protocol,入门,python,IP,self,嗅探器,header,ICMP,socket
From： https://www.cnblogs.com/yaolingyu/p/17184403.html

相关文章

  • Python机器视觉之人脸检测 （二）从摄像头读取视频流做人脸检测
    基于上一篇文章Python机器视觉之人脸检测（一）人脸检测的4中办法，做了这个案列，内容比较简单，文字就可以少一点。前提1、打开和关闭摄像头的基本操作需要会。2、使用Pyqt显示......
  • Python数据分析之K_Means
    1#-*-coding:utf-8-*-23importpandasaspd4#参数初始化5inputfile='../data/consumption_data.xls'#销量及其他属性数据6outputfile='../......
  • CSS 入门
    0x1CSS什么是CSS层叠（Cascading）样式（Style）表（Sheets）用来定义页面元素的样式（字体、颜色、位置、大小、动效等）CSS代码构成选择器{属性:属性值;(声明=......
  • java泛型简单入门
    泛型泛型泛指某一种类型（必须是引用类型） 明确时机：1.有时候创建对象的时候，明确泛型2.有时候调用一个方法时，明确泛型3.有时候创建一个类型，然后实现一个接口......
  • ChatGPT带你入门机器学习：逻辑回归模型博客和小红书风格文案一次搞定！
    打脸了顺手向大家演示一下如何用ChatGPT写技术博客吧，其实蛮简单的，特别需要操心的是它会一本正经的胡说八道，还信誓旦旦的。我们要审查它的回答，万不可全信。为了便于阅......
  • 自学Python
    myage_zhang=32count=0whilecount<3:myage=int(input("请输入你的年龄："))ifmyage==myage_zhang:print("恭喜你，回答正确!")break......
  • EF7创建模型入门篇
    在EF7中，创建一个模型是非常重要的步骤。本文将使用微软官方文档中的指南，来学习EF7中的创建模型篇，外加一点点个人理解。实体类型在EF7中，你需要使用modelBuilder.Entity......
  • python yaml变量替换
    使用场景:yaml文件的数据一般来说是写好的常量，每次读取再使用即可。但有些接口的数据每次都会变化则需要每次进行替换才能执行成功，或者当我们有多个相同字段的值都相同时......
  • MySQL安装入门第一篇
    【1】MySQL的版本：近期主要历史版本有5.0/5.1/5.5/5.6/5.7，目前最新版本是MySQL8.6.0曾经是个内部试验版本，已取消了。MySQL8.0的版本历史1)2016-09-12第一个DM(develop......
  • 如何查看一个python程序调用了哪些库？
    今天才知道有个简单的方法：安装pipreqs就可以了！pipinstallpipreqs进入你的工程所在目录，然后输入下面的命令，等待命令运行结束后，就会生成requirements.txt文件了。(p......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99