Implementing Layer-3 VPN Protocols in Python

In this post, we’ll explore a practical approach to implementing tunneling protocols in Python. Along the way, you’ll gain an understanding of key concepts such as tunneling, TUN/TAP interfaces, and packet encapsulation. This post will be the first in a multipart series, where I will also demonstrate how to analyze tunneling protocols in Wireshark.

VPNs and TUN/TAP Devices Link to heading

A VPN (Virtual Private Network) is a technology that allows network traffic to be securely tunneled from one system to another. By encapsulating packets using cryptographic methods and wrapping the ciphertext in larger packets, VPNs protect data in transit between the endpoints.

A very common way for bytes to enter and exit that tunnel is a virtual network adapter. Its main advantage is the option to apply many of the same tools and concepts used for physical network adapters. On Linux, there are at least two kinds of virtual adapters: TUN and TAP devices. TUN devices operate on layer 3 of the OSI model, the network layer. The most common layer-3 protocols are IPv4 and IPv6. As layer-3 devices, TUN adapters can ingest and emit IP packets, but not Ethernet frames. Network devices on both ends of the tunnel can communicate via IP routing, but are not in the same LAN. Tunneling Ethernet frames can be achieved with a TAP device, which operates on layer 2. While this allows systems to reach each other via broadcast, it also introduces overhead in terms of performance and complexity¹. This post will focus on layer-3 tunneling, but the results can easily be adapted for a layer-2 VPN.

Only a few lines of Python code are required to create a TUN device (taken from this blog post by Julia Evans):

device_name = "tun0"

name_bytes = device_name.encode()
assert len(name_bytes) < 16, "The interface name must be less than 16 bytes"

tuntap = open("/dev/net/tun", "r+b", buffering=0)

LINUX_IFF_TUN = 0x0001
LINUX_IFF_NO_PI = 0x1000
flags = LINUX_IFF_TUN | LINUX_IFF_NO_PI
ifs = struct.pack("16sH22s", name_bytes, flags, b"")

LINUX_TUNSETIFF = 0x400454CA
ioctl(tuntap, LINUX_TUNSETIFF, ifs)

The function returns a file-like object, i.e. you can call read() and write(...) on it. Linux by default does not assign an IP address, nor any routes to this interface. The following lines use the ip command to configure the device in the peer-to-peer configuration:

subprocess.run(("ip", "link", "set", device_name, "up"), check=True)
subprocess.run(
    ("ip", "addr", "add", local_ip, "peer", peer_ip, "dev", device_name),
    check=True,
)

In addition to the local TUN interface, the VPN needs a socket to connect to the remote peer (e.g. via the Internet). For a UDP-based VPN, the code is symmetrical on both peers:

with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as remote:
    remote.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    remote.bind(listen_address)

VPN Main Loop Link to heading

Once both the local TUN interface and the remote socket are set up, the centerpiece can be implemented: A main loop that simultaneously reads packets from the TUN adapter and forwards them via the VPN while also reading packets from the VPN to emit them at the TUN interface. This is a perfect application for the select system call:

    while True:
        rd_sockets, _, _ = select.select([tuntap, remote], [], [], 1.0)

        for sock in rd_sockets:
            if sock is tuntap:
                # Data from TUNTAP needs to be pumped to the peer
                data = tuntap.read(0xFFFF)
                data = prepare_data_for_sending(data, key)
                remote.sendto(data, peer_address)

            elif sock is remote:
                # Data from the peer needs to be pumped to TUNTAP
                data = remote.recv(0xFFFF)
                data = handle_received_data(data, key)
                if data:
                    tuntap.write(data)

During each iteration of the while loop, select will return the socket that has data available for reading (if any). The code reads the data and processes it.

Encapsulation Link to heading

As described earlier, the packets need to be wrapped and optionally encrypted. My implementation uses XOR “encryption” with a static, repeating key (hence the name CrapVPN):

def xor(data: bytes, key: bytes):
    """XOR two byte arrays, repeating the key"""
    retval = bytearray(data)
    for i, _ in enumerate(data):
        retval[i] = data[i] ^ key[i % len(key)]
    return bytes(retval)

