Intel

intel

MMUShell

Bases: MMUShell

Source code in mmushell/architectures/intel.py

class MMUShell(MMUShellDefault):
    def __init__(self, completekey="tab", stdin=None, stdout=None, machine={}):
        super(MMUShell, self).__init__(completekey, stdin, stdout, machine)

        if not self.data:
            self.data = Data(
                is_mem_parsed=False,
                is_radix_found=False,
                page_tables={
                    "global": [
                        {} for i in range(self.machine.mmu.radix_levels["global"])
                    ]
                },
                data_pages=[],
                empty_tables=[],
                reverse_map_tables=[
                    defaultdict(set)
                    for i in range(self.machine.mmu.radix_levels["global"])
                ],
                reverse_map_pages=[
                    defaultdict(set)
                    for i in range(self.machine.mmu.radix_levels["global"])
                ],
                idts=[],
                cr3s={},
            )

    def do_parse_memory(self, args):
        """Find MMU tables and IDTs"""
        if self.data.is_mem_parsed:
            logger.warning("Memory already parsed")
            return

        if type(self.machine.mmu) is IA32:
            self.parse_memory_ia32()
        elif type(self.machine.mmu) is PAE:
            self.parse_memory_pae()
        elif type(self.machine.mmu) is IA64:
            self.parse_memory_ia64()
        else:
            logging.fatal("OOPS... MMU class unkown!")
            exit(-1)
        self.data.is_mem_parsed = True

    def do_show_idt(self, args):
        """Show the IDT at a chosen address. Usage: show_idt ADDRESS"""
        args = args.split()
        if len(args) < 1:
            logger.warning("Missing table address")
            return

        try:
            addr = self.parse_int(args[0])
        except ValueError:
            logger.warning("Invalid table address")
            return

        print(self.machine.cpu.parse_idt(addr))

    def do_show_table(self, args):
        """Show an MMU table at a chosen address. Usage: show_table ADDRESS [level]"""
        args = args.split()
        if len(args) < 1:
            logger.warning("Missing table address")
            return

        try:
            addr = self.parse_int(args[0])
        except ValueError:
            logger.warning("Invalid table address")
            return

        if addr not in self.machine.memory:
            logger.warning("Table not in RAM range")
            return

        lvl = -1
        if len(args) > 1:
            try:
                lvl = self.parse_int(args[1])
                if lvl > (self.machine.mmu.radix_levels["global"] - 1):
                    raise ValueError
            except ValueError:
                logger.warning(
                    "Level must be an integer between 0 and {}".format(
                        str(self.machine.mmu.radix_levels["global"] - 1)
                    )
                )
                return

        if lvl == -1:
            table_size = self.machine.mmu.PAGE_SIZE
        else:
            table_size = self.machine.mmu.map_level_to_table_size["global"][lvl]
        table_buff = self.machine.memory.get_data(addr, table_size)
        invalids, pt_classes, table_obj = self.machine.mmu.parse_frame(
            table_buff, addr, table_size, lvl
        )
        print(table_obj)
        print(f"Invalid entries: {invalids} Table levels: {pt_classes}")

    def parse_memory_ia32(self):
        # Due to the impossibility to differentiate among data and page table (PT) in a simple way,
        # we first look for PDs and then only to PTs pointed by PDs, otherwise memory consumption is unmanageable...

        # Look for only PD, then use that to find only PT
        logger.info("Look for page directories..")
        self.machine.mmu.classify_entry = self.machine.mmu.classify_entry_pd_only
        parallel_results = self.machine.apply_parallel(
            self.machine.mmu.PAGE_SIZE, self.machine.mmu.parse_parallel_frame
        )

        logger.info("Reaggregate threads data...")
        for result in parallel_results:
            page_tables, data_pages, empty_tables = result.get()

            self.data.page_tables["global"][0].update(page_tables[0])
            self.data.data_pages.extend(data_pages)
            self.data.empty_tables.extend(empty_tables)

        logger.info("Collect page tables...")
        # Collect all PT addresses pointed by PD
        pt_candidates = set()
        for pd_obj in self.data.page_tables["global"][0].values():
            for entry_obj in pd_obj.entries[PDE32].values():
                pt_candidates.add(entry_obj.address)
        pt_candidates = list(pt_candidates)
        pt_candidates.sort()
        self.machine.mmu.classify_entry = self.machine.mmu.classify_entry_pt_only

        # Workaround to reduce thread memory consumption
        data = self.data
        self.data = None

        iterators = [
            (len(y), y)
            for y in [list(x) for x in divide(mp.cpu_count(), pt_candidates)]
        ]
        parsing_results_async = self.machine.apply_parallel(
            self.machine.mmu.PAGE_SIZE,
            self.machine.mmu.parse_parallel_frame,
            iterators=iterators,
        )

        # Restore previous data and set classify_entry to full version
        self.data = data
        self.machine.mmu.classify_entry = self.machine.mmu.classify_entry_full

        # Reaggregate data
        logger.info("Reaggregate threads data...")
        for result in parsing_results_async:
            page_tables, data_pages, empty_tables = result.get()
            self.data.page_tables["global"][1].update(page_tables[1])
            self.data.data_pages.extend(data_pages)
            self.data.empty_tables.extend(empty_tables)

        # Remove PT from data pages (in the first phase the alogrith has classified PT as data pages, now that
        # we know which PT is a true one, they must be removed from data pages)
        self.data.data_pages = set(self.data.data_pages)
        self.data.data_pages.difference_update(
            self.data.page_tables["global"][1].keys()
        )
        self.data.empty_tables = set(self.data.empty_tables)

        logger.info("Reduce false positives...")
        # Remove all tables which point to inexistent table of lower level
        for lvl in range(self.machine.mmu.radix_levels["global"] - 1):
            ptr_class = self.machine.mmu.map_ptr_entries_to_levels["global"][lvl]

            referenced_nxt = []
            for table_addr in list(self.data.page_tables["global"][lvl].keys()):
                for entry_obj in (
                    self.data.page_tables["global"][lvl][table_addr]
                    .entries[ptr_class]
                    .values()
                ):
                    if (
                        entry_obj.address
                        not in self.data.page_tables["global"][lvl + 1]
                        and entry_obj.address not in self.data.empty_tables
                    ):
                        # Remove the table
                        self.data.page_tables["global"][lvl].pop(table_addr)
                        break

                    else:
                        referenced_nxt.append(entry_obj.address)

            # Remove table not referenced by upper levels
            referenced_nxt = set(referenced_nxt)
            for table_addr in set(
                self.data.page_tables["global"][lvl + 1].keys()
            ).difference(referenced_nxt):
                self.data.page_tables["global"][lvl + 1].pop(table_addr)

        logger.info("Fill reverse maps...")
        for lvl in range(0, self.machine.mmu.radix_levels["global"]):
            ptr_class = self.machine.mmu.map_ptr_entries_to_levels["global"][lvl]
            page_class = self.machine.mmu.map_datapages_entries_to_levels["global"][
                lvl
            ][
                0
            ]  # Trick! Only one dataclass per level
            for table_addr, table_obj in self.data.page_tables["global"][lvl].items():
                for entry_obj in table_obj.entries[ptr_class].values():
                    self.data.reverse_map_tables[lvl][entry_obj.address].add(
                        table_obj.address
                    )
                for entry_obj in table_obj.entries[page_class].values():
                    self.data.reverse_map_pages[lvl][entry_obj.address].add(
                        table_obj.address
                    )

        logger.info("Look for interrupt tables...")
        self.data.idts = self.machine.cpu.find_idt_tables()

    def parse_memory_pae(self):
        # It uses the same function of IA64 but with a custom parse_parallel_frame
        self.parse_memory_ia64()

    def parse_memory_ia64(self):
        logger.info("Look for paging tables...")
        parallel_results = self.machine.apply_parallel(
            self.machine.mmu.PAGE_SIZE, self.machine.mmu.parse_parallel_frame
        )
        logger.info("Reaggregate threads data...")
        for result in parallel_results:
            page_tables, data_pages, empty_tables = result.get()

            for level in range(self.machine.mmu.radix_levels["global"]):
                self.data.page_tables["global"][level].update(page_tables[level])

            self.data.data_pages.extend(data_pages)
            self.data.empty_tables.extend(empty_tables)

        self.data.data_pages = set(self.data.data_pages)
        self.data.empty_tables = set(self.data.empty_tables)

        logger.info("Reduce false positives...")
        # Remove all tables which point to inexistent table of lower level
        for lvl in range(self.machine.mmu.radix_levels["global"] - 1):
            ptr_class = self.machine.mmu.map_ptr_entries_to_levels["global"][lvl]

            referenced_nxt = []
            for table_addr in list(self.data.page_tables["global"][lvl].keys()):
                for entry_obj in (
                    self.data.page_tables["global"][lvl][table_addr]
                    .entries[ptr_class]
                    .values()
                ):
                    if (
                        entry_obj.address
                        not in self.data.page_tables["global"][lvl + 1]
                        and entry_obj.address not in self.data.empty_tables
                    ):
                        # Remove the table
                        self.data.page_tables["global"][lvl].pop(table_addr)
                        break

                    else:
                        referenced_nxt.append(entry_obj.address)

            # Remove table not referenced by upper levels
            referenced_nxt = set(referenced_nxt)
            for table_addr in set(
                self.data.page_tables["global"][lvl + 1].keys()
            ).difference(referenced_nxt):
                self.data.page_tables["global"][lvl + 1].pop(table_addr)

        logger.info("Fill reverse maps...")
        for lvl in range(0, self.machine.mmu.radix_levels["global"]):
            ptr_class = self.machine.mmu.map_ptr_entries_to_levels["global"][lvl]
            page_class = self.machine.mmu.map_datapages_entries_to_levels["global"][
                lvl
            ][
                0
            ]  # Trick! Only one dataclass per level
            for table_addr, table_obj in self.data.page_tables["global"][lvl].items():
                for entry_obj in table_obj.entries[ptr_class].values():
                    self.data.reverse_map_tables[lvl][entry_obj.address].add(
                        table_obj.address
                    )
                for entry_obj in table_obj.entries[page_class].values():
                    self.data.reverse_map_pages[lvl][entry_obj.address].add(
                        table_obj.address
                    )

        logger.info("Look for interrupt tables...")
        self.data.idts = self.machine.cpu.find_idt_tables()

    def do_show_idtrs(self, args):
        """Show IDT tables founds"""
        if not self.data.is_radix_found:
            logging.info("Please, find them first!")
            return

        table = PrettyTable()
        table.field_names = ["Address", "Size"]
        for idt in self.data.idts:
            table.add_row([hex(idt.address), idt.size])
        print(table)

    def do_find_radix_trees(self, args):
        """Reconstruct radix trees"""

        # Some table level was not found...
        if not len(self.data.page_tables["global"][0]):
            logger.warning("OOPS... no tables in first level... Wrong MMU mode?")
            return

        cr3s = {}
        # No valid IDT found
        if not self.data.idts:
            logger.warning("No valid IDTs found, collect all valid CR3s...")

            # Start from page tables of lower level and derive upper level tables (aka CR3)
            # Filter for self-referencing CR3 (euristic) does not work with microkernels or SMAP/SMEP
            cr3_candidates = []
            already_explored = set()
            for page_addr in tqdm(self.data.data_pages.union(self.data.empty_tables)):
                derived_addresses = self.machine.mmu.derive_page_address(page_addr)
                for derived_address in derived_addresses:
                    if derived_address in already_explored:
                        continue
                    lvl, addr = derived_address
                    cr3_candidates.extend(
                        self.radix_roots_from_data_page(
                            lvl,
                            addr,
                            self.data.reverse_map_pages,
                            self.data.reverse_map_tables,
                        )
                    )
                    already_explored.add(derived_address)
            cr3_candidates = list(
                set(cr3_candidates).intersection(
                    self.data.page_tables["global"][0].keys()
                )
            )

            # Refine dataset and use a fake IDT table
            cr3s = {-1: {}}

            logger.info("Filter candidates...")
            for cr3 in tqdm(cr3_candidates):
                # Obtain radix tree infos
                consistency, pas = self.physpace(
                    cr3,
                    self.data.page_tables["global"],
                    self.data.empty_tables,
                    hierarchical=True,
                )

                # Only consistent trees are valid
                if not consistency:
                    continue

                # Esclude empty trees
                if pas.get_kernel_size() == pas.get_user_size() == 0:
                    continue

                vas = self.virtspace(
                    cr3, 0, self.machine.mmu.top_prefix, hierarchical=True
                )
                cr3s[-1][cr3] = RadixTree(cr3, 0, pas, vas)

            self.data.cr3s = cr3s
            self.data.is_radix_found = True
            return

        for idt_obj in self.data.idts:  # Cycle on all IDT found
            cr3_candidates = set()
            cr3s[idt_obj.address] = {}

            # We cannot filter radix trees root on the basis that the pointed tree is able to address its top table
            # this assumption is not valid for microkernels!

            # Collect all possible CR3: a valid CR3 must be able to address the IDT
            logger.info("Collect all valids CR3s...")
            idt_pg_addresses = self.machine.mmu.derive_page_address(
                idt_obj.address >> 12 << 12
            )

            for level, addr in idt_pg_addresses:
                cr3_candidates.update(
                    self.radix_roots_from_data_page(
                        level,
                        addr,
                        self.data.reverse_map_pages,
                        self.data.reverse_map_tables,
                    )
                )
            cr3_candidates = list(
                cr3_candidates.intersection(self.data.page_tables["global"][0].keys())
            )
            logger.info(
                "Number of possible CR3s for IDT located at {}:{}".format(
                    hex(idt_obj.address), len(cr3_candidates)
                )
            )

            # Collect the page containig each virtual addresses defined inside interrupt handlers
            handlers_pages = set()
            for handler in idt_obj.entries:
                # Task Entry does not point to interrupt hanlder
                if isinstance(handler, IDTTaskEntry32):
                    continue

                # Ignore handler unused
                if not handler.p:
                    continue

                handlers_pages.add(handler.offset >> 12 << 12)

            # Try to resolve interrupt virtual addresses and count the number of unresolved interrupt handlers
            cr3s_for_idt = []
            for cr3_candidate in cr3_candidates:
                errors = 0
                for vaddr in handlers_pages:
                    paddr = self.resolve_vaddr(cr3_candidate, vaddr)
                    if paddr == -1:
                        logging.debug(
                            f"find_radix_trees(): {hex(cr3_candidate)} failed to solve {hex(vaddr)}"
                        )
                        errors += 1

                cr3s_for_idt.append([cr3_candidate, errors])

            # At least one CR3 must be found...
            if not cr3s_for_idt:
                continue

            # Save only CR3s which resolv the max number of addresses
            cr3s_for_idt.sort(key=lambda x: (x[1], x[0]))
            max_value = cr3s_for_idt[0][1]
            logger.debug(
                "Interrupt pages: {}, Maximum pages resolved: {}".format(
                    len(handlers_pages), len(handlers_pages) - max_value
                )
            )

            # Consider only CR3 which resolve the maximum number of interrupt pages
            for cr3 in cr3s_for_idt:
                if max_value != cr3[1]:
                    break

                # Extract an approximation of the kernel and user physical address space
                consistency, pas = self.physpace(
                    cr3[0],
                    self.data.page_tables["global"],
                    self.data.empty_tables,
                    hierarchical=True,
                )

                # Only consistent trees are valid
                if not consistency:
                    continue

                # Esclude empty trees
                if pas.get_kernel_size() == pas.get_user_size() == 0:
                    continue

                vas = self.virtspace(
                    cr3[0], 0, self.machine.mmu.top_prefix, hierarchical=True
                )
                cr3s[idt_obj.address][cr3[0]] = RadixTree(cr3[0], 0, pas, vas)

        self.data.cr3s = cr3s
        self.data.is_radix_found = True

    def do_show_radix_trees(self, args):
        """Show radix trees found able to address a chosen IDT table. Usage: show_radix_trees PHY_IDT_ADDRESS"""
        if not self.data.is_radix_found:
            logging.info("Please, find them first!")
            return

        # Check if the IDT requested is in the list of IDT found
        args = args.split()
        if len(args) < 1:
            logging.warning("Missing IDT")
            return
        idt_addr = self.parse_int(args[0])

        if not self.data.idts:
            logging.info("No IDT found by MMUShell")
            idt_addr = -1
        else:
            for idt in self.data.idts:
                if idt_addr == idt.address:
                    break
            else:
                logging.warning("IDT requested not in IDT found!")
                return

        # Show results
        labels = [
            "Radix address",
            "First level",
            "Kernel size (Bytes)",
            "User size (Bytes)",
        ]
        table = PrettyTable()
        table.field_names = labels
        for cr3 in self.data.cr3s[idt_addr].values():
            table.add_row(cr3.entry_resume_stringified())
        table.sortby = "Radix address"
        print(table)

