constants.py

KVM_GET_API_VERSION = 0xAE00
KVM_CREATE_VM = 0xAE01
KVM_SET_USER_MEMORY_REGION = 0x4020AE46
KVM_CREATE_VCPU = 0xAE41
KVM_GET_SREGS = 0x8138AE83
KVM_SET_SREGS = 0x4138AE84
KVM_NR_INTERRUPTS = 0x100
KVM_GET_VCPU_MMAP_SIZE = 0xAE04
KVM_GET_REGS = 0x8090AE81
KVM_SET_REGS = 0x4090AE82
KVM_GET_VCPU_MMAP_SIZE = 0xAE04
KVM_RUN = 0xAE80
KVM_EXIT_IO = 0x2
KVM_EXIT_SHUTDOWN = 0x8
KVM_SET_TSS_ADDR = 0xAE47
KVM_CREATE_IRQCHIP = 0xAE60
KVM_SET_IDENTITY_MAP_ADDR = 0x4008AE48
KVM_CREATE_PIT2 = 0x4040AE77
KVM_EXIT_IO_IN = 0x0
KVM_EXIT_IO_OUT = 0x1
KVM_GET_SUPPORTED_CPUID = 0xC008AE05
KVM_CPUID_SIGNATURE = 0x40000000
KVM_CPUID_FEATURES = 0x40000001
KVM_SET_CPUID2 = 0x4008AE90
CAN_USE_HEAP = 0x80
KEEP_SEGMENTS = 0x40
BOOT_SIZE = 0x1000

kvm-host.py

#!/usr/bin/python3

import os
import sys
import argparse
from fcntl import ioctl
from mmap import mmap, PROT_READ, PROT_WRITE, MAP_SHARED, MAP_PRIVATE, MAP_ANONYMOUS
from structs import *

RAM_SIZE = 1 << 28


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('image_path', help='Path to linux bzImage')
    return parser.parse_args()


class GuestError(Exception):
    pass


