% send, sniff, sr* tests for Scapy

~ needs_root

############
############
+ Test bridge_and_sniff() using tap sockets

~ tap

= Create two tap interfaces

import subprocess
from threading import Thread

tap0, tap1 = [TunTapInterface("tap%d" % i) for i in range(2)]

chk_kwargs = {"timeout": 3}

if LINUX:
    for i in range(2):
        assert subprocess.check_call(["ip", "link", "set", "tap%d" % i, "up"], **chk_kwargs) == 0
else:
    for i in range(2):
        assert subprocess.check_call(["ifconfig", "tap%d" % i, "up"], **chk_kwargs) == 0

= Run a sniff thread on the tap1 **interface**
* It will terminate when 5 IP packets from 192.0.2.1 have been sniffed
started = threading.Event()
t_sniff = Thread(
    target=sniff,
    kwargs={"iface": "tap1", "count": 5, "prn": Packet.summary,
            "lfilter": lambda p: IP in p and p[IP].src == "192.0.2.1",
            "started_callback": started.set},
    name="tests sniff 1")
t_sniff.start()
started.wait(timeout=5)

= Run a bridge_and_sniff thread between the taps **sockets**
* It will terminate when 5 IP packets from 192.0.2.1 have been forwarded
started = threading.Event()
t_bridge = Thread(target=bridge_and_sniff, args=(tap0, tap1),
                  kwargs={"store": False, "count": 5, 'prn': Packet.summary,
                          "lfilter": lambda p: IP in p and p[IP].src == "192.0.2.1",
                          "started_callback": started.set},
                  name="tests bridge_and_sniff 1")
t_bridge.start()
started.wait(timeout=5)

= Send five IP packets from 192.0.2.1 to the tap0 **interface**
sendp([Ether(dst=ETHER_BROADCAST) / IP(src="192.0.2.1") / ICMP()], iface="tap0",
      count=5)

= Wait for the threads
t_bridge.join(5)
t_sniff.join(5)
assert not t_bridge.is_alive()
assert not t_sniff.is_alive()

= Run a sniff thread on the tap1 **interface**
* It will terminate when 5 IP packets from 198.51.100.1 have been sniffed
started = threading.Event()
t_sniff = Thread(
    target=sniff,
    kwargs={"iface": "tap1", "count": 5, "prn": Packet.summary,
            "lfilter": lambda p: IP in p and p[IP].src == "198.51.100.1",
            "started_callback": started.set},
    name="tests sniff 2")
t_sniff.start()
started.wait(timeout=5)

= Run a bridge_and_sniff thread between the taps **sockets**
* It will "NAT" packets from 192.0.2.1 to 198.51.100.1 and will terminate when 5 IP packets have been forwarded
def nat_1_2(pkt):
    if IP in pkt and pkt[IP].src == "192.0.2.1":
        pkt[IP].src = "198.51.100.1"
        del pkt[IP].chksum
        return pkt
    return False

started = threading.Event()
t_bridge = Thread(target=bridge_and_sniff, args=(tap0, tap1),
                  kwargs={"store": False, "count": 5, 'prn': Packet.summary,
                          "xfrm12": nat_1_2,
                          "lfilter": lambda p: IP in p and p[IP].src == "192.0.2.1",
                          "started_callback": started.set},
                  name="tests bridge_and_sniff 2")
t_bridge.start()
started.wait(timeout=5)

= Send five IP packets from 192.0.2.1 to the tap0 **interface**
sendp([Ether(dst=ETHER_BROADCAST) / IP(src="192.0.2.1") / ICMP()], iface="tap0",
      count=5)

= Wait for the threads
t_bridge.join(5)
t_sniff.join(5)
assert not t_bridge.is_alive()
assert not t_sniff.is_alive()

= Delete the tap interfaces
if conf.use_pypy:
    # See https://pypy.readthedocs.io/en/latest/cpython_differences.html
    tap0.close()
    tap1.close()
else:
    del tap0, tap1


############
############
+ Test bridge_and_sniff() using tun sockets

~ tun not_libpcap

= Create two tun interfaces

import subprocess
from threading import Thread

tun0, tun1 = [TunTapInterface("tun%d" % i) for i in range(2)]

chk_kwargs = {"timeout": 3}

