Remote DLL Injection

remote_dll.py

#!/usr/bin/env python3
# encoding: utf-8
"""
rdll.py (Python3 port)
Original by Charles (2012) - WTFPL v2
Recode By Pari Malam (2025)
"""

import os
import sys
from cutil import *
from _ctypes import FUNCFLAG_CDECL as _FUNCFLAG_CDECL, \
    FUNCFLAG_PYTHONAPI as _FUNCFLAG_PYTHONAPI, \
    FUNCFLAG_USE_ERRNO as _FUNCFLAG_USE_ERRNO, \
    FUNCFLAG_USE_LASTERROR as _FUNCFLAG_USE_LASTERROR, \
    CFuncPtr as _CFuncPtr
from _kernel32 import PLoadLibraryW as PLoadLibrary
from extern import pefile


__all__ = []


def _pack_args(*args):
    class _Args(Structure):
        pass

    fields = []
    for i, arg in enumerate(args):
        fields.append(('arg%d' % i, type(arg)))
    _Args._fields_ = fields
    return _Args(*args)


class _RCFuncPtr(object):
    _addr_ = 0
    _flags_ = None
    _restype_ = None
    _funcptr_ = None
    _hprocess_ = None

    def _valueof(self, arg):
        if not hasattr(arg, '_type_'):
            return arg
        elif hasattr(arg, 'value'):
            return arg.value
        elif hasattr(arg, 'contents'):
            return arg.contents
        else:
            raise Exception("Don't know how to get the value of arg.\nType: %s" % type(arg))

    def _valtoargtype(self, arg, argtype):
        result = 0
        if isinstance(arg, str):
            if argtype == c_char_p:
                result = create_string_buffer(arg.encode())
            elif argtype == c_wchar_p:
                result = create_unicode_buffer(arg)
            elif getattr(argtype, "_type_", None) == c_ubyte:
                result = (c_ubyte * (len(arg) + 1))()
                for i, c in enumerate(arg.encode()):
                    result[i] = c
            else:
                raise Exception("Don't know how to convert string '%s' into type %s" % (arg, argtype))
        elif hasattr(argtype, "_length_") or not isinstance(argtype._type_, str):  # array or pointer
            result = cast(arg, argtype)
        elif hasattr(argtype, "value"):
            result = argtype(arg)
        else:
            raise Exception("Don't know how to convert arg to argtype.\nArg: %s\nArgtype: %s" % (arg, argtype))
        return result

    def _alloc_set_var(self, val):
        buflen = sizeof(val)
        buffer = VirtualAllocEx(self._hprocess_, 0, buflen, MEM_COMMIT, PAGE_READWRITE)
        if not buffer:
            raise Exception("Could not allocate our remote buffer.")

        if not WriteProcessMemory(self._hprocess_, buffer, val, buflen, byref(c_ulong(0))):
            raise Exception("Could not write to our remote variable.")

        return buffer

    def __call__(self, *more):
        funcptr = self._funcptr_
        result = DWORD(0) if funcptr.restype is None else funcptr.restype()
        lpParameter = None

        if funcptr.argtypes and len(funcptr.argtypes) > 0:
            args = []
            argcount = len(more)
            for i, argtype in enumerate(funcptr.argtypes):
                if i >= argcount:
                    arg = argtype()
                elif hasattr(more[i], "_type_") and more[i]._type_ == argtype:
                    arg = more[i]
                else:
                    arg = self._valtoargtype(self._valueof(more[i]), argtype)
                args.append(arg)
            lpParameter = _pack_args(*args) if argcount > 1 else args[0]

        if hasattr(lpParameter, "_b_needsfree_") and getattr(lpParameter, "_b_needsfree_", 0) == 1 and bool(lpParameter):
            lpParameter = self._alloc_set_var(lpParameter)

        hRemoteThread = CreateRemoteThread(
            self._hprocess_, NULL_SECURITY_ATTRIBUTES, 0,
            cast(self._addr_, LPTHREAD_START_ROUTINE),
            lpParameter, 0, byref(c_ulong(0))
        )
        if not hRemoteThread:
            if hasattr(lpParameter, "_b_needsfree_") and getattr(lpParameter, "_b_needsfree_", 0) == 1 and bool(lpParameter):
                VirtualFreeEx(self._hprocess_, lpParameter, 0, MEM_RELEASE)
            CloseHandle(self._hprocess_)
            raise WinError("Failed to start our remote thread.")

        WaitForSingleObject(hRemoteThread, INFINITE)
        GetExitCodeThread(hRemoteThread, cast(byref(result), LPDWORD))
        CloseHandle(hRemoteThread)

        if hasattr(lpParameter, "_b_needsfree_") and getattr(lpParameter, "_b_needsfree_", 0) == 1 and bool(lpParameter):
            VirtualFreeEx(self._hprocess_, lpParameter, 0, MEM_RELEASE)
        return result

    def __init__(self, offset, funcid, rdll):
        self._addr_ = offset
        if self._flags_ == _FUNCFLAG_CDECL:
            self._funcptr_ = CFUNCTYPE(self._restype_)
        elif self._flags_ == _FUNCFLAG_STDCALL:
            self._funcptr_ = WINFUNCTYPE(self._restype_)
        elif self._flags_ == _FUNCFLAG_PYTHONAPI:
            self._funcptr_ = PYFUNCTYPE(self._restype_)
        self._funcptr_._func_flags_ = self._flags_

    def __bool__(self):
        return bool(self._funcptr_)

    def __repr__(self):
        return repr(self._funcptr_)


