% tuntap tests for Scapy

# Packet capture-based tests are in sendsniff.uts

#######
+ Test Linux-specific protocol headers for TunTap
~ linux tun not_libpcap

= Linux-specific protocol headers

p = LinuxTunPacketInfo()/IP()
assert p.type == 2048

p = LinuxTunPacketInfo(raw(p))
assert p.type == 2048
assert isinstance(p.payload, IP)

p = LinuxTunPacketInfo()/IPv6()
assert p.type == 0x86dd

p = LinuxTunPacketInfo(raw(p))
assert p.type == 0x86dd

assert isinstance(p.payload, IPv6)

#######
+ Test tun device

~ tun needs_root not_libpcap

= Create a tun interface

import subprocess
from threading import Thread

tun0 = TunTapInterface("tun0", strip_packet_info=False)

if LINUX:
    assert subprocess.check_call(["ip", "link", "set", "tun0", "up"]) == 0
    assert subprocess.check_call([
        "ip", "addr", "change",
        "192.0.2.1", "peer", "192.0.2.2", "dev", "tun0"]) == 0
elif BSD:
    assert subprocess.check_call(["ifconfig", "tun0", "up"]) == 0
    assert subprocess.check_call([
        "ifconfig", "tun0", "192.0.2.1", "192.0.2.2"]) == 0
else:
    raise NotImplementedError()

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

= Setup ICMPEcho_am on the interface

am = tun0.am(ICMPEcho_am, count=3)
am.defoptsniff['timeout'] = 5
t_am = Thread(target=am)
t_am.start()

= Send ping packets from OS into scapy

send(IP(dst="192.0.2.2")/ICMP(seq=(1,3)))

= Cleanup

t_am.join(timeout=3)

tun0.close()

#######
+ Test strip_packet_info=False on Linux

~ tun linux needs_root not_libpcap

= Create a tun interface

if not LINUX:
    raise NotImplementedError()

import subprocess

tun0 = TunTapInterface("tun0", strip_packet_info=False)

assert subprocess.check_call(["ip", "link", "set", "tun0", "up"]) == 0
assert subprocess.check_call([
    "ip", "addr", "change",
    "192.0.2.1", "peer", "192.0.2.2", "dev", "tun0"]) == 0

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

= Send ping packets from Linux into Scapy

def cb():
    send(IP(dst="192.0.2.2")/ICMP(seq=(1,3)))

t = AsyncSniffer(opened_socket=tun0, lfilter=lambda x: ICMP in x, started_callback=cb, count=3)
t.start()
t.join(timeout=3)

assert len(t.results) >= 3
icmp4_sequences = set()

for pkt in t.results:
    pkt
    assert isinstance(pkt, LinuxTunPacketInfo)
    if not isinstance(pkt.payload, IP) or ICMP not in pkt:
        # We might get IPv6 router solicitation or other traffic...
        continue
    if pkt[IP].src != '192.0.2.1' or pkt[IP].dst != '192.0.2.2' or pkt[ICMP].type != 8:
        continue
    icmp4_sequences.add(pkt.seq)

# Expect to get 3 different ICMP sequence numbers
assert len(icmp4_sequences) == 3

= Delete the tun interface
tun0.close()

+ Test strip_packet_info=True and IPv6

~ tun needs_root ipv6 not_libpcap

= Create a tun interface with IPv4 + IPv6

import subprocess

tun0 = TunTapInterface("tun0", strip_packet_info=True)

if LINUX:
    assert subprocess.check_call(["ip", "link", "set", "tun0", "up"]) == 0
    assert subprocess.check_call([
        "ip", "addr", "change",
        "192.0.2.1", "peer", "192.0.2.2", "dev", "tun0"]) == 0
    assert subprocess.check_call([
        "ip", "-6", "addr", "add",
        "2001:db8::1", "peer", "2001:db8::2", "dev", "tun0"]) == 0
