the start of ebpf

bpf.py

+from ctypes import CDLL, c_int, get_errno, cast, c_void_p, create_string_buffer
+from struct import pack
+
+from os import strerror
+
+libc = CDLL("libc.so.6", use_errno=True)
+
+def addrof(ptr):
+    return cast(ptr, c_void_p).value
+
+def bpf(cmd, fmt, *args):
+    attr = pack(fmt, *args)
+    ret = libc.syscall(386, c_int(cmd), attr, len(attr))
+    if ret == -1:
+        raise OSError(get_errno(), strerror(get_errno()))
+    return ret
+
+def create_map(map_type, key_size, value_size, max_entries):
+    return bpf(0, "IIII", map_type, key_size, value_size, max_entries)
+
+def lookup_elem(fd, key, size):
+    value = create_string_buffer(size)
+    bpf(1, "IQQQ", fd, addrof(key), addrof(value), 0)
+    return value.value
+
+def update_elem(fd, key, value, flags):
+    return bpf(2, "IQQQ", fd, addrof(key), addrof(value), flags)
+
+def prog_load(prog_type, insns, license,
+              log_level=0, log_size=4096, kern_version=0):
+    if log_level == 0:
+        log_buf = 0
+        log_size = 0
+    else:
+        log_buf = addrof(create_string_buffer(log_size))
+    license = license.encode("utf8")
+    bpf(5, "IIQQIIQI", prog_type, int(len(insns) // 8), addrof(license),
+        log_level, log_size, log_buf, kern_version)
+    if log_level != 0:
+        return log_buf.value
+
+if __name__ == "__main__":
+    fd = create_map(1, 4, 4, 10)
+    update_elem(fd, b"asdf", b"ckde", 0)
+    print(lookup_elem(fd, b"asdf", 4))

ebpf.py

+from collections import namedtuple
+from struct import pack
+
+from .bpf import prog_load
+
+Instruction = namedtuple("Instruction",
+                         ["opcode", "dst", "src", "off", "imm"])
+
+def augassign(opcode):
+    def ret(self, value):
+        if isinstance(value, int):
+            return Instruction(opcode + 3 * self.long, self.no, 0, 0, value)
+        elif isinstance(value, Register) and self.long == value.long:
+            return Instruction(opcode + 8 + 3 * self.long,
+                               self.no, value.no, 0, 0)
+        else:
+            return NotImplemented
+    return ret
+
+
+class Sum:
+    def __init__(self, no, offset):
+        self.no = no
+        self.offset = offset
+
+    def __add__(self, value):
+        if isinstance(value, int):
+            return Sum(self.no, self.offset + value)
+        else:
+            return NotImplemented
+
+    __radd__ = __add__
+
+    def __sub__(self, value):
+        if isinstance(value, int):
+            return Sum(self.no, self.offset - value)
+        else:
+            return NotImplemented
+
+
+class Register:
+    offset = 0
+
+    def __init__(self, no, ebpf, long):
+        self.no = no
+        self.ebpf = ebpf
+        self.long = long
+
+    __iadd__ = augassign(4)
+    __isub__ = augassign(0x14)
+    __imul__ = augassign(0x24)
+    __itruediv__ = augassign(0x34)
+    __ior__ = augassign(0x44)
+    __iand__ = augassign(0x54)
+    __ilshift__ = augassign(0x64)
+    __irshift__ = augassign(0x74)
+    __imod__ = augassign(0x94)
+    __ixor__ = augassign(0xa4)
+
+    def __add__(self, value):
+        if isinstance(value, int) and self.long:
+            return Sum(self.no, value)
+        else:
+            return NotImplemented
+
+    __radd__ = __add__
+
+    def __sub__(self, value):
+        if isinstance(value, int) and self.long:
+            return Sum(self.no, -value)
+        else:
+            return NotImplemented
+
+
+class Memory:
+    def __init__(self, ebpf, bits):
+        self.ebpf = ebpf
+        self.bits = bits
+
+    def __setitem__(self, addr, value):
+        if isinstance(value, int):
+            self.ebpf.append(0x62 + self.bits, addr.no, 0, addr.offset, value)
+        elif isinstance(value, Register):
+            self.ebpf.append(0x63 + self.bits, addr.no, value.no,
+                             addr.offset, 0)
+        else:
+            raise RuntimeError("cannot compile")
+
+    def __getitem__(self, addr):
+        pass
+
+
+class RegisterDesc:
+    def __init__(self, no, long):
+        self.no = no
+        self.long = long
+
+    def __get__(self, instance, owner=None):
+        if instance is None:
+            return self
+        else:
+            return Register(self.no, instance, self.long)
+
+    def __set__(self, instance, value):
+        if isinstance(value, int):
+            instance.append(0xb4 + 3 * self.long, self.no, 0, 0, value)
+        elif isinstance(value, Register) and self.long == value.long:
+            instance.append(0xbc + 3 * self.long, self.no, value.no, 0, 0)
+        elif isinstance(value, Instruction):
+            instance.opcodes.append(value)
+        else:
+            raise RuntimeError("cannot compile")
+        
+
+class EBPF:
+    def __init__(self, prog_type=0, license="", kern_version=0):
+        self.opcodes = []
+        self.prog_type = prog_type
+        self.license = license
+        self.kern_version = kern_version
+
+    def append(self, opcode, dst, src, off, imm):
+        self.opcodes.append(Instruction(opcode, dst, src, off, imm))
+
+    def assemble(self):
+        return b"".join(
+            pack("<BBHI", i.opcode, i.dst << 4 | i.src, i.off, i.imm)
+            for i in self.opcodes)
+
+    def load(self, log_level=0, log_size=4096):
+        return prog_load(self.prog_type, self.assemble(), self.license,
+                         log_level, log_size, self.kern_version)
+
+
+for i in range(10):
+    setattr(EBPF, f"r{i}", RegisterDesc(i, True))
+
+for i in range(10):
+    setattr(EBPF, f"s{i}", RegisterDesc(i, False))

ebpf_test.py

+from unittest import TestCase, main
+
+from .ebpf import EBPF, Instruction
+
+
+class Tests(TestCase):
+    def test_assemble(self):
+        e = EBPF()
+        e.append(0x24, 3, 4, 0x2c3d, 0x2d3e4f5e)
+        self.assertEqual(e.assemble(), b"$4=,^O>-")
+
+    def test_assign(self):
+        e = EBPF()
+        e.r5 = 7
+        e.r6 = e.r3
+        self.assertEqual(e.opcodes, 
+            [Instruction(0xb7, 5, 0, 0, 7),
+             Instruction(0xbf, 6, 3, 0, 0)])
+
+    def test_short(self):
+        e = EBPF()
+        e.s3 = 7
+        e.s4 = e.s1
+        e.s2 += 3
+        e.s5 += e.s6
+        self.assertEqual(e.opcodes, 
+            [Instruction(0xb4, 3, 0, 0, 7),
+             Instruction(0xbc, 4, 1, 0, 0),
+             Instruction(opcode=4, dst=2, src=0, off=0, imm=3),
+             Instruction(opcode=0xc, dst=5, src=6, off=0, imm=0)])
+
+    def test_augassign(self):
+        e = EBPF()
+        e.r5 += 7
+        e.r3 += e.r6
+        e.r4 -= 3
+        e.r4 -= e.r7
+        e.r4 *= 3
+        e.r4 *= e.r7
+        e.r4 /= 3
+        e.r4 /= e.r7
+        e.r4 |= 3
+        e.r4 |= e.r7
+        e.r4 &= 3
+        e.r4 &= e.r7
+        e.r4 <<= 3
+        e.r4 <<= e.r7
+        e.r4 >>= 3
+        e.r4 >>= e.r7
+        e.r4 %= 3
+        e.r4 %= e.r7
+        e.r4 ^= 3
+        e.r4 ^= e.r7
+
+        self.assertEqual(e.opcodes, 
+            [Instruction(opcode=7, dst=5, src=0, off=0, imm=7),
+             Instruction(opcode=15, dst=3, src=6, off=0, imm=0),
+             Instruction(opcode=0x17, dst=4, src=0, off=0, imm=3),
+             Instruction(opcode=0x1f, dst=4, src=7, off=0, imm=0),
+             Instruction(opcode=0x27, dst=4, src=0, off=0, imm=3),
+             Instruction(opcode=0x2f, dst=4, src=7, off=0, imm=0),
+             Instruction(opcode=0x37, dst=4, src=0, off=0, imm=3),
+             Instruction(opcode=0x3f, dst=4, src=7, off=0, imm=0),
+             Instruction(opcode=0x47, dst=4, src=0, off=0, imm=3),
+             Instruction(opcode=0x4f, dst=4, src=7, off=0, imm=0),
+             Instruction(opcode=0x57, dst=4, src=0, off=0, imm=3),
+             Instruction(opcode=0x5f, dst=4, src=7, off=0, imm=0),
+             Instruction(opcode=0x67, dst=4, src=0, off=0, imm=3),
+             Instruction(opcode=0x6f, dst=4, src=7, off=0, imm=0),
+             Instruction(opcode=0x77, dst=4, src=0, off=0, imm=3),
+             Instruction(opcode=0x7f, dst=4, src=7, off=0, imm=0),
+             Instruction(opcode=0x97, dst=4, src=0, off=0, imm=3),
+             Instruction(opcode=0x9f, dst=4, src=7, off=0, imm=0),
+             Instruction(opcode=0xa7, dst=4, src=0, off=0, imm=3),
+             Instruction(opcode=0xaf, dst=4, src=7, off=0, imm=0)])
+
+
+if __name__ == "__main__":
+    main()