Network programming with Python Scapy

A memorandum of network programming in Python.

Install Python Scapy

Install by referring to the following site. The environment is MAC OS X 10.11.6. http://nigaky.hatenablog.com/entry/20110716/1310813250

However, even if you install only scapy, you cannot use it, so you need to install pcapy as well. From the following site, download the source and install it. https://pypi.python.org/pypi/pcapy

See below for more information on scapy. http://scapy.readthedocs.io/

Generate and send ARP packets

Code to send an ARP request message to get the MAC address of a specific IP address.

ARP.py

from scapy.all import *
target_ip="192.168.1.1"
frame = Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(op=1, pdst=target_ip)
sendp(frame)

Generate and send ICMP packets

The code that sends an ICMP packet to a specific IP address.

ICMP.py

from scapy.all import *
target_ip="192.168.1.1"
frame = Ether() / IP(dst=target_ip) / ICMP()
sendp(frame)

Generate and send TCP SYN packets

A code that sends a TCP SYN packet to a specific IP, a specific port.

ICMP.py

from scapy.all import *
target_ip="192.168.1.1"
dst_port = 5001
src_port = 5002
frame = IP(dst=target_ip)/TCP(flags = 'S',sport=src_port,dport=dst_port)
send(frame)

Sniff received packets in parallel processing

SniffRecPkt.py

import threading
from scapy.all import *

class SniffRecPkt(threading.Thread):
    def __init__(self,target_ip):
        super(RecPingScan, self).__init__()
        self.target_ip = target_ip
        self.stop_event = threading.Event() #Flag to stop
        self.thread = threading.Thread(target = self.run)
        self.thread.start()

    def run(self):
        while not self.stop_event.is_set():
            sniff(filter="tcp and ip src host " + self.target_ip,prn=packet_show, count=1)

    def stop(self):
        """Stop the thread"""
        self.stop_event.set()
        self.thread.join()    #Wait for the thread to stop


def packet_show(packet):
    if packet[TCP].flags==18: #SYN/Only when it was an ACK packet
        print "IP : " + str(packet[IP].src) + " | TCP PORT : " + str(packet[TCP].sport)

if __name__ == '__main__':
    target_ip = "192.168.1.1"
    Rec_thread=SniffRecPkt(target_ip)

    target_ip="192.168.1.1"
    dst_port = 5001
    src_port = 5002
    frame = IP(dst=target_ip)/TCP(flags = 'S',sport=src_port,dport=dst_port)
    send(frame)

    Rec_thread.stop()

Port SCAN programming

python:port_scan_v1.0.py

# encoding: utf-8
from scapy.all import *
import netifaces
import threading
import time
import sys
import random

class RecPingScan(threading.Thread):
    def __init__(self,target_ip):
        super(RecPingScan, self).__init__()
        self.target_ip = target_ip
        self.stop_event = threading.Event() #Flag to stop
        self.thread = threading.Thread(target = self.run)
        self.thread.start()

    def run(self):
        while not self.stop_event.is_set():
            sniff(filter="tcp and ip src host " + self.target_ip,prn=packet_show, count=1)

    def stop(self):
        """Stop the thread"""
        self.stop_event.set()
        self.thread.join()    #Wait for the thread to stop


def packet_show(packet):
    if packet[TCP].flags==18:
        print "IP : " + str(packet[IP].src) + " | TCP PORT : " + str(packet[TCP].sport)


def send_tcpsyn(target_ip):
    sport = random.randint(50000,51000)
    for i in range(0,65535):
        frame = IP(dst=target_ip)/TCP(flags = 'S',sport=sport,dport=i)
        send(frame)
        send(frame)

if __name__ == '__main__':
    target_ip = "192.168.1.1"

    Rec_thread=RecPingScan(target_ip)
    send_tcpsyn(target_ip)
    time.sleep(2)
    Rec_thread.stop()
    sys.exit()