Python analysis script for /proc/<pid>/maps data under Linux

vmmap.py

#!/usr/bin/python
# coding=utf-8

"""Tool to analyze and display the contents of /proc/<pid>/maps"""

import re
import itertools
import argparse
from dataclasses import dataclass

MAPS_LINE_RE = re.compile(r"""
    (?P<addr_start>[0-9a-f]+)-(?P<addr_end>[0-9a-f]+)\s+  # Address
    (?P<perms>\S+)\s+                                     # Permissions
    (?P<offset>[0-9a-f]+)\s+                              # Map offset
    (?P<dev>\S+)\s+                                       # Device node
    (?P<inode>\d+)\s+                                     # Inode
    (?P<pathname>.*)\s+                                   # Pathname
""", re.VERBOSE)


def human_bytes(size):
    modifier = 1
    while size > 1024:
        modifier *= 1024
        size /= 1024
    return "%.1f%s" % (size, {
        1024**0: 'b',
        1024**1: 'k',
        1024**2: 'M',
        1024**3: 'G',
        1024**4: 'T',
    }.get(modifier, " x%d" % modifier))


@dataclass
class Record:
    addr_start: int
    addr_end: int
    perms: str
    offset: int
    dev: str
    inode: int
    pathname: str

    @property
    def size(self):
        return self.addr_end - self.addr_start

    @property
    def human_size(self):
        return human_bytes(self.size)

    @property
    def readable(self):
        return self.perms[0] == "r"

    @property
    def writable(self):
        return self.perms[1] == "w"

    @property
    def executable(self):
        return self.perms[2] == "x"

    @property
    def shared(self):
        return self.perms[3] == "s"

    @property
    def private(self):
        return self.perms[3] == "p"

    @classmethod
    def parse(self, pid):
        records = []
        with open("/proc/%d/maps" % pid) as fd:
            for line in fd:
                m = MAPS_LINE_RE.match(line)
                if not m:
                    print("Skipping: %s" % line)
                    continue
                addr_start, addr_end, perms, offset, dev, inode, pathname = m.groups()
                addr_start = int(addr_start, 16)
                addr_end = int(addr_end, 16)
                offset = int(offset, 16)
                records.append(Record(
                    addr_start=addr_start,
                    addr_end=addr_end,
                    perms=perms,
                    offset=offset,
                    dev=dev,
                    inode=inode,
                    pathname=pathname,
                ))
        return records

    @classmethod
    def aggregate(self, records, only_used=False, only_private=False):

        named_records = {}
        anonymous_records = []
        for record in records:
            if only_private and not record.private:
                continue
            if only_used and not record.readable and not record.writable and not record.shared and not record.pathname:
                continue
            if record.pathname:
                if record.pathname in named_records:
                    other = named_records[record.pathname]
                    named_records[record.pathname] = Record(
                        min(record.addr_start, other.addr_start),
                        max(record.addr_end, other.addr_end),
                        perms=''.join("?" if c1 != c2 else c1 for c1, c2 in zip(record.perms, other.perms)),
                        offset=0,
                        dev='',
                        inode='',
                        pathname=record.pathname,
                    )
                else:
                    named_records[record.pathname] = record
            else:
                anonymous_records.append(record)

        return list(sorted(
            itertools.chain(anonymous_records, named_records.values()),
            key=lambda r: r.size,
            reverse=True,
        ))



if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument("pid", type=int, help="Process identifier (pid)")
    parser.add_argument("--only-used", "-u", action="store_true", help="Only show used pages (non readable, writable, executable and private pages)")
    parser.add_argument("--only-private", "-p", action="store_true", help="Only show private pages")
    args = parser.parse_args()

    records = Record.parse(args.pid)
    records = Record.aggregate(records, only_used=args.only_used, only_private=args.only_private)

    print("\t".join([
        "% 16s" % "Start of range",
        "% 16s" % "End of range",
        "% 12s" % "Size",
        "% 4s" % "Perms",
        "Path",
    ]))
    for record in records:
        print("\t".join([
            "%016x" % record.addr_start,
            "%016x" % record.addr_end,
            "% 12s" % record.human_size,
            "% 4s" % record.perms,
            record.pathname,
        ]))

    print("")
    print("Total: %s" % human_bytes(sum(r.size for r in records)))