The following diagram shows the encapsulation format:

packet-beta
0-31: "Magic (4 bytes)"
32-47: "Length (2 bytes)"
48-63: "Reserved (2 bytes)"
64-95: "Ciphertext (variable length)"

The entire encapsulation can be implemented with the following two functions:

CRAPVPN_HEADER = ">4sHxx"
CRAPVPN_MAGIC = b"crap"
CRAPVPN_HEADER_SIZE = struct.calcsize(CRAPVPN_HEADER)


def prepare_data_for_sending(data: bytes, key: bytes) -> bytes:
    """Encrypt and wrap data for sending via the VPN"""
    ciphertext = xor(data, key)
    return struct.pack(CRAPVPN_HEADER, CRAPVPN_MAGIC, len(ciphertext)) + ciphertext


def handle_received_data(data: bytes, key: bytes) -> bytes | None:
    """Unwrap and decrypt data from the VPN"""
    magic, length = struct.unpack(CRAPVPN_HEADER, data[:CRAPVPN_HEADER_SIZE])

    if magic != CRAPVPN_MAGIC:
        return None

    ciphertext = data[CRAPVPN_HEADER_SIZE:]
    if len(ciphertext) != length:
        return None

    plaintext = xor(ciphertext, key)
    return plaintext

Putting It All Together Link to heading

At this point all central components for the VPN service are done. Here’s the full source code, also available as download here:

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

#!/usr/bin/env python3

from fcntl import ioctl
import select
import socket
import struct
import subprocess
import typing

import click


def open_tun_device(device_name: str = "tun0") -> typing.BinaryIO:
    """Open a TUN device with the given name"""

    name_bytes = device_name.encode()
    assert len(name_bytes) < 16, "The interface name must be less than 16 bytes"

    tuntap = open("/dev/net/tun", "r+b", buffering=0)

    LINUX_IFF_TUN = 0x0001
    LINUX_IFF_NO_PI = 0x1000
    flags = LINUX_IFF_TUN | LINUX_IFF_NO_PI
    ifs = struct.pack("16sH22s", name_bytes, flags, b"")

    LINUX_TUNSETIFF = 0x400454CA
    ioctl(tuntap, LINUX_TUNSETIFF, ifs)

    return tuntap


def configure_tun_device(device_name: str, local_ip: str, peer_ip: str):
    """Configure the IP address and peer on a given network device"""

    subprocess.run(("ip", "link", "set", device_name, "up"), check=True)
    subprocess.run(
        ("ip", "addr", "add", local_ip, "peer", peer_ip, "dev", device_name),
        check=True,
    )


def xor(data: bytes, key: bytes):
    """XOR two byte arrays, repeating the key"""
    retval = bytearray(data)
    for i, _ in enumerate(data):
        retval[i] = data[i] ^ key[i % len(key)]
    return bytes(retval)


CRAPVPN_HEADER = ">4sHxx"
CRAPVPN_MAGIC = b"crap"
CRAPVPN_HEADER_SIZE = struct.calcsize(CRAPVPN_HEADER)


def prepare_data_for_sending(data: bytes, key: bytes) -> bytes:
    """Encrypt and wrap data for sending via the VPN"""
    ciphertext = xor(data, key)
    return struct.pack(CRAPVPN_HEADER, CRAPVPN_MAGIC, len(ciphertext)) + ciphertext


def handle_received_data(data: bytes, key: bytes) -> bytes | None:
    """Unwrap and decrypt data from the VPN"""
    magic, length = struct.unpack(CRAPVPN_HEADER, data[:CRAPVPN_HEADER_SIZE])

    if magic != CRAPVPN_MAGIC:
        return None

    ciphertext = data[CRAPVPN_HEADER_SIZE:]
    if len(ciphertext) != length:
        return None

    plaintext = xor(ciphertext, key)
    return plaintext