do_find_radix_trees(args)

Reconstruct radix trees

Source code in mmushell/architectures/intel.py

def do_find_radix_trees(self, args):
    """Reconstruct radix trees"""

    # Some table level was not found...
    if not len(self.data.page_tables["global"][0]):
        logger.warning("OOPS... no tables in first level... Wrong MMU mode?")
        return

    cr3s = {}
    # No valid IDT found
    if not self.data.idts:
        logger.warning("No valid IDTs found, collect all valid CR3s...")

        # Start from page tables of lower level and derive upper level tables (aka CR3)
        # Filter for self-referencing CR3 (euristic) does not work with microkernels or SMAP/SMEP
        cr3_candidates = []
        already_explored = set()
        for page_addr in tqdm(self.data.data_pages.union(self.data.empty_tables)):
            derived_addresses = self.machine.mmu.derive_page_address(page_addr)
            for derived_address in derived_addresses:
                if derived_address in already_explored:
                    continue
                lvl, addr = derived_address
                cr3_candidates.extend(
                    self.radix_roots_from_data_page(
                        lvl,
                        addr,
                        self.data.reverse_map_pages,
                        self.data.reverse_map_tables,
                    )
                )
                already_explored.add(derived_address)
        cr3_candidates = list(
            set(cr3_candidates).intersection(
                self.data.page_tables["global"][0].keys()
            )
        )

        # Refine dataset and use a fake IDT table
        cr3s = {-1: {}}

        logger.info("Filter candidates...")
        for cr3 in tqdm(cr3_candidates):
            # Obtain radix tree infos
            consistency, pas = self.physpace(
                cr3,
                self.data.page_tables["global"],
                self.data.empty_tables,
                hierarchical=True,
            )

            # Only consistent trees are valid
            if not consistency:
                continue

            # Esclude empty trees
            if pas.get_kernel_size() == pas.get_user_size() == 0:
                continue

            vas = self.virtspace(
                cr3, 0, self.machine.mmu.top_prefix, hierarchical=True
            )
            cr3s[-1][cr3] = RadixTree(cr3, 0, pas, vas)

        self.data.cr3s = cr3s
        self.data.is_radix_found = True
        return

    for idt_obj in self.data.idts:  # Cycle on all IDT found
        cr3_candidates = set()
        cr3s[idt_obj.address] = {}

        # We cannot filter radix trees root on the basis that the pointed tree is able to address its top table
        # this assumption is not valid for microkernels!

        # Collect all possible CR3: a valid CR3 must be able to address the IDT
        logger.info("Collect all valids CR3s...")
        idt_pg_addresses = self.machine.mmu.derive_page_address(
            idt_obj.address >> 12 << 12
        )

        for level, addr in idt_pg_addresses:
            cr3_candidates.update(
                self.radix_roots_from_data_page(
                    level,
                    addr,
                    self.data.reverse_map_pages,
                    self.data.reverse_map_tables,
                )
            )
        cr3_candidates = list(
            cr3_candidates.intersection(self.data.page_tables["global"][0].keys())
        )
        logger.info(
            "Number of possible CR3s for IDT located at {}:{}".format(
                hex(idt_obj.address), len(cr3_candidates)
            )
        )

        # Collect the page containig each virtual addresses defined inside interrupt handlers
        handlers_pages = set()
        for handler in idt_obj.entries:
            # Task Entry does not point to interrupt hanlder
            if isinstance(handler, IDTTaskEntry32):
                continue

            # Ignore handler unused
            if not handler.p:
                continue

            handlers_pages.add(handler.offset >> 12 << 12)

        # Try to resolve interrupt virtual addresses and count the number of unresolved interrupt handlers
        cr3s_for_idt = []
        for cr3_candidate in cr3_candidates:
            errors = 0
            for vaddr in handlers_pages:
                paddr = self.resolve_vaddr(cr3_candidate, vaddr)
                if paddr == -1:
                    logging.debug(
                        f"find_radix_trees(): {hex(cr3_candidate)} failed to solve {hex(vaddr)}"
                    )
                    errors += 1

            cr3s_for_idt.append([cr3_candidate, errors])

        # At least one CR3 must be found...
        if not cr3s_for_idt:
            continue

        # Save only CR3s which resolv the max number of addresses
        cr3s_for_idt.sort(key=lambda x: (x[1], x[0]))
        max_value = cr3s_for_idt[0][1]
        logger.debug(
            "Interrupt pages: {}, Maximum pages resolved: {}".format(
                len(handlers_pages), len(handlers_pages) - max_value
            )
        )

        # Consider only CR3 which resolve the maximum number of interrupt pages
        for cr3 in cr3s_for_idt:
            if max_value != cr3[1]:
                break

            # Extract an approximation of the kernel and user physical address space
            consistency, pas = self.physpace(
                cr3[0],
                self.data.page_tables["global"],
                self.data.empty_tables,
                hierarchical=True,
            )

            # Only consistent trees are valid
            if not consistency:
                continue

            # Esclude empty trees
            if pas.get_kernel_size() == pas.get_user_size() == 0:
                continue

            vas = self.virtspace(
                cr3[0], 0, self.machine.mmu.top_prefix, hierarchical=True
            )
            cr3s[idt_obj.address][cr3[0]] = RadixTree(cr3[0], 0, pas, vas)

    self.data.cr3s = cr3s
    self.data.is_radix_found = True

