blob: 2ed25c987fefff4aa2894339b1226eee9212bd2c [file] [log] [blame]
# Copyright 2017 The Android Open Source Project
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
"""A utility for "twisting" packets on a tun/tap interface.
TunTwister and TapTwister echo packets on a tun/tap while swapping the source
and destination at the ethernet and IP layers. This allows sockets to
effectively loop back packets through the full networking stack, avoiding any
shortcuts the kernel may take for actual IP loopback. Additionally, users can
inspect each packet to assert testing invariants.
import os
import select
import threading
from scapy import all as scapy
class TunTwister(object):
"""TunTwister transports traffic travelling twixt two terminals.
TunTwister is a context manager that will read packets from a tun file
descriptor, swap the source and dest of the IP header, and write them back.
To use this class, tests also need to set up routing so that packets will be
routed to the tun interface.
Two sockets can communicate with each other through a TunTwister as if they
were each connecting to a remote endpoint. Both sockets will have the
perspective that the address of the other is a remote address.
Packet inspection can be done with a validator function. This can be any
function that takes a scapy packet object as its only argument. Exceptions
raised by your validator function will be re-raised on the main thread to fail
your tests.
NOTE: Exceptions raised by a validator function will supercede exceptions
raised in the context.
def testFeatureFoo(self):
my_tun = MakeTunInterface()
# Set up routing so packets go to my_tun.
def ValidatePortNumber(packet):
self.assertEquals(8080, packet.getlayer(scapy.UDP).sport)
self.assertEquals(8080, packet.getlayer(scapy.UDP).dport)
with TunTwister(tun_fd=my_tun, validator=ValidatePortNumber):
sock = socket(AF_INET, SOCK_DGRAM, 0)
sock.bind(("", 8080))
sock.sendto("hello", ("", 8080))
data, addr = sock.recvfrom(1024)
self.assertEquals("hello", data)
self.assertEquals(("", 8080), addr)
# Hopefully larger than any packet.
def __init__(self, fd=None, validator=None):
"""Construct a TunTwister.
The TunTwister will listen on the given TUN fd.
The validator is called for each packet *before* twisting. The packet is
passed in as a scapy packet object, and is the only argument passed to the
fd: File descriptor of a TUN interface.
validator: Function taking one scapy packet object argument.
self._fd = fd
# Use a pipe to signal the thread to exit.
self._signal_read, self._signal_write = os.pipe()
self._thread = threading.Thread(target=self._RunLoop, name="TunTwister")
self._validator = validator
self._error = None
def __enter__(self):
def __exit__(self, *args):
# Signal thread exit.
os.write(self._signal_write, "bye")
if self._thread.isAlive():
raise RuntimeError("Timed out waiting for thread exit")
# Re-raise any error thrown from our thread.
if isinstance(self._error, Exception):
raise self._error # pylint: disable=raising-bad-type
def _RunLoop(self):
"""Twist packets until exit signal."""
while True:
read_fds, _, _ =[self._fd, self._signal_read], [], [],
if self._signal_read in read_fds:
if self._fd in read_fds:
except Exception as e: # pylint: disable=broad-except
self._error = e
def _Flush(self):
"""Ensure no packets are left in the buffer."""
p = select.poll()
p.register(self._fd, select.POLLIN)
while p.poll(TunTwister._POLL_FAST_TIMEOUT_MS):
def _ProcessPacket(self):
"""Read, twist, and write one packet on the tun/tap."""
# TODO: Handle EAGAIN "errors".
bytes_in =, TunTwister._READ_BUF_SIZE)
packet = self.DecodePacket(bytes_in)
# the user may wish to filter certain packets, such as
# Ethernet multicast packets
if self._DropPacket(packet):
if self._validator:
packet = self.TwistPacket(packet)
def _DropPacket(self, packet):
"""Determine whether to drop the provided packet by inspection"""
return False
def DecodePacket(cls, bytes_in):
"""Decode a byte array into a scapy object."""
return cls._DecodeIpPacket(bytes_in)
def TwistPacket(cls, packet):
"""Swap the src and dst in the IP header."""
ip_type = type(packet)
if ip_type not in (scapy.IP, scapy.IPv6):
raise TypeError("Expected an IPv4 or IPv6 packet.")
packet.src, packet.dst = packet.dst, packet.src
packet = ip_type( # Fix the IP checksum.
return packet
def _DecodeIpPacket(packet_bytes):
"""Decode 'packet_bytes' as an IPv4 or IPv6 scapy object."""
ip_ver = (ord(packet_bytes[0]) & 0xF0) >> 4
if ip_ver == 4:
return scapy.IP(packet_bytes)
elif ip_ver == 6:
return scapy.IPv6(packet_bytes)
raise ValueError("packet_bytes is not a valid IPv4 or IPv6 packet")
class TapTwister(TunTwister):
"""Test util for tap interfaces.
TapTwister works just like TunTwister, except it operates on tap interfaces
instead of tuns. Ethernet headers will have their sources and destinations
swapped in addition to IP headers.
def _IsMulticastPacket(eth_pkt):
return int(eth_pkt.dst.split(":")[0], 16) & 0x1
def __init__(self, fd=None, validator=None, drop_multicast=True):
"""Construct a TapTwister.
TapTwister works just like TunTwister, but handles both ethernet and IP
fd: File descriptor of a TAP interface.
validator: Function taking one scapy packet object argument.
drop_multicast: Drop Ethernet multicast packets
super(TapTwister, self).__init__(fd=fd, validator=validator)
self._drop_multicast = drop_multicast
def _DropPacket(self, packet):
return self._drop_multicast and self._IsMulticastPacket(packet)
def DecodePacket(cls, bytes_in):
return scapy.Ether(bytes_in)
def TwistPacket(cls, packet):
"""Swap the src and dst in the ethernet and IP headers."""
packet.src, packet.dst = packet.dst, packet.src
ip_layer = packet.payload
twisted_ip_layer = super(TapTwister, cls).TwistPacket(ip_layer)
packet.payload = twisted_ip_layer
return packet