A crude IOCTL fuzzer for windows driver testing

crude_ioctl_fuzzer.py

import random
import sys
import io
from ctypes import windll, POINTER, byref
from ctypes.wintypes import LPVOID, DWORD, LPCSTR, LPSTR, BOOL, HANDLE

DeviceIoControl = windll.kernel32.DeviceIoControl
CreateFileA = windll.kernel32.CreateFileA
CloseHandle = windll.kernel32.CloseHandle

DeviceIoControl.restype = BOOL
DeviceIoControl.argtypes = [HANDLE, DWORD, LPVOID, DWORD, LPVOID, DWORD,
                            POINTER(DWORD), LPVOID]
CreateFileA.restype = HANDLE
CreateFileA.argtypes = [LPCSTR, DWORD, DWORD, LPVOID, DWORD, DWORD, HANDLE]
CloseHandle.restype = BOOL
CloseHandle.argtypes = [HANDLE]

GENERIC_READ = (1 << 30)
GENERIC_WRITE = (1 << 31)
FILE_SHARE_READ = 1
FILE_SHARE_WRITE = 2
OPEN_EXISTING = 3
FILE_ATTRIBUTE_NORMAL = 0x80


def opendevice(dev):
    # open the device
    hdev = CreateFileA(dev, GENERIC_READ | GENERIC_WRITE,
                       FILE_SHARE_READ | FILE_SHARE_WRITE, None, OPEN_EXISTING,
                       FILE_ATTRIBUTE_NORMAL, None)
    return hdev

def send_ioctl(hdev, ioctl, input, outbuf_len):
    # send IOCTLs to the driver
    outbuf = LPSTR(b"\x00" * outbuf_len) if outbuf_len else None

    outret = DWORD()
    DeviceIoControl(hdev, ioctl, input, len(input), outbuf, outbuf_len,
                    byref(outret), None)
    return outret.value, (outbuf.value if outbuf else b'')

def runner(args, event=None):
    import os
    pid = os.getpid()
    dev = opendevice(args.device)
    length = args.bufferlen
    sys.stdout = io.StringIO()  # suppress output

    while 1:
        num = random.choice(args.ioctls)
        inbuf = "A" * length
        outlen = length
        try:
            send_ioctl(dev, num, inbuf, outlen)
        except: pass

if __name__ == '__main__':
    from multiprocessing import Event, cpu_count, Process
    import argparse
    import time

    def _ioctls(x):
        try:
            return [int(y, 0) for y in x.split(",")]
        except ValueError:
            pass
        return open(x, "r").readlines()

    ap = argparse.ArgumentParser()
    ap.add_argument("-i", "--ioctls", type=_ioctls,
                    help=("List of IOCTL codes to fuzz. A comma separated "
                          "list of IOCTLs"))
    ap.add_argument("-d", "--device", required=True, type=lambda x: x.encode(),
                    help="The device to fuzz")
    ap.add_argument("-l", "--bufferlen", type=int, default=(0, 0),
                    help=("The length range of input and output buffers "
                          "passed on each call to DeviceIoControl"))

    args = ap.parse_args()

    processes = cpu_count() # try to run one processes on each available CPU
    stopevent = Event()
    procs = []
    for _ in range(processes):
        p = Process(target=runner, args=(args, stopevent))
        p.daemon = True
        procs.append(p)
        p.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        stopevent.set()