do_parse_memory(args)

Find MMU tables and IDTs

Source code in mmushell/architectures/intel.py

def do_parse_memory(self, args):
    """Find MMU tables and IDTs"""
    if self.data.is_mem_parsed:
        logger.warning("Memory already parsed")
        return

    if type(self.machine.mmu) is IA32:
        self.parse_memory_ia32()
    elif type(self.machine.mmu) is PAE:
        self.parse_memory_pae()
    elif type(self.machine.mmu) is IA64:
        self.parse_memory_ia64()
    else:
        logging.fatal("OOPS... MMU class unkown!")
        exit(-1)
    self.data.is_mem_parsed = True

do_show_idt(args)

Show the IDT at a chosen address. Usage: show_idt ADDRESS

Source code in mmushell/architectures/intel.py

def do_show_idt(self, args):
    """Show the IDT at a chosen address. Usage: show_idt ADDRESS"""
    args = args.split()
    if len(args) < 1:
        logger.warning("Missing table address")
        return

    try:
        addr = self.parse_int(args[0])
    except ValueError:
        logger.warning("Invalid table address")
        return

    print(self.machine.cpu.parse_idt(addr))

do_show_idtrs(args)

Show IDT tables founds

Source code in mmushell/architectures/intel.py

def do_show_idtrs(self, args):
    """Show IDT tables founds"""
    if not self.data.is_radix_found:
        logging.info("Please, find them first!")
        return

    table = PrettyTable()
    table.field_names = ["Address", "Size"]
    for idt in self.data.idts:
        table.add_row([hex(idt.address), idt.size])
    print(table)