elif BSD:
    assert subprocess.check_call(["ifconfig", "tun0", "up"]) == 0
    assert subprocess.check_call([
        "ifconfig", "tun0", "192.0.2.1", "192.0.2.2"]) == 0
    assert subprocess.check_call([
        "ifconfig", "tun0", "inet6", "2001:db8::1/128", "2001:db8::2"]) == 0
else:
    raise NotImplementedError()

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

= Send ping packets from OS into Scapy

def cb():
    send(IP(dst="192.0.2.2")/ICMP(seq=(1,3)))
    send(IPv6(dst="2001:db8::2")/ICMPv6EchoRequest(seq=(1,3)))

t = AsyncSniffer(opened_socket=tun0, lfilter=lambda x: ICMP in x or ICMPv6EchoRequest in x, started_callback=cb, count=6)
t.start()
t.join(timeout=3)

assert len(t.results) >= 6
icmp4_sequences = set()
icmp6_sequences = set()

for pkt in t.results:
    pkt
    assert isinstance(pkt, (IP, IPv6))
    if (isinstance(pkt, IP) and
        pkt[IP].src == "192.0.2.1" and pkt[IP].dst == "192.0.2.2" and
        ICMP in pkt and pkt[ICMP].type == 8):
        icmp4_sequences.add(pkt[ICMP].seq)
    if (isinstance(pkt, IPv6) and
        pkt[IPv6].src == "2001:db8::1" and pkt[IPv6].dst == "2001:db8::2" and
        ICMPv6EchoRequest in pkt):
        icmp6_sequences.add(pkt[ICMPv6EchoRequest].seq)

# Expect to get 3 different ICMP sequence numbers
assert len(icmp4_sequences) == 3, (
    "Expected 3 IPv4 ICMP ping packets, got: " + repr(icmp4_sequences))
assert len(icmp6_sequences) == 3, (
    "Expected 3 IPv6 ICMP ping packets, got: " + repr(icmp6_sequences))

= Delete the tun interface
tun0.close()

+ Test tap interfaces

~ tap needs_root not_libpcap

= Create a tap interface with IPv4

import subprocess

tap0 = TunTapInterface("tap0")

if LINUX:
    assert subprocess.check_call(["ip", "link", "set", "tap0", "up"]) == 0
    assert subprocess.check_call([
        "ip", "addr", "change", "192.0.2.1/30", "dev", "tap0"]) == 0
    assert subprocess.check_call([
        "ip", "neigh", "replace",
        "192.0.2.2", "lladdr", "20:00:00:20:00:00", "dev", "tap0"]) == 0
else:
    assert subprocess.check_call(["ifconfig", "tap0", "up"]) == 0
    assert subprocess.check_call([
        "ifconfig", "tap0", "192.0.2.1", "netmask", "255.255.255.252"]) == 0
    assert subprocess.check_call([
        "arp", "-s", "192.0.2.2", "20:00:00:20:00:00", "temp"]) == 0

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

= Send ping packets from OS into Scapy

conf.ifaces
conf.route

def cb():
    sendp(Ether(dst="ff:ff:ff:ff:ff:ff")/IP(dst="192.0.2.2")/ICMP(seq=(1,3)), iface="tap0")

t = AsyncSniffer(opened_socket=tap0, lfilter=lambda x: ICMP in x, started_callback=cb, count=3)
t.start()
t.join(timeout=3)

assert len(t.results) >= 3
icmp4_sequences = set()

for pkt in t.results:
    pkt
    assert isinstance(pkt, Ether)
    if (IP in pkt and
        pkt[IP].src == "192.0.2.1" and pkt[IP].dst == "192.0.2.2" and
        ICMP in pkt and pkt[ICMP].type == 8):
        icmp4_sequences.add(pkt[ICMP].seq)

# Expect to get 3 different ICMP sequence numbers
assert len(icmp4_sequences) == 3, (
    "Expected 3 IPv4 ICMP ping packets, got: " + repr(icmp4_sequences))

= Delete the tap interface
tap0.close()

+ Refresh interfaces
~ linux tun tap not_libpcap

= Cleanup

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