[python-win32] Opening existing memory mapped files with pywin32

[Eryk Sun]

On 2/21/21, rhyslloyd1 <rhyslloyd1 at protonmail.com> wrote:
>
> In my head my ideal outcome is being able to map the whole memory mapped
> file and not have to specify a size for it incase it changes size however I
> don't understand the concept of a memory mapped file so I maybe my very
> limited understanding is just incorrect, thanks in advance for any response!

The address space of a process is managed in pages of a given fixed
size (e.g. 4 KiB). The state of each page is tracked in an entry in
the process page table, i.e. a page-table entry (PTE). When a page is
sharable between processes, its PTE in a particular process references
a prototype PTE that manages the overall state of the page (e.g.
whether the page is resident in RAM or paged out to a filesystem
paging file or data file).

The system uses a kernel object called a Section to reference the
prototype PTEs in a block of shared memory. In the Windows API, a
Section is called a "file mapping" because the pages mapped into
memory are always a byte range in a filesystem paging file or data
file. At the time of creation via CreateFileMappingW, a file mapping
can be assigned a name, either a global name for all sessions, or a
local name in the current desktop session. This allows a process to
easily try to open a file mapping (e.g. via OpenFileMapping) and use
it to map all or a subset of the shared pages into its address space
(e.g. via MapViewOfFile).

In your case, I gather that you need to open a file mapping named
"{8BA1E16C-FC54-4595-9782-E370A5FBE8DA}" and map the whole block of
shared memory into the current process, without knowing how large it
is. You'll need to query the size of the mapped region (i.e. the
RegionSize) via VirtualQuery. Preferably also wrap the memory in an
easy interface such as a memoryview, which should automatically unmap
the memory when finalized.

Here's a map_section(name, mode) function that implements this. Note
that for reliable operation it requires memoryview.toreadonly(), which
was added in Python 3.8. The underlying ctypes array allows writing to
readonly memory, which will crash the process with an access
violation. That's avoided in 3.8+ by returning a readonly memoryview
when the shared memory is mapped without write access.

import warnings
import ctypes
from ctypes import wintypes

kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)

FILE_MAP_READ = SECTION_MAP_READ = 0x0004
FILE_MAP_WRITE = SECTION_MAP_WRITE = 0x0002

class MEMORY_BASIC_INFORMATION(ctypes.Structure):
    _fields_ = (('BaseAddress', wintypes.LPVOID),
               ('AllocationBase', wintypes.LPVOID),
               ('AllocationProtect', wintypes.DWORD),
               ('PartitionId', wintypes.WORD),
               ('RegionSize', ctypes.c_size_t),
               ('State', wintypes.DWORD),
               ('Protect', wintypes.DWORD),
               ('Type', wintypes.DWORD))

PMEMORY_BASIC_INFORMATION = ctypes.POINTER(MEMORY_BASIC_INFORMATION)

kernel32.OpenFileMappingW.restype = wintypes.HANDLE
kernel32.OpenFileMappingW.argtypes = (wintypes.DWORD, wintypes.BOOL,
    wintypes.LPWSTR)

kernel32.MapViewOfFile.restype = wintypes.LPVOID
kernel32.MapViewOfFile.argtypes = (wintypes.HANDLE, wintypes.DWORD,
    wintypes.DWORD, wintypes.DWORD, ctypes.c_size_t)

kernel32.UnmapViewOfFile.restype = wintypes.BOOL
kernel32.UnmapViewOfFile.argtypes = (wintypes.LPCVOID,)

kernel32.VirtualQuery.restype = ctypes.c_size_t
kernel32.VirtualQuery.argtypes = (wintypes.LPCVOID,
    PMEMORY_BASIC_INFORMATION, ctypes.c_size_t)

class BaseSharedMem(ctypes.Array):
    _type_ = ctypes.c_char
    _length_ = 0

    def __del__(self, *, UnmapViewOfFile=kernel32.UnmapViewOfFile,
                warn=warnings.warn):
        if not UnmapViewOfFile(self):
            warn("UnmapViewOfFile failed", ResourceWarning, source=self)


def map_section(name, mode='r'):
    mbi = MEMORY_BASIC_INFORMATION()

    if mode == 'r':
        access = FILE_MAP_READ
    elif mode == 'w':
        access = FILE_MAP_WRITE
    elif mode in ('rw', 'wr'):
        access = FILE_MAP_READ | FILE_MAP_WRITE
    else:
        raise ValueError('invalid mode')

    h = kernel32.OpenFileMappingW(access, False, name)
    if not h:
        raise ctypes.WinError(ctypes.get_last_error())

    try:
        address = kernel32.MapViewOfFile(h, access, 0, 0, 0)
        if not address:
            raise ctypes.WinError(ctypes.get_last_error())
    finally:
        kernel32.CloseHandle(h)

    result = kernel32.VirtualQuery(address, ctypes.byref(mbi),
                ctypes.sizeof(mbi))
    if not result:
        ex = ctypes.WinError(ctypes.get_last_error())
        if not kernel32.UnmapViewOfFile(address):
            ex2 = ctypes.WinError(ctypes.get_last_error())
            raise ex2 from ex
        raise ex

    array_t = type('SharedMem_{}'.format(mbi.RegionSize),
                (BaseSharedMem,), {'_length_': mbi.RegionSize})
    mv = memoryview(array_t.from_address(address)).cast('B')

    if 'w' not in mode:
        return getattr(mv, 'toreadonly', lambda: mv)()
    return mv