do_show_radix_trees(args)

Show radix trees found able to address a chosen IDT table. Usage: show_radix_trees PHY_IDT_ADDRESS

Source code in mmushell/architectures/intel.py

def do_show_radix_trees(self, args):
    """Show radix trees found able to address a chosen IDT table. Usage: show_radix_trees PHY_IDT_ADDRESS"""
    if not self.data.is_radix_found:
        logging.info("Please, find them first!")
        return

    # Check if the IDT requested is in the list of IDT found
    args = args.split()
    if len(args) < 1:
        logging.warning("Missing IDT")
        return
    idt_addr = self.parse_int(args[0])

    if not self.data.idts:
        logging.info("No IDT found by MMUShell")
        idt_addr = -1
    else:
        for idt in self.data.idts:
            if idt_addr == idt.address:
                break
        else:
            logging.warning("IDT requested not in IDT found!")
            return

    # Show results
    labels = [
        "Radix address",
        "First level",
        "Kernel size (Bytes)",
        "User size (Bytes)",
    ]
    table = PrettyTable()
    table.field_names = labels
    for cr3 in self.data.cr3s[idt_addr].values():
        table.add_row(cr3.entry_resume_stringified())
    table.sortby = "Radix address"
    print(table)

do_show_table(args)

Show an MMU table at a chosen address. Usage: show_table ADDRESS [level]