class RDLL(object):
    _func_flags_ = _FUNCFLAG_CDECL
    _func_restype_ = c_int
    _hprocess_ = 0
    _hthread_ = 0
    _exports_ = {}

    def __init__(self, name=None, pid=0, thid=0, mode=DEFAULT_MODE, handle=None, use_errno=False, use_last_error=False):
        if name is None and handle is None:
            raise WindowsError("We need either a name or a handle to a preloaded DLL.")
        elif name is None:
            self._name = GetModuleFileName(handle)
        else:
            self._name = name

        flags = self._func_flags_
        if use_errno:
            flags |= _FUNCFLAG_USE_ERRNO
        if use_last_error:
            flags |= _FUNCFLAG_USE_LASTERROR

        self._hthread_ = thid
        if pid == 0:
            check = find_parent_process()
            if check is None:
                raise WinError("Failed to open parent process and no pid specified.")
            pi, ti = check
        else:
            pi, ti = bypid(pid)

        if self._hthread_ == 0:
            self._hthread_ = ti
        self._hprocess_ = OpenProcess(PROCESS_MOST, FALSE, pi)

        class _FuncPtr(_RCFuncPtr):
            _flags_ = flags
            _restype_ = self._func_restype_
            _hprocess_ = self._hprocess_

        self._FuncPtr = _FuncPtr
        self._handle = self.__inject__()
        if self._handle == 0:
            raise WindowsError("Could not inject library: %s" % self._name)

        self.__populate_exports__()

    def __inject__(self):
        val = create_unicode_buffer(self._name, len(self._name) + 1)
        buflen = sizeof(val)

        buffer = VirtualAllocEx(self._hprocess_, 0, buflen, MEM_COMMIT, PAGE_READWRITE)
        if not buffer:
            raise Exception("Could not allocate remote buffer.")

        if not WriteProcessMemory(self._hprocess_, buffer, cast(val, LPCVOID), buflen, byref(c_ulong(0))):
            raise Exception("Could not write to remote variable.")

        hRemoteThread = CreateRemoteThread(
            self._hprocess_, NULL_SECURITY_ATTRIBUTES, 0,
            PLoadLibrary, buffer, 0, byref(c_ulong(0))
        )
        if not hRemoteThread:
            VirtualFreeEx(self._hprocess_, buffer, 0, MEM_RELEASE)
            CloseHandle(self._hprocess_)
            raise WinError("Failed to start remote thread.")

        WaitForSingleObject(hRemoteThread, INFINITE)
        result = c_ulong(0)
        GetExitCodeThread(hRemoteThread, byref(result))
        CloseHandle(hRemoteThread)
        VirtualFreeEx(self._hprocess_, buffer, 0, MEM_RELEASE)
        return result.value

    def __populate_exports__(self):
        if len(os.path.splitext(self._name)[1].lower()) == 0:
            self._name += ".dll"
        pe = pefile.PE(self._name, fast_load=True)
        direxport = pe.OPTIONAL_HEADER.DATA_DIRECTORY[0]
        exportsobj = pe.parse_export_directory(direxport.VirtualAddress, direxport.Size)
        pe.close()
        for export in exportsobj.symbols:
            self._exports_[export.name] = self._handle + export.address
            self._exports_[export.ordinal] = self._handle + export.address

    def __repr__(self):
        return "<%s '%s', handle %x at %x>" % (
            self.__class__.__name__,
            self._name,
            self._handle & (sys.maxsize * 2 + 1),
            id(self) & (sys.maxsize * 2 + 1),
        )

    def __getattr__(self, name):
        if name.startswith("__") and name.endswith("__"):
            raise AttributeError(name)
        func = self.__getitem__(name)
        setattr(self, name, func)
        return func

    def __getitem__(self, name_or_ordinal):
        ordinal = isinstance(name_or_ordinal, int)
        if name_or_ordinal not in self._exports_:
            if ordinal:
                raise WindowsError("Could not find address of function at ordinal: %d" % name_or_ordinal)
            else:
                raise WindowsError("Could not find address of function named: %s" % name_or_ordinal)
        func = self._FuncPtr(self._exports_[name_or_ordinal], name_or_ordinal, self)
        if not ordinal:
            func.__name__ = name_or_ordinal
        return func


if __name__ == "__main__":
    testdll = RDLL("testdll.dll")
    Initialize = testdll.Initialize
    Initialize.restype = None
    Initialize.argtypes = []
    Initialize()
    testdll.Finalize()