43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import os
|
|
import struct
|
|
import subprocess
|
|
import threading
|
|
|
|
TUNSETIFF = 0x400454CA
|
|
IFF_TAP = 0x0002
|
|
IFF_NO_PI = 0x1000
|
|
|
|
|
|
class TapDevice:
|
|
def __init__(self, name: str, mtu: int = 9000) -> None:
|
|
self.name = name
|
|
self.mtu = mtu
|
|
self.fd: int | None = None
|
|
self._lock = threading.Lock()
|
|
|
|
def open(self) -> None:
|
|
if self.fd is not None:
|
|
return
|
|
fd = os.open("/dev/net/tun", os.O_RDWR)
|
|
ifreq = struct.pack("16sH22x", self.name.encode("ascii"), IFF_TAP | IFF_NO_PI)
|
|
fcntl.ioctl(fd, TUNSETIFF, ifreq)
|
|
subprocess.run(["ip", "link", "set", "dev", self.name, "mtu", str(self.mtu)], check=True)
|
|
subprocess.run(["ip", "link", "set", "dev", self.name, "up"], check=True)
|
|
self.fd = fd
|
|
|
|
def write(self, frame: bytes) -> int:
|
|
if self.fd is None:
|
|
raise RuntimeError("TAP is not open")
|
|
with self._lock:
|
|
return os.write(self.fd, frame)
|
|
|
|
def close(self) -> None:
|
|
if self.fd is not None:
|
|
try:
|
|
os.close(self.fd)
|
|
finally:
|
|
self.fd = None
|