def run(
    listen_address: tuple[str, int],
    local_ip: str,
    peer_address: tuple[str, int],
    peer_ip: str,
    key: bytes,
):
    """Run the VPN service"""

    # Open TUN device
    device_name = "tun0"
    tuntap = open_tun_device("tun0")

    # Bring device up, configure IP address and route
    configure_tun_device(device_name, local_ip, peer_ip)

    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as remote:
        remote.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        remote.bind(listen_address)

        while True:
            rd_sockets, _, _ = select.select([tuntap, remote], [], [], 1.0)

            for sock in rd_sockets:
                if sock is tuntap:
                    # Data from TUNTAP needs to be pumped to the peer
                    data = tuntap.read(0xFFFF)
                    data = prepare_data_for_sending(data, key)
                    remote.sendto(data, peer_address)

                elif sock is remote:
                    # Data from the peer needs to be pumped to TUNTAP
                    data = remote.recv(0xFFFF)
                    data = handle_received_data(data, key)
                    if data:
                        tuntap.write(data)


@click.command()
@click.option("-k", "--hex-key", required=True, help="Encryption key (hex encoded)")
@click.option("-p", "--peer-host", required=True)
@click.argument("local-ip")
@click.argument("peer-ip")
def main(
    hex_key: str,
    peer_host: str,
    local_ip: str,
    peer_ip: str,
):
    """CrapVPN - a demo VPN implementation by Jonas Lieb (github.com/jojonas)

    See jonaslieb.de/blog/python-vpn/ for details.
    """

    run(
        listen_address=("", 1337),
        peer_address=(peer_host, 1337),
        local_ip=local_ip,
        peer_ip=peer_ip,
        key=bytes.fromhex(hex_key),
    )


if __name__ == "__main__":
    main()

Running the VPN Client Link to heading

The full version has a command-line interface that accepts the key, the remote peer’s IP address and the addresses to be configured inside the tunnel.

Start the CrapVPN client on host A, specifying host B’s IP address with the -p (“peer”) parameter:

$ sudo ./crapvpn.py -k 1234ABCD -p <Host_B_IP> 192.168.2.1 192.168.2.2

Then start the script on host B, this time specifying host A’s IP address. Also note that the IP addresses in the tunnel (192.168.2.2 and 192.168.2.1) are reversed in this second invocation:

$ sudo ./crapvpn.py -k 1234ABCD -p <Host_A_IP> 192.168.2.2 192.168.2.1

Now you should be able to send data from the host 192.168.2.1 to host 192.168.2.2 and vice versa, for example an ICMP echo request:

$ ping -c 3 192.168.2.2
PING 192.168.2.2 (192.168.2.2) 56(84) bytes of data.
64 bytes from 192.168.2.2: icmp_seq=1 ttl=64 time=0.625 ms
64 bytes from 192.168.2.2: icmp_seq=2 ttl=64 time=0.697 ms
64 bytes from 192.168.2.2: icmp_seq=3 ttl=64 time=0.623 ms

--- 192.168.2.2 ping statistics ---
3 packets transmitted, 3 received, 0% packet loss, time 2006ms
rtt min/avg/max/mdev = 0.623/0.648/0.697/0.034 ms

If you run an HTTP server on one of the hosts, you can send queries from the other endpoint:

$ curl -i http://192.168.2.2:8000/      
HTTP/1.0 200 OK
Server: SimpleHTTP/0.6 Python/3.11.9
Date: Sun, 25 Aug 2024 17:54:13 GMT
Content-type: text/html; charset=utf-8
Content-Length: 652

[...]

Conclusion Link to heading

It is very easy to implement a VPN in Python using TUN/TAP devices. The centerpiece is a main-loop that simultaneously handles packets going in both directions. The code I presented is supposed to act as a template for learning and for tinkering with unknown encapsulation protocols. Here are some directions in which the code can be adapted:

• switch to a layer-2 VPN (if you need it)
• implement tunneling over TCP (not recommended, but still necessary sometimes)
• improve encryption and add integrity protection (ideally you would use an AEAD cipher)

I will use this implementation in the next blog post, where I will demonstrate how to write a Wireshark plugin (“dissector”) for CrapVPN.