if LINUX:
    for i in range(2):
        assert subprocess.check_call(["ip", "link", "set", "tun%d" % i, "up"], **chk_kwargs) == 0
    assert subprocess.check_call([
        "ip", "addr", "change",
        "192.0.2.1", "peer", "192.0.2.2", "dev", "tun0"], **chk_kwargs) == 0
else:
    for i in range(2):
        assert subprocess.check_call(["ifconfig", "tun%d" % i, "up"], **chk_kwargs) == 0
    assert subprocess.check_call(["ifconfig", "tun0", "192.0.2.1", "192.0.2.2"], **chk_kwargs) == 0

= Run a sniff thread on the tun1 **interface**
* It will terminate when 5 IP packets from 192.0.2.1 have been sniffed
started = threading.Event()
t_sniff = Thread(target=sniff,
                 kwargs={"iface": "tun1", "count": 5,
                         "prn": Packet.summary,
                         "lfilter": lambda p: IP in p and p[IP].src == "192.0.2.1",
                         "started_callback": started.set},
                 name="tests sniff 3")

t_sniff.start()
started.wait(timeout=5)

= Run a bridge_and_sniff thread between the tuns **sockets**
* It will terminate when 5 IP packets from 192.0.2.1 have been forwarded.
started = threading.Event()
t_bridge = Thread(target=bridge_and_sniff, args=(tun0, tun1),
                  kwargs={"store": False, "count": 5, 'prn': Packet.summary,
                          "xfrm12": lambda pkt: pkt,
                          "lfilter": lambda p: IP in p and p[IP].src == "192.0.2.1",
                          "started_callback": started.set},
                          name="tests bridge_and_sniff 3")
t_bridge.start()
started.wait(timeout=5)

= Send five IP packets from 192.0.2.1 to the tun0 **interface**
conf.route.add(net="192.0.2.2/32", dev="tun0")
send([IP(src="192.0.2.1", dst="192.0.2.2") / ICMP()], count=5, iface="tun0")
conf.route.delt(net="192.0.2.2/32", dev="tun0")

= Wait for the threads
t_bridge.join(5)
t_sniff.join(5)
assert not t_bridge.is_alive()
assert not t_sniff.is_alive()

= Run a sniff thread on the tun1 **interface**
* It will terminate when 5 IP packets from 198.51.100.1 have been sniffed
started = threading.Event()
t_sniff = Thread(target=sniff,
                 kwargs={"iface": "tun1", "count": 5, "prn": Packet.summary,
                         "lfilter": lambda p: IP in p and p[IP].src == "198.51.100.1",
                         "started_callback": started.set},
                 name="tests sniff 4")

t_sniff.start()
started.wait(timeout=5)

= Run a bridge_and_sniff thread between the tuns **sockets**
* It will "NAT" packets from 192.0.2.1 to 198.51.100.1 and will terminate when 5 IP packets have been forwarded
def nat_1_2(pkt):
    if IP in pkt and pkt[IP].src == "192.0.2.1":
        pkt[IP].src = "198.51.100.1"
        del pkt[IP].chksum
        return pkt
    return False

started = threading.Event()
t_bridge = Thread(target=bridge_and_sniff, args=(tun0, tun1),
                  kwargs={"store": False, "count": 5, 'prn': Packet.summary,
                          "xfrm12": nat_1_2,
                          "lfilter": lambda p: IP in p and p[IP].src == "192.0.2.1",
                          "started_callback": started.set},
                          name="tests bridge_and_sniff 4")
t_bridge.start()
started.wait(timeout=5)

= Send five IP packets from 192.0.2.1 to the tun0 **interface**
conf.route.add(net="192.0.2.2/32", dev="tun0")
send([IP(src="192.0.2.1", dst="192.0.2.2") / ICMP()], count=5, iface="tun0")
conf.route.delt(net="192.0.2.2/32", dev="tun0")

= Wait for the threads
t_bridge.join(5)
t_sniff.join(5)
assert not t_bridge.is_alive()
assert not t_sniff.is_alive()

= Delete the tun interfaces
if conf.use_pypy:
    # See https://pypy.readthedocs.io/en/latest/cpython_differences.html
    tun0.close()
    tun1.close()
else:
    del tun0, tun1


