Python3: Translate virtual memory address into physical memory address using `/proc/self/pagemap`

virt2phy.py

import ctypes
import os
import sys

buffer = bytearray(1_000_000)

vaddr = ctypes.addressof(ctypes.c_char.from_buffer(buffer))

print(f'Allocated an array of {len(buffer)} bytes at 0x{vaddr:x}.')

# Constants
PAGEMAP_ENTRY = 8  # Size of each entry in pagemap file (8 bytes)

# Helper functions
def is_bigendian():
    return sys.byteorder == 'big'

def GET_BIT(val, bit):
    return (val >> bit) & 1

def GET_PFN(val):
    return val & 0x7FFFFFFFFFFFFF  # Extract bits 0-54 (PFN)

print(f"Big endian? {is_bigendian()}")

def get_pfn(virtual_address):
    print('')
    with open('/proc/self/pagemap', "rb") as f:
        # Calculate file offset
        page_size = os.sysconf("SC_PAGE_SIZE")
        file_offset = (virtual_address // page_size) * PAGEMAP_ENTRY

        print(f"Vaddr: 0x{virtual_address:x}, Page_size: {page_size}, Entry_size: {PAGEMAP_ENTRY}")
        print(f"Reading /proc/self/pagemap at 0x{file_offset:x}")

        # Seek to the position
        f.seek(file_offset)

        # Read bytes
        c_buf = bytearray(PAGEMAP_ENTRY)
        bytes_read = f.read(PAGEMAP_ENTRY)

        if len(bytes_read) < PAGEMAP_ENTRY:
            print("\nReached end of the file")
            sys.exit(1)

        # Process bytes based on endianness
        for i in range(PAGEMAP_ENTRY):
            if is_bigendian():
                c_buf[i] = bytes_read[i]
            else:
                c_buf[PAGEMAP_ENTRY - i - 1] = bytes_read[i]
            print(f"[{i}]0x{bytes_read[i]:x} ", end="")

        # Convert bytes to integer
        read_val = 0
        for i in range(PAGEMAP_ENTRY):
            read_val = (read_val << 8) + c_buf[i]

        print(f"\nResult: 0x{read_val:x}")

        # Check page status
        if GET_BIT(read_val, 63):
            print(f"PFN: 0x{GET_PFN(read_val):x}")
        else:
            print("Page not present")

        if GET_BIT(read_val, 62):
            print("Page swapped")

# Test with multiple allocations
def test_multiple_allocations():
    results = []
    buffers = []
    for i in range(5):
        # Allocate a new buffer each time
        buffer = bytearray(4096)  # One page
        buffers.append(buffer)
        addr = ctypes.addressof(ctypes.c_char.from_buffer(buffer))
        pfn = get_pfn(addr)
        results.append((addr, pfn))
    return results

# Test with allocations after memory pressure
def test_with_memory_pressure():
    print("\nTesting with memory pressure:")
    # Create some memory pressure
    temp_arrays = [bytearray(1024*1024) for _ in range(100)]  # Allocate ~100MB

    # Force some page faults and memory movement
    for arr in temp_arrays:
        arr[0] = 1  # Touch each array to ensure it's in physical memory

    # Now test allocation
    buffer = bytearray(4096)
    addr = ctypes.addressof(ctypes.c_char.from_buffer(buffer))
    pfn = get_pfn(addr)
    # Clean up
    del temp_arrays

# Run tests
print("Testing multiple allocations:")
test_multiple_allocations()
test_with_memory_pressure()