Source code in mmushell/architectures/intel.py

def do_show_table(self, args):
    """Show an MMU table at a chosen address. Usage: show_table ADDRESS [level]"""
    args = args.split()
    if len(args) < 1:
        logger.warning("Missing table address")
        return

    try:
        addr = self.parse_int(args[0])
    except ValueError:
        logger.warning("Invalid table address")
        return

    if addr not in self.machine.memory:
        logger.warning("Table not in RAM range")
        return

    lvl = -1
    if len(args) > 1:
        try:
            lvl = self.parse_int(args[1])
            if lvl > (self.machine.mmu.radix_levels["global"] - 1):
                raise ValueError
        except ValueError:
            logger.warning(
                "Level must be an integer between 0 and {}".format(
                    str(self.machine.mmu.radix_levels["global"] - 1)
                )
            )
            return

    if lvl == -1:
        table_size = self.machine.mmu.PAGE_SIZE
    else:
        table_size = self.machine.mmu.map_level_to_table_size["global"][lvl]
    table_buff = self.machine.memory.get_data(addr, table_size)
    invalids, pt_classes, table_obj = self.machine.mmu.parse_frame(
        table_buff, addr, table_size, lvl
    )
    print(table_obj)
    print(f"Invalid entries: {invalids} Table levels: {pt_classes}")