############
############
+ Test bridge_and_sniff() using veth pairs
~ linux needs_root veth

= Ensure bridge_and_sniff does not close sockets if data is send within xfrm on ingress interface

from scapy.arch.linux import VEthPair

with VEthPair('a_0', 'a_1') as veth_0:
    with VEthPair('b_0', 'b_1') as veth_1:
        xfrm_count = {
            'a_0':0,
            'b_0': 0
        }
        def xfrm_x(pkt):
            pkt_tx = pkt.copy()
            ether_lyr = pkt_tx[Ether]
            ether_lyr.type = 0x1234  # we send to peer interface - avoid loop
            # send on receiving interface - triggers return None on recv() in L2Socket
            sendp(pkt_tx, iface=pkt.sniffed_on)
            global xfrm_count
            xfrm_count[pkt.sniffed_on] = xfrm_count[pkt.sniffed_on] + 1
            return True
        started = threading.Event()
        t_bridge = Thread(target=bridge_and_sniff,
                          args=('a_0', 'b_0'),
                          kwargs={
                              'xfrm12': xfrm_x,
                              'xfrm21': xfrm_x,
                              'store': False,
                              'count': 4,
                              'lfilter': lambda p: Ether in p and p[Ether].type == 0xbeef,
                              "started_callback": started.set},
                          name="tests bridge_and_sniff VEthPair")
        t_bridge.start()
        started.wait(timeout=5)
        # send frames in both directions
        for if_name in ['a_1', 'b_1', 'a_1', 'b_1']:
            sendp([Ether(type=0xbeef) /
                   Raw(b'On a scale from one to ten what is your favourite colour of the alphabet?')],
                  iface=if_name)
        t_bridge.join(1)
        # now test of the socket used in bridge_and_sniff() was alive all the time
        assert (xfrm_count['a_0'] == 2)
        assert (xfrm_count['b_0'] == 2)


############
############
+ Test arpleak() using a tap socket

~ tap tcpdump

= Create a tap interface

from unittest import mock
import struct
import subprocess
from threading import Thread
import time

tap0 = TunTapInterface("tap0")

chk_kwargs = {"timeout": 3}

if LINUX:
    assert subprocess.check_call(["ip", "link", "set", "tap0", "up"], **chk_kwargs) == 0
else:
    assert subprocess.check_call(["ifconfig", "tap0", "up"], **chk_kwargs) == 0

= Check for arpleak

def answer_arp_leak(pkt):
    mymac = b"\x00\x01\x02\x03\x04\x06"
    myip = b"\xc0\x00\x02\x02" # 192.0.2.2
    if not ARP in pkt:
        return
    e_src = pkt.src
    pkt = raw(pkt[ARP])
    if pkt[:4] != b'\x00\x01\x08\x00':
        print("Invalid ARP")
        return
    hwlen, plen, op = struct.unpack('>BBH', pkt[4:8])
    if op != 1:
        print("Invalid ARP op")
        return
    fmt = ('%ds%ds' % (hwlen, plen)) * 2
    hwsrc, psrc, hwdst, pdst = struct.unpack(fmt,
                                             pkt[8:8 + (plen + hwlen) * 2])
    if pdst[:4] != myip[:plen]:
        print("Invalid ARP pdst %r" % pdst)
        return
    ans = Ether(dst=e_src, src=mymac, type=0x0806)
    ans /= (b'\x00\x01\x08\x00' +
            struct.pack('>BBH' + fmt,
                        hwlen, plen, 2, mymac, myip, hwsrc, psrc))
    tap0.send(ans)
    print('Answered!')

started = threading.Event()
t_answer = Thread(
    target=sniff,
    kwargs={"prn": answer_arp_leak, "timeout": 10, "store": False,
            "opened_socket": tap0,
            "started_callback": started.set},
    name="tests answer_arp_leak")

t_answer.start()
started.wait(timeout=5)

@mock.patch("scapy.layers.l2.get_if_addr")
@mock.patch("scapy.layers.l2.get_if_hwaddr")
def test_arpleak(mock_get_if_hwaddr, mock_get_if_addr, hwlen=255, plen=255):
    conf.route.ifadd("tap0", "192.0.2.0/24")
    mock_get_if_addr.side_effect = lambda _: "192.0.2.1"
    mock_get_if_hwaddr.side_effect = lambda _: "00:01:02:03:04:05"
    return arpleak("192.0.2.2/31", timeout=2, hwlen=hwlen, plen=plen)

