#!/usr/bin/env python3 import argparse import os import struct import sys # Constants for pagemap and kpageflags PAGEMAP_ENTRY_SIZE = 8 KPAGEFLAGS_ENTRY_SIZE = 8 KPF_DIRTY_BIT_SHIFT = 4 PM_PAGE_PRESENT_BIT = 63 PM_PFN_MASK = (1 << 55) - 1 # Global debug flag DEBUG_MODE = False try: PAGE_SIZE = os.sysconf("SC_PAGE_SIZE") except ValueError: print("Assuming page size is 4 kB.") PAGE_SIZE = 4096 def debug_print(msg): if DEBUG_MODE: print(f"DEBUG: {msg}", file=sys.stderr) def count_dirty_pages_in_region( pid: int, base_address_str: str, map_size_bytes: int, count_present: bool ) -> int: global DEBUG_MODE # Allow modification if set by CLI try: base_address = int(base_address_str, 0) except ValueError: print( f"Error: Invalid base address format: '{base_address_str}'. Use hex (e.g., 0x...) or decimal.", file=sys.stderr, ) sys.exit(1) if map_size_bytes < 0: print( f"Error: Map size ({map_size_bytes} bytes) must be non-negative.", file=sys.stderr, ) sys.exit(1) if map_size_bytes == 0: debug_print("Map size is zero, no pages to check.") return 0 if base_address % PAGE_SIZE != 0: print( f"Warning: Base address {hex(base_address)} is not page-aligned (page size: {hex(PAGE_SIZE)}). " "The scan will cover all pages that overlap with the specified region.", file=sys.stderr, ) first_vpage_index_in_scan = base_address // PAGE_SIZE last_byte_address_in_region = base_address + map_size_bytes - 1 last_vpage_index_in_scan = last_byte_address_in_region // PAGE_SIZE num_pages_to_scan = ( last_vpage_index_in_scan - first_vpage_index_in_scan + 1 ) debug_print(f"Page size: {PAGE_SIZE} (0x{PAGE_SIZE:x})") debug_print( f"Region: {hex(base_address)} - {hex(base_address + map_size_bytes -1)} (size: {map_size_bytes} bytes)" ) debug_print( f"Scanning virtual page indices from {hex(first_vpage_index_in_scan)} to {hex(last_vpage_index_in_scan)} ({num_pages_to_scan} pages)." ) if num_pages_to_scan <= 0: return 0 pagemap_file_path = f"/proc/{pid}/pagemap" kpageflags_file_path = "/proc/kpageflags" dirty_page_count = 0 pagemap_fd = -1 kpageflags_fd = -1 try: pagemap_fd = os.open(pagemap_file_path, os.O_RDONLY) kpageflags_fd = os.open(kpageflags_file_path, os.O_RDONLY) except OSError as e: print( f"Error: Could not open proc files. PID: {pid}, Path affected: '{e.filename}'.", file=sys.stderr, ) print( f"Details: {os.strerror(e.errno)}. Ensure you have necessary permissions (e.g., run as root).", file=sys.stderr, ) sys.exit(1) try: for i in range(num_pages_to_scan): current_virtual_page_index = first_vpage_index_in_scan + i vpage_addr = current_virtual_page_index * PAGE_SIZE debug_print( f"Processing vpage index {hex(current_virtual_page_index)} (addr ~{hex(vpage_addr)})" ) pagemap_offset = current_virtual_page_index * PAGEMAP_ENTRY_SIZE current_pfn_for_debug = -1 try: os.lseek(pagemap_fd, pagemap_offset, os.SEEK_SET) pagemap_entry_bytes = os.read(pagemap_fd, PAGEMAP_ENTRY_SIZE) if len(pagemap_entry_bytes) < PAGEMAP_ENTRY_SIZE: debug_print( f" Short read from pagemap for vpage idx {hex(current_virtual_page_index)}. Skipping." ) continue pagemap_entry = struct.unpack('> PM_PAGE_PRESENT_BIT) & 1 debug_print(f" Page present bit: {page_is_present}") if page_is_present: pfn = pagemap_entry & PM_PFN_MASK current_pfn_for_debug = pfn debug_print(f" PFN: 0x{pfn:x}") if pfn == 0: debug_print( " PFN is 0 (likely global zero page). Skipping dirty check for this page." ) continue kpageflags_offset = pfn * KPAGEFLAGS_ENTRY_SIZE os.lseek(kpageflags_fd, kpageflags_offset, os.SEEK_SET) kpageflags_bytes = os.read( kpageflags_fd, KPAGEFLAGS_ENTRY_SIZE ) if len(kpageflags_bytes) < KPAGEFLAGS_ENTRY_SIZE: debug_print( f" Short read from kpageflags for PFN {hex(pfn)}. Skipping." ) continue kpageflags_entry = struct.unpack('> 62) & 1 # if page_is_swapped: # swap_info = pagemap_entry & PM_PFN_MASK # debug_print(f" Page is swapped. Swap info: 0x{swap_info:x}") # else: # debug_print(f" Page not present and not swapped (hole or not yet mapped).") pass except OSError as e: pfn_str = ( hex(current_pfn_for_debug) if current_pfn_for_debug != -1 else "N/A" ) debug_print( f" OS error processing data for vpage {hex(current_virtual_page_index)} " f"(PFN: {pfn_str}): {os.strerror(e.errno)}. Skipping page." ) continue finally: if pagemap_fd != -1: os.close(pagemap_fd) if kpageflags_fd != -1: os.close(kpageflags_fd) return dirty_page_count def main(): global DEBUG_MODE parser = argparse.ArgumentParser( description="Count dirty memory pages in a specific region of a Linux process. " "Typically requires root privileges.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=f"""\ Example: sudo {sys.argv[0]} --debug 1234 0x7fa000000000 65536 Where: 1234 is the Process ID (PID). 0x7fa000000000 is the base address of the memory region. 65536 is the size of the memory region in bytes. """, ) parser.add_argument("pid", type=int, help="Process ID.") parser.add_argument( "base_address", type=str, help="Base address of the memory region (hex or decimal).", ) parser.add_argument( "map_size", type=int, help="Size of the memory region in bytes." ) parser.add_argument( "--debug", action="store_true", help="Enable detailed debug printing to stderr.", ) parser.add_argument( "--present", action="store_true", help="Count present pages, not dirty", ) args = parser.parse_args() if args.debug: DEBUG_MODE = True num_dirty = count_dirty_pages_in_region( args.pid, args.base_address, args.map_size, args.present ) print(f'map size {args.base_address} {args.map_size:,d}') print(' used pages', num_dirty) print(' used bytes', f'{num_dirty * 4096:,}') if __name__ == "__main__": main()