class Guest:
    kvm_fd = None
    vm_fd = None
    vcpu_fd = None
    mem = None

    def __init__(self):
        pass

    def _init_regs(self):
        regs = Regs()
        sregs = Sregs()
        try:
            ioctl(self.vcpu_fd, KVM_GET_SREGS, sregs)
        except OSError as err:
            raise GuestError('Failed to get special registers') from err

        # Initialize selector and base with zeros
        sregs.cs.base = 0
        sregs.cs.limit = ~0
        sregs.cs.g = 1

        sregs.ds.base = 0
        sregs.ds.limit = ~0
        sregs.ds.g = 1

        sregs.fs.base = 0
        sregs.fs.limit = ~0
        sregs.fs.g = 1

        sregs.gs.base = 0
        sregs.gs.limit = ~0
        sregs.gs.g = 1

        sregs.es.base = 0
        sregs.es.limit = ~0
        sregs.es.g = 1

        sregs.ss.base = 0
        sregs.ss.limit = ~0
        sregs.ss.g = 1

        sregs.cs.db = 1
        sregs.ss.db = 1
        sregs.cr0 |= 1  # enable protected mode

        # Save special registers
        try:
            ioctl(self.vcpu_fd, KVM_SET_SREGS, sregs)
        except OSError as err:
            raise GuestError('Failed to set special registers') from err

        try:
            ioctl(self.vcpu_fd, KVM_GET_REGS, regs)
        except OSError as err:
            raise GuestError('failed to get registers') from err

        regs.rflags = 2
        regs.rip = 0x100000  # This is where our kernel code starts
        regs.rsi = 0x10000  # This is where our boot parameters start

        try:
            ioctl(self.vcpu_fd, KVM_SET_REGS, regs)
        except OSError as err:
            raise GuestError('failed to set registers') from err

    def _init_cpu_id(self):
        # Set fake cpu id
        kvm_cpuid = Cpuid()
        kvm_cpuid.nent = 100
        ioctl(self.kvm_fd, KVM_GET_SUPPORTED_CPUID, kvm_cpuid)
        
        for entry in kvm_cpuid.entries:
            if entry.function == KVM_CPUID_SIGNATURE:
                entry.eax = KVM_CPUID_FEATURES
                entry.ebx = 0x4b4d564b  # KVMK
                entry.ecx = 0x564b4d56  # VMKV
                entry.edx = 0x4d        # M
        ioctl(self.vcpu_fd, KVM_SET_CPUID2, kvm_cpuid)

    def init(self):
        self.kvm_fd = open('/dev/kvm', 'wb+')
        try:
            self.vm_fd = ioctl(self.kvm_fd, KVM_CREATE_VM, 0)
        except OSError as err:
            raise GuestError('Failed to create vm') from err

        try:
            ioctl(self.vm_fd, KVM_SET_TSS_ADDR, -12288)  # 0xffffc000
        except OSError as err:
            raise GuestError('Failed to set tss addr') from err

        map_addr = ctypes.c_int32(-12288)  # 0xffffc000
        try:
            ioctl(self.vm_fd, KVM_SET_IDENTITY_MAP_ADDR, ctypes.pointer(map_addr))
        except OSError as err:
            raise GuestError('Failed to set identity map addr') from err

        try:
            ioctl(self.vm_fd, KVM_CREATE_IRQCHIP, 0)
        except OSError as err:
            raise GuestError('Failed to create irq chip') from err

        pit = PitConfig(flags=0)
        try:
            ioctl(self.vm_fd, KVM_CREATE_PIT2, pit)
        except OSError as err:
            raise GuestError('Failed to create i8254 interval timer') from err

        self.mem = mmap(-1, RAM_SIZE, MAP_PRIVATE | MAP_ANONYMOUS, PROT_READ | PROT_WRITE)
        pmem = ctypes.c_uint.from_buffer(self.mem)
        mem_region = UserspaceMemoryRegion(slot=0, flags=0,
                                           guest_phys_addr=0, memory_size=RAM_SIZE,
                                           userspace_addr=ctypes.addressof(pmem))
        try:
            ioctl(self.vm_fd, KVM_SET_USER_MEMORY_REGION, mem_region)
        except OSError as err:
            raise GuestError('Failed to set user memory region') from err

        try:
            self.vcpu_fd = ioctl(self.vm_fd, KVM_CREATE_VCPU, 0)
        except OSError as err:
            raise GuestError('Failed to create vcpu') from err

        self._init_regs()
        self._init_cpu_id()

    def load(self, image_path):
        with open(image_path, 'rb') as boot:
            boot_bin = boot.read()

        boot = bytearray(boot_bin[:BOOT_SIZE])
        boot_header = SetupHeader.from_buffer(boot, 0x1F1)
        boot_header.vid_mode = 0xFFFF
        boot_header.type_of_loader = 0xFF
        boot_header.ramdisk_image = 0x0
        boot_header.ramdisk_size = 0x0
        boot_header.loadflags |= CAN_USE_HEAP | 0x01 | KEEP_SEGMENTS
        boot_header.heap_end_ptr = 0xFE00
        boot_header.ext_loader_ver = 0x0
        boot_header.cmd_line_ptr = 0x20000

        self.mem.seek(0x10000)
        self.mem.write(boot)

        self.mem.seek(0x20000)
        self.mem.write(b'console=ttyS0\x00')
        self.mem.write(b'\x00' * (boot_header.cmdline_size - 14))

        setupsz = (boot_header.setup_sects + 1) * 512
        self.mem.seek(0x100000)
        self.mem.write(boot_bin[setupsz:])

    def exit(self):
        self.kvm_fd.close()
        os.close(self.vm_fd)
        os.close(self.vcpu_fd)
        self.mem.close()

    def run(self):
        runsz = ioctl(self.kvm_fd, KVM_GET_VCPU_MMAP_SIZE, 0)
        run_buf = mmap(self.vcpu_fd, runsz, MAP_SHARED, PROT_READ | PROT_WRITE)
        run = Run.from_buffer(run_buf)

        while True:
            ret = ioctl(self.vcpu_fd, KVM_RUN, 0)
            if ret < 0:
                print('KVM_RUN failed')
                return
            if run.exit_reason == KVM_EXIT_IO:
                if run.io.port == 0x3f8 and run.io.direction == KVM_EXIT_IO_OUT:
                    sys.stdout.write(run_buf[run.io.data_offset:run.io.data_offset+run.io.size].decode('utf8'))
                elif run.io.port == 0x3f8 + 5 and run.io.direction == KVM_EXIT_IO_IN:
                    run_buf[run.io.data_offset] = 0x20
            elif run.exit_reason == KVM_EXIT_SHUTDOWN:
                print('Shutdown')
                return
            else:
                print(f'KVM_RUN failed, reason: {run.exit_reason}')
                return