MMUShellGTruth

Bases: MMUShell

Source code in mmushell/architectures/intel.py

class MMUShellGTruth(MMUShell):
    def do_show_idtrs_gtruth(self, args):
        """Compare IDTs found with the ground truth"""
        if not self.data.is_mem_parsed:
            logging.info("Please, find them first!")
            return

        # We need to use a CR3 to translate the IDTR in virtual address, we use the last CR3 defined
        cr3_class = self.machine.mmu.cr3_class

        # Filter CR3 for valid one and the find the last used one
        keys = list(self.gtruth["CR3"].keys())
        keys.sort(key=lambda x: self.gtruth["CR3"][x][1], reverse=True)

        for cr3 in keys:
            cr3_obj = cr3_class(cr3)

            # Validate CR3
            if cr3_obj.address not in self.data.page_tables["global"][0]:
                continue
            consistency, pas = self.physpace(
                cr3_obj.address,
                self.data.page_tables["global"],
                self.data.empty_tables,
                hierarchical=True,
            )
            if not consistency or (
                not pas.get_kernel_size() and not pas.get_user_size()
            ):
                continue
            else:
                valid_cr3_obj = cr3_obj
                break
        else:
            logging.warning("OOPS.. no valid CR3 found")
            return

        # Collect all found physical addresses
        idts = [x.address for x in self.data.idts]

        # Resolve the IDT virtual address
        table = PrettyTable()
        table.field_names = [
            "Virtual address",
            "Physical address",
            "Found",
            "First seen",
            "Last seen",
        ]

        tp = 0
        unresolved = 0
        keys = list(self.gtruth["IDTR"].keys())
        keys.sort(key=lambda x: self.gtruth["IDTR"][x][1])
        for idtr in keys:
            idtr_obj = IDTR(idtr)

            paddr = self.resolve_vaddr(valid_cr3_obj.address, idtr_obj.address)
            # Not solved by the CR3...
            if paddr == -1:
                unresolved += 1
                table.add_row(
                    [
                        hex(idtr_obj.address),
                        "?",
                        "?",
                        self.gtruth["IDTR"][idtr_obj.value][0],
                        self.gtruth["IDTR"][idtr_obj.value][1],
                    ]
                )
            else:
                if paddr in idts:
                    tp += 1
                    found = "X"
                else:
                    found = ""
                table.add_row(
                    [
                        hex(idtr_obj.address),
                        hex(paddr),
                        found,
                        self.gtruth["IDTR"][idtr_obj.value][0],
                        self.gtruth["IDTR"][idtr_obj.value][1],
                    ]
                )

        print(f"Use CR3 address: {hex(valid_cr3_obj.address)}")
        print(table)
        print(f"TP:{tp} FP:{len(idts) - tp} Unresolved: {unresolved}")

        # Export results for next analysis
        if len(args) == 2 and args[1] == "export":
            from pickle import dump as dump_p

            with open("dump.mmu", "wb") as f:
                results = [{"cr3": tp} for tp in sorted(tps)]
                dump_p(results, f)

    def do_show_radix_trees_gtruth(self, args):
        """Compare radix trees found able to address a chosen IDT table with the ground truth. Usage: show_radix_trees_gtruth PHY_IDT_ADDRESS"""
        if not self.data.is_radix_found:
            logging.info("Please, find them first!")
            return

        # Check if the IDT requested is in the list of IDT found
        args = args.split()
        if len(args) < 1:
            logging.warning("Missing IDT")
            return
        idt_addr = self.parse_int(args[0])
        if idt_addr not in self.machine.memory:
            logging.warning("IDT address not in RAM")
            return

        if not self.data.idts:
            # If no valid IDT has been found by MMUShell, it parses and uses the IDT
            # address pass by the user as filter to identify TP radix trees
            logging.info("No IDT found by MMUShell")

            # Parse IDT
            idt = self.machine.cpu.parse_idt(idt_addr)
            if not len(idt):
                logging.warning("No IDT at address")
                return
        else:
            for idt in self.data.idts:
                if idt_addr == idt.address:
                    break
            else:
                logging.warning("IDT requested not in IDT found!")
                return

        # Collect all valid CR3
        latest_idt_va_used = IDTR(
            sorted(
                list(self.gtruth["IDTR"].keys()),
                key=lambda x: self.gtruth["IDTR"][x][1],
            )[-1]
        )
        idts = {}
        cr3_errors = defaultdict(list)

        for cr3 in self.gtruth["CR3"]:
            cr3_obj = self.machine.mmu.cr3_class(cr3)
            if cr3_obj.address not in self.data.page_tables["global"][0]:
                continue
            consistency, pas = self.physpace(
                cr3_obj.address,
                self.data.page_tables["global"],
                self.data.empty_tables,
                hierarchical=True,
            )
            if not consistency or (
                not pas.get_kernel_size() and not pas.get_user_size()
            ):
                continue

            # Check if they are able to address the IDT table
            derived_addresses = self.machine.mmu.derive_page_address(
                idt_addr >> 12 << 12
            )
            if not any([x[1] in pas for x in derived_addresses]):
                continue  # Trick! Only one dataclass per level

            # Check if the CR3 is able to resolve the latest IDTR value used
            # (we check this for simplicity instead of the VA associated with the selected IDT)
            idt_phys = self.resolve_vaddr(cr3_obj.address, latest_idt_va_used.address)
            if idt_phys == -1 or idt_phys not in self.machine.memory:
                continue

            # Collect interrupt VA for resolved IDR
            if idt_phys not in idts:
                vas_interrupts = set()
                idt_obj = self.machine.cpu.parse_idt(idt_phys)
                for handler in idt_obj.entries:
                    if isinstance(handler, IDTTaskEntry32) or not handler.p:
                        continue
                    vas_interrupts.add(handler.offset >> 12 << 12)
                idts[idt_phys] = vas_interrupts

            # Check how much IDT interrupt VA is able to resolve
            errors = 0
            for va in idts[idt_phys]:
                if self.resolve_vaddr(cr3_obj.address, va) == -1:
                    errors += 1
            cr3_errors[errors].append(cr3_obj)

        # Use only CR3 with the minimum number of errors
        valid_cr3s = {}
        for cr3_obj in cr3_errors[sorted(list(cr3_errors.keys()))[0]]:
            valid_cr3s[cr3_obj.address] = cr3_obj

        # Use fake IDT address if no IDT are found
        if not self.data.idts:
            idt_addr = -1

        # True positives, false negatives, false positives
        tps = set(valid_cr3s.keys()).intersection(set(self.data.cr3s[idt_addr].keys()))
        fns = set(valid_cr3s.keys()).difference(set(self.data.cr3s[idt_addr].keys()))
        fps = set(self.data.cr3s[idt_addr].keys()).difference(set(valid_cr3s.keys()))

        # Show results
        table = PrettyTable()
        table.field_names = ["Address", "Found", "First seen", "Last seen"]
        for tp in sorted(tps):
            table.add_row(
                [
                    hex(tp),
                    "X",
                    self.gtruth["CR3"][valid_cr3s[tp].value][0],
                    self.gtruth["CR3"][valid_cr3s[tp].value][1],
                ]
            )

        for fn in sorted(fns):
            table.add_row(
                [
                    hex(fn),
                    "",
                    self.gtruth["CR3"][valid_cr3s[fn].value][0],
                    self.gtruth["CR3"][valid_cr3s[fn].value][1],
                ]
            )

        for fp in sorted(fps):
            table.add_row([hex(fp), "False positive", "", ""])

        print(table)
        print(f"TP:{len(tps)} FN:{len(fns)} FP:{len(fps)}")

        # Export results for next analysis
        if len(args) == 2 and args[1] == "export":
            from pickle import dump as dump_p

            with open("dump.mmu", "wb") as f:
                results = [{"cr3": tp} for tp in sorted(tps)]
                dump_p(results, f)