ans, unans = test_arpleak()
assert len(ans) == 1
assert len(unans) == 1
ans, unans = test_arpleak(hwlen=6)
assert len(ans) == 1
assert len(unans) == 1
ans, unans = test_arpleak(plen=4)
assert len(ans) == 1
assert len(unans) == 1

t_answer.join(15)

if t_answer.is_alive():
    raise Exception("Test timed out")

if conf.use_pypy:
    # See https://pypy.readthedocs.io/en/latest/cpython_differences.html
    tap0.close()
else:
    del tap0

#####
#####
+ Test sr() on multiple interfaces

= Setup multiple linux interfaces and ranges
~ linux needs_root dbg

import os
exit_status = os.system("ip netns add blob0")
exit_status |= os.system("ip netns add blob1")
exit_status |= os.system("ip link add name scapy0.0 type veth peer name scapy0.1")
exit_status |= os.system("ip link add name scapy1.0 type veth peer name scapy1.1")
exit_status |= os.system("ip link set scapy0.1 netns blob0 up")
exit_status |= os.system("ip link set scapy1.1 netns blob1 up")
exit_status |= os.system("ip addr add 100.64.2.1/24 dev scapy0.0")
exit_status |= os.system("ip addr add 100.64.3.1/24 dev scapy1.0")
exit_status |= os.system("ip --netns blob0 addr add 100.64.2.2/24 dev scapy0.1")
exit_status |= os.system("ip --netns blob1 addr add 100.64.3.2/24 dev scapy1.1")
exit_status |= os.system("ip link set scapy0.0 up")
exit_status |= os.system("ip link set scapy1.0 up")
assert exit_status == 0

conf.ifaces.reload()
conf.route.resync()

try:
    pkts = sr(IP(dst=["100.64.2.2", "100.64.3.2"])/ICMP(), timeout=1)[0]
    assert len(pkts) == 2
    assert pkts[0].answer.src in ["100.64.2.2", "100.64.3.2"]
    assert pkts[1].answer.src in ["100.64.2.2", "100.64.3.2"]
finally:
    e = os.system("ip netns del blob0")
    e = os.system("ip netns del blob1")
    conf.ifaces.reload()
    conf.route.resync()


= sr() performance test
~ linux needs_root veth not_pypy

import subprocess
import shlex

try:
    # Create a dedicated network name space to simulate remote host
    subprocess.check_call(shlex.split("sudo ip netns add scapy"))
    # Create a virtual Ethernet pair to connect default and new NS
    subprocess.check_call(shlex.split("sudo ip link add type veth"))
    # Move veth1 to the new NS
    subprocess.check_call(shlex.split("sudo ip link set veth1 netns scapy"))
    # Setup vNIC in the default NS
    subprocess.check_call(shlex.split("sudo ip link set veth0 up"))
    subprocess.check_call(shlex.split("sudo ip addr add 192.168.168.1/24 dev veth0"))
    # Setup vNIC in the dedicated NS
    subprocess.check_call(shlex.split("sudo ip netns exec scapy ip link set lo up"))
    subprocess.check_call(shlex.split("sudo ip netns exec scapy ip link set veth1 up"))
    subprocess.check_call(shlex.split("sudo ip netns exec scapy ip addr add 192.168.168.2/24 dev veth1"))
    # Perform test
    conf.route.resync()
    res, unansw = sr(IP(dst='192.168.168.2') / ICMP(seq=(1, 1000)), timeout=1, verbose=False)
finally:
    try:
        # Bring down the interfaces
        subprocess.check_call(shlex.split("sudo ip netns exec scapy ip link set veth1 down"))
        subprocess.check_call(shlex.split("sudo ip netns exec scapy ip link set lo down"))
        # Delete the namespace
        subprocess.check_call(shlex.split("sudo ip netns delete scapy"))
        # Remove the virtual Ethernet pair
        subprocess.check_call(shlex.split("sudo ip link delete veth0"))
    except subprocess.CalledProcessError as e:
        print(f"Error during cleanup: {e}")

len(res) == 1000