def main():
    args = parse_args()
    vm = Guest()

    try:
        vm.init()
    except GuestError as err:
        print(f'Failed to initialize guest vm: {err.args[0]}, '
              f'errno={err.__cause__.args[0]}, '
              f'error={err.__cause__.args[1]}')
        return 1

    try:
        vm.load(args.image_path)
    except GuestError as err:
        print(f'Failed to initialize guest vm: {err.args[0]}, '
              f'errno={err.__cause__.args[0]}, '
              f'error={err.__cause__.args[1]}')
        return 1

    try:
        vm.run()
    except KeyboardInterrupt:
        print('CTRL-C detected, exiting...')
    vm.exit()

    return 0


if __name__ == '__main__':
    sys.exit(main())

structs.py

import ctypes
from constants import *


class UserspaceMemoryRegion(ctypes.Structure):
    _fields_ = [
        ('slot', ctypes.c_uint32),
        ('flags', ctypes.c_uint32),
        ('guest_phys_addr', ctypes.c_uint64),
        ('memory_size', ctypes.c_uint64),
        ('userspace_addr', ctypes.c_uint64),
    ]


class Segment(ctypes.Structure):
    _fields_ = [
        ('base', ctypes.c_uint64),
        ('limit', ctypes.c_uint32),
        ('selector', ctypes.c_uint16),
        ('type', ctypes.c_uint8),
        ('present', ctypes.c_uint8),
        ('dpl', ctypes.c_uint8),
        ('db', ctypes.c_uint8),
        ('s', ctypes.c_uint8),
        ('l', ctypes.c_uint8),
        ('g', ctypes.c_uint8),
        ('avl', ctypes.c_uint8),
        ('unusable', ctypes.c_uint8),
        ('padding', ctypes.c_uint8),
    ]


class DTable(ctypes.Structure):
    _fields_ = [
        ('base', ctypes.c_uint64),
        ('limit', ctypes.c_uint16),
        ('padding', ctypes.c_uint16 * 3),
    ]


class Sregs(ctypes.Structure):
    _fields_ = [
        ('cs', Segment),
        ('ds', Segment),
        ('es', Segment),
        ('fs', Segment),
        ('gs', Segment),
        ('ss', Segment),
        ('tr', Segment),
        ('ldt', Segment),
        ('gdt', DTable),
        ('idt', DTable),
        ('cr0', ctypes.c_uint64),
        ('cr2', ctypes.c_uint64),
        ('cr3', ctypes.c_uint64),
        ('cr4', ctypes.c_uint64),
        ('cr8', ctypes.c_uint64),
        ('efer', ctypes.c_uint64),
        ('apic_base', ctypes.c_uint64),
        ('interrupt_bitmap', ctypes.c_uint64 * int((KVM_NR_INTERRUPTS + 63) / 64)),
    ]


class Regs(ctypes.Structure):
    _fields_ = [
        ('rax', ctypes.c_uint64),
        ('rbx', ctypes.c_uint64),
        ('rcx', ctypes.c_uint64),
        ('rdx', ctypes.c_uint64),
        ('rsi', ctypes.c_uint64),
        ('rdi', ctypes.c_uint64),
        ('rsp', ctypes.c_uint64),
        ('rbp', ctypes.c_uint64),
        ('r8', ctypes.c_uint64),
        ('r9', ctypes.c_uint64),
        ('r10', ctypes.c_uint64),
        ('r11', ctypes.c_uint64),
        ('r12', ctypes.c_uint64),
        ('r13', ctypes.c_uint64),
        ('r14', ctypes.c_uint64),
        ('r15', ctypes.c_uint64),
        ('rip', ctypes.c_uint64),
        ('rflags', ctypes.c_uint64),
    ]