do_show_idtrs_gtruth(args)

Compare IDTs found with the ground truth

Source code in mmushell/architectures/intel.py

def do_show_idtrs_gtruth(self, args):
    """Compare IDTs found with the ground truth"""
    if not self.data.is_mem_parsed:
        logging.info("Please, find them first!")
        return

    # We need to use a CR3 to translate the IDTR in virtual address, we use the last CR3 defined
    cr3_class = self.machine.mmu.cr3_class

    # Filter CR3 for valid one and the find the last used one
    keys = list(self.gtruth["CR3"].keys())
    keys.sort(key=lambda x: self.gtruth["CR3"][x][1], reverse=True)

    for cr3 in keys:
        cr3_obj = cr3_class(cr3)

        # Validate CR3
        if cr3_obj.address not in self.data.page_tables["global"][0]:
            continue
        consistency, pas = self.physpace(
            cr3_obj.address,
            self.data.page_tables["global"],
            self.data.empty_tables,
            hierarchical=True,
        )
        if not consistency or (
            not pas.get_kernel_size() and not pas.get_user_size()
        ):
            continue
        else:
            valid_cr3_obj = cr3_obj
            break
    else:
        logging.warning("OOPS.. no valid CR3 found")
        return

    # Collect all found physical addresses
    idts = [x.address for x in self.data.idts]

    # Resolve the IDT virtual address
    table = PrettyTable()
    table.field_names = [
        "Virtual address",
        "Physical address",
        "Found",
        "First seen",
        "Last seen",
    ]

    tp = 0
    unresolved = 0
    keys = list(self.gtruth["IDTR"].keys())
    keys.sort(key=lambda x: self.gtruth["IDTR"][x][1])
    for idtr in keys:
        idtr_obj = IDTR(idtr)

        paddr = self.resolve_vaddr(valid_cr3_obj.address, idtr_obj.address)
        # Not solved by the CR3...
        if paddr == -1:
            unresolved += 1
            table.add_row(
                [
                    hex(idtr_obj.address),
                    "?",
                    "?",
                    self.gtruth["IDTR"][idtr_obj.value][0],
                    self.gtruth["IDTR"][idtr_obj.value][1],
                ]
            )
        else:
            if paddr in idts:
                tp += 1
                found = "X"
            else:
                found = ""
            table.add_row(
                [
                    hex(idtr_obj.address),
                    hex(paddr),
                    found,
                    self.gtruth["IDTR"][idtr_obj.value][0],
                    self.gtruth["IDTR"][idtr_obj.value][1],
                ]
            )

    print(f"Use CR3 address: {hex(valid_cr3_obj.address)}")
    print(table)
    print(f"TP:{tp} FP:{len(idts) - tp} Unresolved: {unresolved}")

    # Export results for next analysis
    if len(args) == 2 and args[1] == "export":
        from pickle import dump as dump_p

        with open("dump.mmu", "wb") as f:
            results = [{"cr3": tp} for tp in sorted(tps)]
            dump_p(results, f)

