80 lines
2.9 KiB
Python
Executable File
80 lines
2.9 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Inspect a Windows kernel driver (.sys): confirm native subsystem and flag kernel primitives.
|
|
|
|
Parses the PE optional header for the native subsystem and scans strings for kernel imports,
|
|
IRP/IOCTL dispatch, arbitrary read/write primitives, and callback registration. Reads bytes
|
|
statically; never loads the driver.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import struct
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ASCII_RE = re.compile(rb"[\x20-\x7e]{4,}")
|
|
|
|
KERNEL_IMPORTS = ["ntoskrnl.exe", "hal.dll", "ndis.sys", "fltmgr.sys", "wdfldr.sys",
|
|
"storport.sys"]
|
|
DISPATCH = ["iocreatedevice", "iocreatesymboliclink", "iodeletedevice", "irp_mj_device_control",
|
|
"iocreatedevicesecure", "wdfdevicecreate"]
|
|
RW_PRIMITIVES = ["mmmapiospace", "zwmapviewofsection", "mmcopymemory", "mmmapiospaceex",
|
|
"zwopensection", "halgetbusdatabyoffset", "mmgetphysicaladdress"]
|
|
CALLBACKS = ["pssetcreateprocessnotifyroutine", "pssetloadimagenotifyroutine",
|
|
"obregistercallbacks", "cmregistercallback", "pssetcreatethreadnotifyroutine"]
|
|
|
|
|
|
def _subsystem(data: bytes) -> int | None:
|
|
if data[:2] != b"MZ":
|
|
return None
|
|
e_lfanew = struct.unpack_from("<I", data, 0x3c)[0]
|
|
if data[e_lfanew:e_lfanew + 4] != b"PE\x00\x00":
|
|
return None
|
|
return struct.unpack_from("<H", data, e_lfanew + 24 + 68)[0]
|
|
|
|
|
|
def _strings(data: bytes) -> str:
|
|
return "\n".join(m.decode("latin-1", "ignore") for m in ASCII_RE.findall(data)).lower()
|
|
|
|
|
|
def inspect(path: str) -> dict:
|
|
data = Path(path).read_bytes()
|
|
subsystem = _subsystem(data)
|
|
blob = _strings(data)
|
|
is_driver = subsystem == 1 or any(k in blob for k in KERNEL_IMPORTS)
|
|
capabilities = []
|
|
rw = sorted({p for p in RW_PRIMITIVES if p in blob})
|
|
cbs = sorted({c for c in CALLBACKS if c in blob})
|
|
if rw:
|
|
capabilities.append("arbitrary_memory_access")
|
|
if cbs:
|
|
capabilities.append("kernel_callback_registration")
|
|
if any(d in blob for d in DISPATCH):
|
|
capabilities.append("ioctl_dispatch")
|
|
return {
|
|
"is_kernel_driver": is_driver,
|
|
"pe_subsystem": subsystem,
|
|
"kernel_imports": sorted({k for k in KERNEL_IMPORTS if k in blob}),
|
|
"dispatch_indicators": sorted({d for d in DISPATCH if d in blob}),
|
|
"read_write_primitives": rw,
|
|
"callback_registration": cbs,
|
|
"capabilities": capabilities,
|
|
"byovd_risk": bool(rw),
|
|
}
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Inspect a Windows kernel driver")
|
|
sub = parser.add_subparsers(dest="cmd", required=True)
|
|
p = sub.add_parser("inspect")
|
|
p.add_argument("file")
|
|
args = parser.parse_args(argv)
|
|
print(json.dumps(inspect(args.file), indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|