class IO(ctypes.Structure):
    _fields_ = [
        ('direction', ctypes.c_uint8),
        ('size', ctypes.c_uint8),
        ('port', ctypes.c_uint16),
        ('count', ctypes.c_uint32),
        ('data_offset', ctypes.c_uint64),
    ]


class Run(ctypes.Structure):
    _fields_ = [
        ('request_interrupt_window', ctypes.c_uint8),
        ('padding1', ctypes.c_uint8 * 7),
        ('exit_reason', ctypes.c_uint32),
        ('ready_for_interrupt_injection', ctypes.c_uint8),
        ('if_flag', ctypes.c_uint8),
        ('flags', ctypes.c_uint16),
        ('cr8', ctypes.c_uint64),
        ('apic_base', ctypes.c_uint64),
        ('io', IO)
    ]


class PitConfig(ctypes.Structure):
    _fields_ = [
        ('flags', ctypes.c_uint32),
        ('pad', ctypes.c_uint32 * 15),
    ]


class CpuidEntry2(ctypes.Structure):
    _fields_ = [
        ('function', ctypes.c_uint32),
        ('index', ctypes.c_uint32),
        ('flags', ctypes.c_uint32),
        ('eax', ctypes.c_uint32),
        ('ebx', ctypes.c_uint32),
        ('ecx', ctypes.c_uint32),
        ('edx', ctypes.c_uint32),
        ('padding', ctypes.c_uint32 * 3),
    ]


class Cpuid(ctypes.Structure):
    _fields_ = [
        ('nent', ctypes.c_uint32),
        ('padding', ctypes.c_uint32),
        ('entries', CpuidEntry2 * 100),
    ]


class SetupHeader(ctypes.Structure):
    _pack_ = 1
    _fields_ =[
        ('setup_sects',           ctypes.c_uint8),
        ('root_flags',            ctypes.c_uint16),
        ('syssize',               ctypes.c_uint32),
        ('ram_size',              ctypes.c_uint16),
        ('vid_mode',              ctypes.c_uint16),
        ('root_dev',              ctypes.c_uint16),
        ('boot_flag',             ctypes.c_uint16),
        ('jump',                  ctypes.c_uint16),
        ('header',                ctypes.c_uint32),
        ('version',               ctypes.c_uint16),
        ('realmode_swtch',        ctypes.c_uint32),
        ('start_sys_seg',         ctypes.c_uint16),
        ('kernel_version',        ctypes.c_uint16),
        ('type_of_loader',        ctypes.c_uint8),
        ('loadflags',             ctypes.c_uint8),
        ('setup_move_size',       ctypes.c_uint16),
        ('code32_start',          ctypes.c_uint32),
        ('ramdisk_image',         ctypes.c_uint32),
        ('ramdisk_size',          ctypes.c_uint32),
        ('bootsect_kludge',       ctypes.c_uint32),
        ('heap_end_ptr',          ctypes.c_uint16),
        ('ext_loader_ver',        ctypes.c_uint8),
        ('ext_loader_type',       ctypes.c_uint8),
        ('cmd_line_ptr',          ctypes.c_uint32),
        ('initrd_addr_max',       ctypes.c_uint32),
        ('kernel_alignment',      ctypes.c_uint32),
        ('relocatable_kernel',    ctypes.c_uint8),
        ('min_alignment',         ctypes.c_uint8),
        ('xloadflags',            ctypes.c_uint16),
        ('cmdline_size',          ctypes.c_uint32),
        ('hardware_subarch',      ctypes.c_uint32),
        ('hardware_subarch_data', ctypes.c_uint64),
        ('payload_offset',        ctypes.c_uint32),
        ('payload_length',        ctypes.c_uint32),
        ('setup_data',            ctypes.c_uint64),
        ('pref_address',          ctypes.c_uint64),
        ('init_size',             ctypes.c_uint32),
        ('handover_offset',       ctypes.c_uint32),
        ('kernel_info_offset',    ctypes.c_uint32),
    ]