do_show_radix_trees_gtruth(args)

Compare radix trees found able to address a chosen IDT table with the ground truth. Usage: show_radix_trees_gtruth PHY_IDT_ADDRESS

Source code in mmushell/architectures/intel.py

def do_show_radix_trees_gtruth(self, args):
    """Compare radix trees found able to address a chosen IDT table with the ground truth. Usage: show_radix_trees_gtruth PHY_IDT_ADDRESS"""
    if not self.data.is_radix_found:
        logging.info("Please, find them first!")
        return

    # Check if the IDT requested is in the list of IDT found
    args = args.split()
    if len(args) < 1:
        logging.warning("Missing IDT")
        return
    idt_addr = self.parse_int(args[0])
    if idt_addr not in self.machine.memory:
        logging.warning("IDT address not in RAM")
        return

    if not self.data.idts:
        # If no valid IDT has been found by MMUShell, it parses and uses the IDT
        # address pass by the user as filter to identify TP radix trees
        logging.info("No IDT found by MMUShell")

        # Parse IDT
        idt = self.machine.cpu.parse_idt(idt_addr)
        if not len(idt):
            logging.warning("No IDT at address")
            return
    else:
        for idt in self.data.idts:
            if idt_addr == idt.address:
                break
        else:
            logging.warning("IDT requested not in IDT found!")
            return

    # Collect all valid CR3
    latest_idt_va_used = IDTR(
        sorted(
            list(self.gtruth["IDTR"].keys()),
            key=lambda x: self.gtruth["IDTR"][x][1],
        )[-1]
    )
    idts = {}
    cr3_errors = defaultdict(list)

    for cr3 in self.gtruth["CR3"]:
        cr3_obj = self.machine.mmu.cr3_class(cr3)
        if cr3_obj.address not in self.data.page_tables["global"][0]:
            continue
        consistency, pas = self.physpace(
            cr3_obj.address,
            self.data.page_tables["global"],
            self.data.empty_tables,
            hierarchical=True,
        )
        if not consistency or (
            not pas.get_kernel_size() and not pas.get_user_size()
        ):
            continue

        # Check if they are able to address the IDT table
        derived_addresses = self.machine.mmu.derive_page_address(
            idt_addr >> 12 << 12
        )
        if not any([x[1] in pas for x in derived_addresses]):
            continue  # Trick! Only one dataclass per level

        # Check if the CR3 is able to resolve the latest IDTR value used
        # (we check this for simplicity instead of the VA associated with the selected IDT)
        idt_phys = self.resolve_vaddr(cr3_obj.address, latest_idt_va_used.address)
        if idt_phys == -1 or idt_phys not in self.machine.memory:
            continue

        # Collect interrupt VA for resolved IDR
        if idt_phys not in idts:
            vas_interrupts = set()
            idt_obj = self.machine.cpu.parse_idt(idt_phys)
            for handler in idt_obj.entries:
                if isinstance(handler, IDTTaskEntry32) or not handler.p:
                    continue
                vas_interrupts.add(handler.offset >> 12 << 12)
            idts[idt_phys] = vas_interrupts

        # Check how much IDT interrupt VA is able to resolve
        errors = 0
        for va in idts[idt_phys]:
            if self.resolve_vaddr(cr3_obj.address, va) == -1:
                errors += 1
        cr3_errors[errors].append(cr3_obj)

    # Use only CR3 with the minimum number of errors
    valid_cr3s = {}
    for cr3_obj in cr3_errors[sorted(list(cr3_errors.keys()))[0]]:
        valid_cr3s[cr3_obj.address] = cr3_obj

    # Use fake IDT address if no IDT are found
    if not self.data.idts:
        idt_addr = -1

    # True positives, false negatives, false positives
    tps = set(valid_cr3s.keys()).intersection(set(self.data.cr3s[idt_addr].keys()))
    fns = set(valid_cr3s.keys()).difference(set(self.data.cr3s[idt_addr].keys()))
    fps = set(self.data.cr3s[idt_addr].keys()).difference(set(valid_cr3s.keys()))

    # Show results
    table = PrettyTable()
    table.field_names = ["Address", "Found", "First seen", "Last seen"]
    for tp in sorted(tps):
        table.add_row(
            [
                hex(tp),
                "X",
                self.gtruth["CR3"][valid_cr3s[tp].value][0],
                self.gtruth["CR3"][valid_cr3s[tp].value][1],
            ]
        )

    for fn in sorted(fns):
        table.add_row(
            [
                hex(fn),
                "",
                self.gtruth["CR3"][valid_cr3s[fn].value][0],
                self.gtruth["CR3"][valid_cr3s[fn].value][1],
            ]
        )

    for fp in sorted(fps):
        table.add_row([hex(fp), "False positive", "", ""])

    print(table)
    print(f"TP:{len(tps)} FN:{len(fns)} FP:{len(fps)}")

    # Export results for next analysis
    if len(args) == 2 and args[1] == "export":
        from pickle import dump as dump_p

        with open("dump.mmu", "wb") as f:
            results = [{"cr3": tp} for tp in sorted(tps)]
            dump_p(results, f)