ARM

ARM

MMUShell

Bases: MMUShell

Source code in mmushell/architectures/arm.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_tables_found=False,
                is_radix_found=False,
                is_registers_found=False,
                opcodes={},
                regs_values={},
                page_tables={
                    "user": [
                        {} for i in range(self.machine.mmu.radix_levels["global"])
                    ],
                    "kernel": [
                        {} for i in range(self.machine.mmu.radix_levels["global"])
                    ],
                },
                data_pages=[],
                empty_tables=[],
                reverse_map_tables={
                    "user": [
                        defaultdict(set)
                        for i in range(self.machine.mmu.radix_levels["global"])
                    ],
                    "kernel": [
                        defaultdict(set)
                        for i in range(self.machine.mmu.radix_levels["global"])
                    ],
                },
                reverse_map_pages={
                    "user": [
                        defaultdict(set)
                        for i in range(self.machine.mmu.radix_levels["global"])
                    ],
                    "kernel": [
                        defaultdict(set)
                        for i in range(self.machine.mmu.radix_levels["global"])
                    ],
                },
                used_ttbcr=None,
                ttbrs=defaultdict(dict),
            )

    def reload_data_from_file(self, data_filename):
        super(MMUShell, self).reload_data_from_file(data_filename)
        SHORT.ttbcr_n = self.data.used_ttbcr.n

    def do_find_registers_values(self, arg):
        """Find MMU load opcodes and execute MMU related functions inside the memory dump in order to extract MMU registers values"""

        if self.data.is_registers_found:
            logging.warning("Registers already searched")
            return

        logger.info("Look for opcodes related to MMU setup...")
        parallel_results = self.machine.apply_parallel(
            self.machine.mmu.PAGE_SIZE, self.machine.cpu.parse_opcodes_parallel
        )

        opcodes = {}
        logger.info("Reaggregate threads data...")
        for result in parallel_results:
            opcodes.update(result.get())

        self.data.opcodes = opcodes

        # Filter to look only for opcodes which write on MMU register only and not read from them or from other registers
        filter_f = (
            lambda it: True
            if it[1]["register"] == "TTBCR" and it[1]["instruction"] == "MCR"
            else False
        )
        mmu_wr_opcodes = {k: v for k, v in filter(filter_f, opcodes.items())}

        logging.info("Use heuristics to find function addresses...")
        logging.info("This analysis could be extremely slow!")
        self.machine.cpu.identify_functions_start(mmu_wr_opcodes)

        logging.info("Identify register values using data flow analysis...")

        # We use data flow analysis and merge the results
        dataflow_values = self.machine.cpu.find_registers_values_dataflow(
            mmu_wr_opcodes
        )

        filtered_values = defaultdict(set)
        for register, values in dataflow_values.items():
            for value in values:
                reg_obj = CPURegARM32.get_register_obj(register, value)
                if reg_obj.valid and not any(
                    [
                        val_obj.is_mmu_equivalent_to(reg_obj)
                        for val_obj in filtered_values[register]
                    ]
                ):
                    filtered_values[register].add(reg_obj)

        # Add default values
        reg_obj = CPURegARM32.get_register_obj(
            "TTBCR", self.machine.cpu.registers_values["TTBCR"]
        )
        if reg_obj.valid and all(
            [not reg_obj.is_mmu_equivalent_to(x) for x in filtered_values["TTBCR"]]
        ):
            filtered_values["TTBCR"].add(reg_obj)

        self.data.regs_values = filtered_values
        self.data.is_registers_found = True

        # Show results
        logging.info("TTBCR values recovered:")
        for reg_obj in self.data.regs_values["TTBCR"]:
            logging.info(reg_obj)

    def do_show_registers(self, args):
        """Show registers values found"""
        if not self.data.is_registers_found:
            logging.info("Please, find them first!")
            return

        for registers in sorted(self.data.regs_values.keys()):
            for register in self.data.regs_values[registers]:
                print(register)

    def do_set_ttbcr(self, args):
        """Set the value of TTBCR register to be used"""
        args = args.split()
        if len(args) == 0:
            logging.error("Please use find_tables TTBCR_VALUE")
            return

        # The shape of PTL0 tables depends on N value of TTBCR register
        try:
            ttbcr_val = self.parse_int(args[0])
            ttbcr = TTBCR(ttbcr_val)
            if not ttbcr.valid:
                raise ValueError
        except ValueError:
            logger.warning("Invalid TTBCR value")
            return

        self.data.used_ttbcr = ttbcr
        SHORT.ttbcr_n = ttbcr.n

    def do_find_tables(self, args):
        """Find MMU tables in memory"""
        if not self.data.used_ttbcr:
            logging.error("Please set a TTBCR register to use, using set_ttbcr TTBCR")
            return
        ttbcr = self.data.used_ttbcr

        # Delete all the previous table data
        if self.data.is_tables_found:
            for mode in ["user", "kernel"]:
                self.data.reverse_map_pages[mode].clear()
                self.data.reverse_map_tables[mode].clear()
                self.data.page_tables[mode].clear()
            self.data.empty_tables = []
            self.data.data_pages = []

        # Parse memory in chunk of 16KiB
        PTL0_USER_TABLE_SIZE = ttbcr.get_ptl0_user_table_size()
        logger.info("Look for paging tables...")
        parallel_results = self.machine.apply_parallel(
            16384,
            self.machine.mmu.parse_parallel_frame,
            ptl0_u_size=PTL0_USER_TABLE_SIZE,
        )
        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["user"][level].update(page_tables["user"][level])
                self.data.page_tables["kernel"][level].update(
                    page_tables["kernel"][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)

        # Prepare fo dumplicates removing
        modes = ["kernel"]
        fps = {"kernel": []}
        if PTL0_USER_TABLE_SIZE != 16384:
            modes.append("user")
            fps["user"] = []
            self.data.page_tables["user"][1] = deepcopy(
                self.data.page_tables["kernel"][1]
            )

        # Remove all tables which point to inexistent table of lower level
        logger.info("Reduce false positives...")
        ptr_class = self.machine.mmu.map_ptr_entries_to_levels["global"][0]

        for mode in modes:
            referenced_nxt = []
            for table_addr in list(self.data.page_tables[mode][0].keys()):
                for entry_obj in (
                    self.data.page_tables[mode][0][table_addr]
                    .entries[ptr_class]
                    .values()
                ):
                    if (
                        entry_obj.address not in self.data.page_tables[mode][1]
                        and entry_obj.address not in self.data.empty_tables
                    ):
                        # Remove the table
                        self.data.page_tables[mode][0].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[mode][1].keys()).difference(
                referenced_nxt
            ):
                self.data.page_tables[mode][1].pop(table_addr)

        logger.info("Fill reverse maps...")
        for mode in modes:
            for lvl in range(0, self.machine.mmu.radix_levels["global"]):
                ptr_class = self.machine.mmu.map_ptr_entries_to_levels["global"][lvl]
                page_classes = self.machine.mmu.map_datapages_entries_to_levels[
                    "global"
                ][lvl]
                for table_addr, table_obj in self.data.page_tables[mode][lvl].items():
                    for entry_obj in table_obj.entries[ptr_class].values():
                        self.data.reverse_map_tables[mode][lvl][entry_obj.address].add(
                            table_obj.address
                        )
                    for page_class in page_classes:
                        for entry_obj in table_obj.entries[page_class].values():
                            self.data.reverse_map_pages[mode][lvl][
                                entry_obj.address
                            ].add(table_obj.address)

        self.data.is_tables_found = True

    def do_show_table(self, args):
        """Show an MMU table at a chosen address. Usage show_table ADDRESS [level table size]"""
        if not self.data.used_ttbcr:
            logging.error("Please set a TTBCR register to use, using set_ttbcr TTBCR")
            return

        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

        if len(args) == 3:
            valid_sizes = {0: set([0x4000]), 1: set([0x400])}
            valid_sizes[0].add(self.data.used_ttbcr.get_ptl0_user_table_size())

            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

            try:
                table_size = self.parse_int(args[2])
                if table_size not in valid_sizes[lvl]:
                    logging.warning(
                        f"Size not allowed for choosen level! Valid sizes are:{valid_sizes[lvl]}"
                    )
                    return
            except ValueError:
                logger.warning("Invalid size value")
                return
        else:
            table_size = 0x4000
            lvl = -1

        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 virtspace_short(
        self, addr, page_tables, lvl=0, prefix=0, ukx=True, cache=defaultdict(dict)
    ):
        """Recursively reconstruct virtual address space for SHORT mode"""

        virtspace = VASShort()
        data_classes = self.machine.mmu.map_datapages_entries_to_levels["global"][lvl]
        ptr_class = self.machine.mmu.map_ptr_entries_to_levels["global"][lvl]
        cache[lvl][addr] = defaultdict(portion.empty)

        if lvl == self.machine.mmu.radix_levels["global"] - 1:
            for data_class in data_classes:
                shift = self.machine.mmu.map_entries_to_shifts["global"][data_class]
                for entry_idx, entry in (
                    page_tables[lvl][addr].entries[data_class].items()
                ):
                    permissions = entry.extract_permissions()
                    kx = entry.is_kernel_executable_entry() and ukx
                    x = entry.is_executable_entry()

                    virt_addr = prefix | (entry_idx << shift)
                    virtspace[(permissions, x, kx)] |= portion.closedopen(
                        virt_addr, virt_addr + entry.size
                    )
                    cache[lvl][addr][(permissions, x, kx)] |= portion.closedopen(
                        virt_addr, virt_addr + entry.size
                    )

            return virtspace

        else:
            if ptr_class in page_tables[lvl][addr].entries:
                shift = self.machine.mmu.map_entries_to_shifts["global"][ptr_class]
                for entry_idx, entry in (
                    page_tables[lvl][addr].entries[ptr_class].items()
                ):
                    if entry.address not in page_tables[lvl + 1]:
                        continue
                    else:
                        if entry.address not in cache[lvl + 1]:
                            permissions = entry.extract_permissions()
                            kx = entry.is_kernel_executable_entry() and ukx
                            x = entry.is_executable_entry()

                            virt_addr = prefix | (entry_idx << shift)
                            low_virts = self.virtspace_short(
                                entry.address,
                                page_tables,
                                lvl + 1,
                                virt_addr,
                                kx,
                                cache=cache,
                            )
                        else:
                            low_virts = cache[lvl + 1][entry.address]

                        for perm, virts_fragment in low_virts.items():
                            virtspace[perm] |= virts_fragment
                            cache[lvl][addr][perm] |= virts_fragment

            for data_class in data_classes:
                if (
                    data_class in page_tables[lvl][addr].entries
                    and data_class is not None
                ):
                    shift = self.machine.mmu.map_entries_to_shifts["global"][data_class]
                    for entry_idx, entry in (
                        page_tables[lvl][addr].entries[data_class].items()
                    ):
                        permissions = entry.extract_permissions()
                        kx = entry.is_kernel_executable_entry() and ukx
                        x = entry.is_executable_entry()

                        virt_addr = prefix | (entry_idx << shift)
                        virtspace[(permissions, x, kx)] |= portion.closedopen(
                            virt_addr, virt_addr + entry.size
                        )
                        cache[lvl][addr][(permissions, x, kx)] |= portion.closedopen(
                            virt_addr, virt_addr + entry.size
                        )

            return virtspace

    def do_find_radix_trees(self, args):
        """Reconstruct radix trees"""
        if not self.data.is_tables_found:
            logging.info("Please, parse the memory first!")
            return

        if not self.data.is_registers_found:
            logging.info("Please find MMU related opcodes first!")
            return

        if self.data.ttbrs:
            self.data.ttbrs.clear()

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

        # Go back from PTL1 up to PTL0, the particular form of PTLn permits to find PTL0
        logging.info("Go up the paging trees starting from data pages...")
        candidates = {"kernel": []}
        modes = ["kernel"]
        if SHORT.ttbcr_n != 0:
            modes.append("user")
            candidates["user"] = []

        # TTBR0 can be used by a process which manages HW so it can point only to not in RAM pages!
        # This is a special case which it can occur only in ARM
        not_ram_pages = []
        for p in self.machine.memory.physpace["not_ram"]:
            not_ram_pages.extend([x for x in range(p.lower, p.upper, 4096)])
        not_ram_pages = set(not_ram_pages)

        for mode in modes:
            already_explored = set()
            for page_addr in tqdm(
                self.data.data_pages.union(self.data.empty_tables).union(not_ram_pages)
            ):
                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
                    candidates[mode].extend(
                        self.radix_roots_from_data_page(
                            lvl,
                            addr,
                            self.data.reverse_map_pages[mode],
                            self.data.reverse_map_tables[mode],
                        )
                    )
                    already_explored.add(derived_address)

            candidates[mode] = list(
                set(candidates[mode]).intersection(
                    self.data.page_tables[mode][0].keys()
                )
            )
            candidates[mode].sort()

        # Collect interrupt/paging opcodes
        filter_f_read = (
            lambda it: True
            if it[1]["register"] in ["DFSR", "IFSR"] and it[1]["instruction"] == "MRC"
            else False
        )
        kernel_opcodes_read = [
            x[0] for x in filter(filter_f_read, self.data.opcodes.items())
        ]
        filter_f_write = (
            lambda it: True
            if it[1]["register"] in ["TTBR0", "TTBCR"] and it[1]["instruction"] == "MCR"
            else False
        )
        kernel_opcodes_write = [
            x[0] for x in filter(filter_f_write, self.data.opcodes.items())
        ]

        logging.info("Filtering candidates...")
        filtered = {"kernel": {}, "user": {}}
        for mode in modes:
            physpace_cache = defaultdict(
                dict
            )  # We need to use different caches for user and kernel modes
            virtspace_cache = defaultdict(dict)
            for candidate in tqdm(candidates[mode]):
                consistency, pas = self.physpace(
                    candidate,
                    self.data.page_tables[mode],
                    self.data.empty_tables,
                    cache=physpace_cache,
                )

                # Ignore inconsistent radix-tress or which maps zero spaces
                if not consistency or (
                    pas.get_kernel_size() == 0 and pas.get_user_size() == 0
                ):
                    continue

                # Look for kernel trees able to map at least one interrupt/paging related opcodes
                if mode == "kernel":
                    # We check also in user pages (when ttbr1 is not used!) because user pages are always accessible also by the kernel!
                    if not any(
                        [opcode_addr in pas for opcode_addr in kernel_opcodes_read]
                    ) or (
                        SHORT.ttbcr_n != 0
                        and not any(
                            [opcode_addr in pas for opcode_addr in kernel_opcodes_write]
                        )
                    ):
                        continue

                    vas = self.virtspace_short(
                        candidate, self.data.page_tables[mode], cache=virtspace_cache
                    )

                    # At least a kernel executable page must be exist
                    for _, _, kx in vas:
                        if kx:
                            break
                    else:
                        continue

                    radix_tree = RadixTree(
                        candidate, 0, pas, vas, kernel=True, user=False
                    )
                    filtered[mode][candidate] = radix_tree

                else:
                    # No kernel pages are allowed on user radix trees!
                    if pas.get_kernel_size() != 0:
                        continue

                    # At least an executable page must exists
                    vas = self.virtspace_short(
                        candidate, self.data.page_tables[mode], cache=virtspace_cache
                    )
                    for _, x, _ in vas:
                        if x:
                            break
                    else:
                        continue

                    # Filter for at least a writable page for user (AP[1] == 1)
                    for p, _, _ in vas:
                        if p & 0b10 == 2:
                            break
                    else:
                        continue

                    radix_tree = RadixTree(
                        candidate, 0, pas, vas, kernel=False, user=True
                    )
                    filtered[mode][candidate] = radix_tree

        self.data.ttbrs = filtered
        self.data.is_radix_found = True

    def do_show_radix_trees(self, args):
        """Show radix trees found"""
        if not self.data.is_radix_found:
            logging.info("Please, find them first!")
            return

        labels = [
            "Radix address",
            "First level",
            "Kernel size (Bytes)",
            "User size (Bytes)",
            "Kernel",
        ]
        table = PrettyTable()
        table.field_names = labels
        for mode in ["kernel", "user"]:
            for ttbr in self.data.ttbrs[mode].values():
                table.add_row(
                    ttbr.entry_resume_stringified() + ["X" if mode == "kernel" else ""]
                )
        table.sortby = "Radix address"
        print(table)

do_find_radix_trees(args)

Reconstruct radix trees

Source code in mmushell/architectures/arm.py

def do_find_radix_trees(self, args):
    """Reconstruct radix trees"""
    if not self.data.is_tables_found:
        logging.info("Please, parse the memory first!")
        return

    if not self.data.is_registers_found:
        logging.info("Please find MMU related opcodes first!")
        return

    if self.data.ttbrs:
        self.data.ttbrs.clear()

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

    # Go back from PTL1 up to PTL0, the particular form of PTLn permits to find PTL0
    logging.info("Go up the paging trees starting from data pages...")
    candidates = {"kernel": []}
    modes = ["kernel"]
    if SHORT.ttbcr_n != 0:
        modes.append("user")
        candidates["user"] = []

    # TTBR0 can be used by a process which manages HW so it can point only to not in RAM pages!
    # This is a special case which it can occur only in ARM
    not_ram_pages = []
    for p in self.machine.memory.physpace["not_ram"]:
        not_ram_pages.extend([x for x in range(p.lower, p.upper, 4096)])
    not_ram_pages = set(not_ram_pages)

    for mode in modes:
        already_explored = set()
        for page_addr in tqdm(
            self.data.data_pages.union(self.data.empty_tables).union(not_ram_pages)
        ):
            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
                candidates[mode].extend(
                    self.radix_roots_from_data_page(
                        lvl,
                        addr,
                        self.data.reverse_map_pages[mode],
                        self.data.reverse_map_tables[mode],
                    )
                )
                already_explored.add(derived_address)

        candidates[mode] = list(
            set(candidates[mode]).intersection(
                self.data.page_tables[mode][0].keys()
            )
        )
        candidates[mode].sort()

    # Collect interrupt/paging opcodes
    filter_f_read = (
        lambda it: True
        if it[1]["register"] in ["DFSR", "IFSR"] and it[1]["instruction"] == "MRC"
        else False
    )
    kernel_opcodes_read = [
        x[0] for x in filter(filter_f_read, self.data.opcodes.items())
    ]
    filter_f_write = (
        lambda it: True
        if it[1]["register"] in ["TTBR0", "TTBCR"] and it[1]["instruction"] == "MCR"
        else False
    )
    kernel_opcodes_write = [
        x[0] for x in filter(filter_f_write, self.data.opcodes.items())
    ]

    logging.info("Filtering candidates...")
    filtered = {"kernel": {}, "user": {}}
    for mode in modes:
        physpace_cache = defaultdict(
            dict
        )  # We need to use different caches for user and kernel modes
        virtspace_cache = defaultdict(dict)
        for candidate in tqdm(candidates[mode]):
            consistency, pas = self.physpace(
                candidate,
                self.data.page_tables[mode],
                self.data.empty_tables,
                cache=physpace_cache,
            )

            # Ignore inconsistent radix-tress or which maps zero spaces
            if not consistency or (
                pas.get_kernel_size() == 0 and pas.get_user_size() == 0
            ):
                continue

            # Look for kernel trees able to map at least one interrupt/paging related opcodes
            if mode == "kernel":
                # We check also in user pages (when ttbr1 is not used!) because user pages are always accessible also by the kernel!
                if not any(
                    [opcode_addr in pas for opcode_addr in kernel_opcodes_read]
                ) or (
                    SHORT.ttbcr_n != 0
                    and not any(
                        [opcode_addr in pas for opcode_addr in kernel_opcodes_write]
                    )
                ):
                    continue

                vas = self.virtspace_short(
                    candidate, self.data.page_tables[mode], cache=virtspace_cache
                )

                # At least a kernel executable page must be exist
                for _, _, kx in vas:
                    if kx:
                        break
                else:
                    continue

                radix_tree = RadixTree(
                    candidate, 0, pas, vas, kernel=True, user=False
                )
                filtered[mode][candidate] = radix_tree

            else:
                # No kernel pages are allowed on user radix trees!
                if pas.get_kernel_size() != 0:
                    continue

                # At least an executable page must exists
                vas = self.virtspace_short(
                    candidate, self.data.page_tables[mode], cache=virtspace_cache
                )
                for _, x, _ in vas:
                    if x:
                        break
                else:
                    continue

                # Filter for at least a writable page for user (AP[1] == 1)
                for p, _, _ in vas:
                    if p & 0b10 == 2:
                        break
                else:
                    continue

                radix_tree = RadixTree(
                    candidate, 0, pas, vas, kernel=False, user=True
                )
                filtered[mode][candidate] = radix_tree

    self.data.ttbrs = filtered
    self.data.is_radix_found = True

do_find_registers_values(arg)

Find MMU load opcodes and execute MMU related functions inside the memory dump in order to extract MMU registers values

Source code in mmushell/architectures/arm.py

def do_find_registers_values(self, arg):
    """Find MMU load opcodes and execute MMU related functions inside the memory dump in order to extract MMU registers values"""

    if self.data.is_registers_found:
        logging.warning("Registers already searched")
        return

    logger.info("Look for opcodes related to MMU setup...")
    parallel_results = self.machine.apply_parallel(
        self.machine.mmu.PAGE_SIZE, self.machine.cpu.parse_opcodes_parallel
    )

    opcodes = {}
    logger.info("Reaggregate threads data...")
    for result in parallel_results:
        opcodes.update(result.get())

    self.data.opcodes = opcodes

    # Filter to look only for opcodes which write on MMU register only and not read from them or from other registers
    filter_f = (
        lambda it: True
        if it[1]["register"] == "TTBCR" and it[1]["instruction"] == "MCR"
        else False
    )
    mmu_wr_opcodes = {k: v for k, v in filter(filter_f, opcodes.items())}

    logging.info("Use heuristics to find function addresses...")
    logging.info("This analysis could be extremely slow!")
    self.machine.cpu.identify_functions_start(mmu_wr_opcodes)

    logging.info("Identify register values using data flow analysis...")

    # We use data flow analysis and merge the results
    dataflow_values = self.machine.cpu.find_registers_values_dataflow(
        mmu_wr_opcodes
    )

    filtered_values = defaultdict(set)
    for register, values in dataflow_values.items():
        for value in values:
            reg_obj = CPURegARM32.get_register_obj(register, value)
            if reg_obj.valid and not any(
                [
                    val_obj.is_mmu_equivalent_to(reg_obj)
                    for val_obj in filtered_values[register]
                ]
            ):
                filtered_values[register].add(reg_obj)

    # Add default values
    reg_obj = CPURegARM32.get_register_obj(
        "TTBCR", self.machine.cpu.registers_values["TTBCR"]
    )
    if reg_obj.valid and all(
        [not reg_obj.is_mmu_equivalent_to(x) for x in filtered_values["TTBCR"]]
    ):
        filtered_values["TTBCR"].add(reg_obj)

    self.data.regs_values = filtered_values
    self.data.is_registers_found = True

    # Show results
    logging.info("TTBCR values recovered:")
    for reg_obj in self.data.regs_values["TTBCR"]:
        logging.info(reg_obj)

do_find_tables(args)

Find MMU tables in memory

Source code in mmushell/architectures/arm.py

def do_find_tables(self, args):
    """Find MMU tables in memory"""
    if not self.data.used_ttbcr:
        logging.error("Please set a TTBCR register to use, using set_ttbcr TTBCR")
        return
    ttbcr = self.data.used_ttbcr

    # Delete all the previous table data
    if self.data.is_tables_found:
        for mode in ["user", "kernel"]:
            self.data.reverse_map_pages[mode].clear()
            self.data.reverse_map_tables[mode].clear()
            self.data.page_tables[mode].clear()
        self.data.empty_tables = []
        self.data.data_pages = []

    # Parse memory in chunk of 16KiB
    PTL0_USER_TABLE_SIZE = ttbcr.get_ptl0_user_table_size()
    logger.info("Look for paging tables...")
    parallel_results = self.machine.apply_parallel(
        16384,
        self.machine.mmu.parse_parallel_frame,
        ptl0_u_size=PTL0_USER_TABLE_SIZE,
    )
    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["user"][level].update(page_tables["user"][level])
            self.data.page_tables["kernel"][level].update(
                page_tables["kernel"][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)

    # Prepare fo dumplicates removing
    modes = ["kernel"]
    fps = {"kernel": []}
    if PTL0_USER_TABLE_SIZE != 16384:
        modes.append("user")
        fps["user"] = []
        self.data.page_tables["user"][1] = deepcopy(
            self.data.page_tables["kernel"][1]
        )

    # Remove all tables which point to inexistent table of lower level
    logger.info("Reduce false positives...")
    ptr_class = self.machine.mmu.map_ptr_entries_to_levels["global"][0]

    for mode in modes:
        referenced_nxt = []
        for table_addr in list(self.data.page_tables[mode][0].keys()):
            for entry_obj in (
                self.data.page_tables[mode][0][table_addr]
                .entries[ptr_class]
                .values()
            ):
                if (
                    entry_obj.address not in self.data.page_tables[mode][1]
                    and entry_obj.address not in self.data.empty_tables
                ):
                    # Remove the table
                    self.data.page_tables[mode][0].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[mode][1].keys()).difference(
            referenced_nxt
        ):
            self.data.page_tables[mode][1].pop(table_addr)

    logger.info("Fill reverse maps...")
    for mode in modes:
        for lvl in range(0, self.machine.mmu.radix_levels["global"]):
            ptr_class = self.machine.mmu.map_ptr_entries_to_levels["global"][lvl]
            page_classes = self.machine.mmu.map_datapages_entries_to_levels[
                "global"
            ][lvl]
            for table_addr, table_obj in self.data.page_tables[mode][lvl].items():
                for entry_obj in table_obj.entries[ptr_class].values():
                    self.data.reverse_map_tables[mode][lvl][entry_obj.address].add(
                        table_obj.address
                    )
                for page_class in page_classes:
                    for entry_obj in table_obj.entries[page_class].values():
                        self.data.reverse_map_pages[mode][lvl][
                            entry_obj.address
                        ].add(table_obj.address)

    self.data.is_tables_found = True

do_set_ttbcr(args)

Set the value of TTBCR register to be used

Source code in mmushell/architectures/arm.py

def do_set_ttbcr(self, args):
    """Set the value of TTBCR register to be used"""
    args = args.split()
    if len(args) == 0:
        logging.error("Please use find_tables TTBCR_VALUE")
        return

    # The shape of PTL0 tables depends on N value of TTBCR register
    try:
        ttbcr_val = self.parse_int(args[0])
        ttbcr = TTBCR(ttbcr_val)
        if not ttbcr.valid:
            raise ValueError
    except ValueError:
        logger.warning("Invalid TTBCR value")
        return

    self.data.used_ttbcr = ttbcr
    SHORT.ttbcr_n = ttbcr.n

do_show_radix_trees(args)

Show radix trees found

Source code in mmushell/architectures/arm.py

def do_show_radix_trees(self, args):
    """Show radix trees found"""
    if not self.data.is_radix_found:
        logging.info("Please, find them first!")
        return

    labels = [
        "Radix address",
        "First level",
        "Kernel size (Bytes)",
        "User size (Bytes)",
        "Kernel",
    ]
    table = PrettyTable()
    table.field_names = labels
    for mode in ["kernel", "user"]:
        for ttbr in self.data.ttbrs[mode].values():
            table.add_row(
                ttbr.entry_resume_stringified() + ["X" if mode == "kernel" else ""]
            )
    table.sortby = "Radix address"
    print(table)

do_show_registers(args)

Show registers values found

Source code in mmushell/architectures/arm.py

def do_show_registers(self, args):
    """Show registers values found"""
    if not self.data.is_registers_found:
        logging.info("Please, find them first!")
        return

    for registers in sorted(self.data.regs_values.keys()):
        for register in self.data.regs_values[registers]:
            print(register)

do_show_table(args)

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

Source code in mmushell/architectures/arm.py

def do_show_table(self, args):
    """Show an MMU table at a chosen address. Usage show_table ADDRESS [level table size]"""
    if not self.data.used_ttbcr:
        logging.error("Please set a TTBCR register to use, using set_ttbcr TTBCR")
        return

    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

    if len(args) == 3:
        valid_sizes = {0: set([0x4000]), 1: set([0x400])}
        valid_sizes[0].add(self.data.used_ttbcr.get_ptl0_user_table_size())

        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

        try:
            table_size = self.parse_int(args[2])
            if table_size not in valid_sizes[lvl]:
                logging.warning(
                    f"Size not allowed for choosen level! Valid sizes are:{valid_sizes[lvl]}"
                )
                return
        except ValueError:
            logger.warning("Invalid size value")
            return
    else:
        table_size = 0x4000
        lvl = -1

    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}")

virtspace_short(addr, page_tables, lvl=0, prefix=0, ukx=True, cache=defaultdict(dict))

Recursively reconstruct virtual address space for SHORT mode

Source code in mmushell/architectures/arm.py

def virtspace_short(
    self, addr, page_tables, lvl=0, prefix=0, ukx=True, cache=defaultdict(dict)
):
    """Recursively reconstruct virtual address space for SHORT mode"""

    virtspace = VASShort()
    data_classes = self.machine.mmu.map_datapages_entries_to_levels["global"][lvl]
    ptr_class = self.machine.mmu.map_ptr_entries_to_levels["global"][lvl]
    cache[lvl][addr] = defaultdict(portion.empty)

    if lvl == self.machine.mmu.radix_levels["global"] - 1:
        for data_class in data_classes:
            shift = self.machine.mmu.map_entries_to_shifts["global"][data_class]
            for entry_idx, entry in (
                page_tables[lvl][addr].entries[data_class].items()
            ):
                permissions = entry.extract_permissions()
                kx = entry.is_kernel_executable_entry() and ukx
                x = entry.is_executable_entry()

                virt_addr = prefix | (entry_idx << shift)
                virtspace[(permissions, x, kx)] |= portion.closedopen(
                    virt_addr, virt_addr + entry.size
                )
                cache[lvl][addr][(permissions, x, kx)] |= portion.closedopen(
                    virt_addr, virt_addr + entry.size
                )

        return virtspace

    else:
        if ptr_class in page_tables[lvl][addr].entries:
            shift = self.machine.mmu.map_entries_to_shifts["global"][ptr_class]
            for entry_idx, entry in (
                page_tables[lvl][addr].entries[ptr_class].items()
            ):
                if entry.address not in page_tables[lvl + 1]:
                    continue
                else:
                    if entry.address not in cache[lvl + 1]:
                        permissions = entry.extract_permissions()
                        kx = entry.is_kernel_executable_entry() and ukx
                        x = entry.is_executable_entry()

                        virt_addr = prefix | (entry_idx << shift)
                        low_virts = self.virtspace_short(
                            entry.address,
                            page_tables,
                            lvl + 1,
                            virt_addr,
                            kx,
                            cache=cache,
                        )
                    else:
                        low_virts = cache[lvl + 1][entry.address]

                    for perm, virts_fragment in low_virts.items():
                        virtspace[perm] |= virts_fragment
                        cache[lvl][addr][perm] |= virts_fragment

        for data_class in data_classes:
            if (
                data_class in page_tables[lvl][addr].entries
                and data_class is not None
            ):
                shift = self.machine.mmu.map_entries_to_shifts["global"][data_class]
                for entry_idx, entry in (
                    page_tables[lvl][addr].entries[data_class].items()
                ):
                    permissions = entry.extract_permissions()
                    kx = entry.is_kernel_executable_entry() and ukx
                    x = entry.is_executable_entry()

                    virt_addr = prefix | (entry_idx << shift)
                    virtspace[(permissions, x, kx)] |= portion.closedopen(
                        virt_addr, virt_addr + entry.size
                    )
                    cache[lvl][addr][(permissions, x, kx)] |= portion.closedopen(
                        virt_addr, virt_addr + entry.size
                    )

        return virtspace

MMUShellGTruth

Bases: MMUShell

Source code in mmushell/architectures/arm.py

class MMUShellGTruth(MMUShell):
    def do_show_registers_gtruth(self, args):
        """Compare TTBCR register values found with the ground truth"""
        if not self.data.is_registers_found:
            logging.info("Please, find them first!")
            return

        # Check if the last value of TTBCR was found
        all_ttbcrs = (
            {}
        )  # QEMU export TTBCR inside various registers as TTBCR, TTBCR, TCR_S etc. due to it's capability to emulate different ARM/ARM64 systems
        for reg_name, value_data in self.gtruth.items():
            if "TCR" in reg_name or "TTBCR" in reg_name:
                for value, value_info in value_data.items():
                    if value not in all_ttbcrs or (
                        value_info[1] > all_ttbcrs[value][1]
                    ):
                        all_ttbcrs[value] = (value_info[0], value_info[1])

        last_ttbcr = TTBCR(
            sorted(all_ttbcrs.keys(), key=lambda x: all_ttbcrs[x][1], reverse=True)[0]
        )

        ttbcr_fields_equals = {}
        for value_found_obj in self.data.regs_values["TTBCR"]:
            ttbcr_fields_equals[value_found_obj] = value_found_obj.count_fields_equals(
                last_ttbcr
            )
        k_sorted = sorted(
            ttbcr_fields_equals.keys(),
            key=lambda x: ttbcr_fields_equals[x],
            reverse=True,
        )
        tcr_found = k_sorted[0]
        correct_fields_found = ttbcr_fields_equals[tcr_found]

        if correct_fields_found:
            print(f"Correct TTBCR value: {last_ttbcr}, Found: {tcr_found}")
            print("TTBCR fields found:... {}/2".format(correct_fields_found))
            print("FP: {}".format(str(len(self.data.regs_values["TTBCR"]) - 1)))
        else:
            print(f"Correct TTBCR value: {last_ttbcr}")
            print("TTBCR fields found:... 0/2")
            print("FP: {}".format(str(len(self.data.regs_values["TTBCR"]))))

    def do_show_radix_trees_gtruth(self, args):
        """Compare found radix trees with the ground truth"""
        if not self.data.is_radix_found:
            logging.info("Please, find them first!")
            return

        # Collect TTBR0 and TTBR1 values from the gtruth
        ttbr0s = {}
        ttbr1s = {}
        ttbr0_phy_cache = defaultdict(dict)
        ttbr1_phy_cache = defaultdict(dict)
        virt_cache = defaultdict(dict)
        filter_f_read = (
            lambda it: True
            if it[1]["register"] in ["DFSR", "IFSR"] and it[1]["instruction"] == "MRC"
            else False
        )
        kernel_opcodes_read = [
            x[0] for x in filter(filter_f_read, self.data.opcodes.items())
        ]
        filter_f_write = (
            lambda it: True
            if it[1]["register"] in ["TTBR0", "TTBCR"] and it[1]["instruction"] == "MCR"
            else False
        )
        kernel_opcodes_write = [
            x[0] for x in filter(filter_f_write, self.data.opcodes.items())
        ]

        # User or kernel+user radix trees
        for key in ["TTBR0", "TTBR0_S", "TTBR0_EL1", "TTBR0_EL1_S"]:
            for value, data in self.gtruth.get(key, {}).items():
                ttbr = TTBR0(value)
                if any([ttbr.is_mmu_equivalent_to(x[0]) for x in ttbr0s.values()]):
                    continue

                if SHORT.ttbcr_n != 0:
                    if ttbr.address not in self.data.page_tables["user"][0]:
                        continue

                    consistency, pas = self.physpace(
                        ttbr.address,
                        self.data.page_tables["user"],
                        self.data.empty_tables,
                        cache=ttbr0_phy_cache,
                    )
                    if not consistency:
                        continue

                    if pas.get_kernel_size() != 0:
                        continue

                    virtspace = self.virtspace_short(
                        ttbr.address, self.data.page_tables["user"], cache=virt_cache
                    )
                    for _, x, _ in virtspace:
                        if x:
                            break
                    else:
                        continue

                    # Filter for at least a writable page
                    for p, _, _ in virtspace:
                        if p & 0b10 == 2:
                            break
                    else:
                        continue

                    ttbr0s[ttbr.address] = (ttbr, data)

                else:
                    if ttbr.address not in self.data.page_tables["kernel"][0]:
                        continue

                    consistency, pas = self.physpace(
                        ttbr.address,
                        self.data.page_tables["kernel"],
                        self.data.empty_tables,
                        cache=ttbr0_phy_cache,
                    )
                    if not consistency or (
                        pas.get_kernel_size() == 0 and pas.get_user_size() == 0
                    ):
                        continue

                    if not any(
                        [opcode_addr in pas for opcode_addr in kernel_opcodes_read]
                    ) or not any(
                        [opcode_addr in pas for opcode_addr in kernel_opcodes_write]
                    ):
                        continue

                    # At least a kernel executable page must be exist
                    virtspace = self.virtspace_short(
                        ttbr.address, self.data.page_tables["kernel"], cache=virt_cache
                    )
                    for _, _, kx in virtspace:
                        if kx:
                            break
                    else:
                        continue

                    ttbr0s[ttbr.address] = (ttbr, data)

        # Use only TTBR0 if TTBCR.N = 0
        if SHORT.ttbcr_n != 0:
            virt_cache = defaultdict(dict)
            for key in ["TTBR1", "TTBR1_S", "TTBR1_EL1", "TTBR1_EL1_S"]:
                for value, data in self.gtruth.get(key, {}).items():
                    ttbr = TTBR1(value)
                    if any([ttbr.is_mmu_equivalent_to(x[0]) for x in ttbr1s.values()]):
                        continue

                    if ttbr.address not in self.data.page_tables["kernel"][0]:
                        continue

                    consistency, pas = self.physpace(
                        ttbr.address,
                        self.data.page_tables["kernel"],
                        self.data.empty_tables,
                        cache=ttbr1_phy_cache,
                    )
                    if not consistency or (
                        pas.get_kernel_size() == 0 and pas.get_user_size() == 0
                    ):
                        continue

                    if not any(
                        [opcode_addr in pas for opcode_addr in kernel_opcodes_read]
                    ):
                        continue

                    # At least a kernel executable page must be exist
                    virtspace = self.virtspace_short(
                        ttbr.address, self.data.page_tables["kernel"], cache=virt_cache
                    )
                    for _, _, kx in virtspace:
                        if kx:
                            break
                    else:
                        continue

                    ttbr1s[ttbr.address] = (ttbr, data)

        # True positives, false negatives, false positives
        if SHORT.ttbcr_n == 0:
            tps = sorted(
                set(ttbr0s.keys()).intersection(set(self.data.ttbrs["kernel"].keys()))
            )
            fns = sorted(
                set(ttbr0s.keys()).difference(set(self.data.ttbrs["kernel"].keys()))
            )
            fps = sorted(
                set(self.data.ttbrs["kernel"].keys()).difference(set(ttbr0s.keys()))
            )
        else:
            tps = sorted(
                set(ttbr1s.keys()).intersection(set(self.data.ttbrs["kernel"].keys()))
            )
            fns = sorted(
                set(ttbr1s.keys()).difference(set(self.data.ttbrs["kernel"].keys()))
            )
            fps = sorted(
                set(self.data.ttbrs["kernel"].keys()).difference(set(ttbr1s.keys()))
            )
            tpsu = sorted(
                set(ttbr0s.keys()).intersection(set(self.data.ttbrs["user"].keys()))
            )
            fnsu = sorted(
                set(ttbr0s.keys()).difference(set(self.data.ttbrs["user"].keys()))
            )
            fpsu = sorted(
                set(self.data.ttbrs["user"].keys()).difference(set(ttbr0s.keys()))
            )

        # Show results
        table = PrettyTable()
        table.field_names = ["Address", "Found", "Mode", "First seen", "Last seen"]
        if SHORT.ttbcr_n == 0:
            kernel_regs = ttbr0s
            mode = "K/U"
        else:
            kernel_regs = ttbr1s
            mode = "K"

        for tp in sorted(tps):
            table.add_row(
                [hex(tp), "X", mode, kernel_regs[tp][1][0], kernel_regs[tp][1][1]]
            )

        for fn in sorted(fns):
            table.add_row(
                [hex(fn), "", mode, kernel_regs[fn][1][0], kernel_regs[fn][1][1]]
            )

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

        # User
        if SHORT.ttbcr_n != 0:
            for tp in sorted(tpsu):
                table.add_row([hex(tp), "X", "U", ttbr0s[tp][1][0], ttbr0s[tp][1][1]])

            for fn in sorted(fnsu):
                table.add_row([hex(fn), "", "U", ttbr0s[fn][1][0], ttbr0s[fn][1][1]])

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

        print(table)
        print(f"TP:{len(tps)} FN:{len(fns)} FP:{len(fps)}")
        if SHORT.ttbcr_n != 0:
            print(f"USER TP:{len(tpsu)} FN:{len(fnsu)} FP:{len(fpsu)}")

do_show_radix_trees_gtruth(args)

Compare found radix trees with the ground truth

Source code in mmushell/architectures/arm.py

def do_show_radix_trees_gtruth(self, args):
    """Compare found radix trees with the ground truth"""
    if not self.data.is_radix_found:
        logging.info("Please, find them first!")
        return

    # Collect TTBR0 and TTBR1 values from the gtruth
    ttbr0s = {}
    ttbr1s = {}
    ttbr0_phy_cache = defaultdict(dict)
    ttbr1_phy_cache = defaultdict(dict)
    virt_cache = defaultdict(dict)
    filter_f_read = (
        lambda it: True
        if it[1]["register"] in ["DFSR", "IFSR"] and it[1]["instruction"] == "MRC"
        else False
    )
    kernel_opcodes_read = [
        x[0] for x in filter(filter_f_read, self.data.opcodes.items())
    ]
    filter_f_write = (
        lambda it: True
        if it[1]["register"] in ["TTBR0", "TTBCR"] and it[1]["instruction"] == "MCR"
        else False
    )
    kernel_opcodes_write = [
        x[0] for x in filter(filter_f_write, self.data.opcodes.items())
    ]

    # User or kernel+user radix trees
    for key in ["TTBR0", "TTBR0_S", "TTBR0_EL1", "TTBR0_EL1_S"]:
        for value, data in self.gtruth.get(key, {}).items():
            ttbr = TTBR0(value)
            if any([ttbr.is_mmu_equivalent_to(x[0]) for x in ttbr0s.values()]):
                continue

            if SHORT.ttbcr_n != 0:
                if ttbr.address not in self.data.page_tables["user"][0]:
                    continue

                consistency, pas = self.physpace(
                    ttbr.address,
                    self.data.page_tables["user"],
                    self.data.empty_tables,
                    cache=ttbr0_phy_cache,
                )
                if not consistency:
                    continue

                if pas.get_kernel_size() != 0:
                    continue

                virtspace = self.virtspace_short(
                    ttbr.address, self.data.page_tables["user"], cache=virt_cache
                )
                for _, x, _ in virtspace:
                    if x:
                        break
                else:
                    continue

                # Filter for at least a writable page
                for p, _, _ in virtspace:
                    if p & 0b10 == 2:
                        break
                else:
                    continue

                ttbr0s[ttbr.address] = (ttbr, data)

            else:
                if ttbr.address not in self.data.page_tables["kernel"][0]:
                    continue

                consistency, pas = self.physpace(
                    ttbr.address,
                    self.data.page_tables["kernel"],
                    self.data.empty_tables,
                    cache=ttbr0_phy_cache,
                )
                if not consistency or (
                    pas.get_kernel_size() == 0 and pas.get_user_size() == 0
                ):
                    continue

                if not any(
                    [opcode_addr in pas for opcode_addr in kernel_opcodes_read]
                ) or not any(
                    [opcode_addr in pas for opcode_addr in kernel_opcodes_write]
                ):
                    continue

                # At least a kernel executable page must be exist
                virtspace = self.virtspace_short(
                    ttbr.address, self.data.page_tables["kernel"], cache=virt_cache
                )
                for _, _, kx in virtspace:
                    if kx:
                        break
                else:
                    continue

                ttbr0s[ttbr.address] = (ttbr, data)

    # Use only TTBR0 if TTBCR.N = 0
    if SHORT.ttbcr_n != 0:
        virt_cache = defaultdict(dict)
        for key in ["TTBR1", "TTBR1_S", "TTBR1_EL1", "TTBR1_EL1_S"]:
            for value, data in self.gtruth.get(key, {}).items():
                ttbr = TTBR1(value)
                if any([ttbr.is_mmu_equivalent_to(x[0]) for x in ttbr1s.values()]):
                    continue

                if ttbr.address not in self.data.page_tables["kernel"][0]:
                    continue

                consistency, pas = self.physpace(
                    ttbr.address,
                    self.data.page_tables["kernel"],
                    self.data.empty_tables,
                    cache=ttbr1_phy_cache,
                )
                if not consistency or (
                    pas.get_kernel_size() == 0 and pas.get_user_size() == 0
                ):
                    continue

                if not any(
                    [opcode_addr in pas for opcode_addr in kernel_opcodes_read]
                ):
                    continue

                # At least a kernel executable page must be exist
                virtspace = self.virtspace_short(
                    ttbr.address, self.data.page_tables["kernel"], cache=virt_cache
                )
                for _, _, kx in virtspace:
                    if kx:
                        break
                else:
                    continue

                ttbr1s[ttbr.address] = (ttbr, data)

    # True positives, false negatives, false positives
    if SHORT.ttbcr_n == 0:
        tps = sorted(
            set(ttbr0s.keys()).intersection(set(self.data.ttbrs["kernel"].keys()))
        )
        fns = sorted(
            set(ttbr0s.keys()).difference(set(self.data.ttbrs["kernel"].keys()))
        )
        fps = sorted(
            set(self.data.ttbrs["kernel"].keys()).difference(set(ttbr0s.keys()))
        )
    else:
        tps = sorted(
            set(ttbr1s.keys()).intersection(set(self.data.ttbrs["kernel"].keys()))
        )
        fns = sorted(
            set(ttbr1s.keys()).difference(set(self.data.ttbrs["kernel"].keys()))
        )
        fps = sorted(
            set(self.data.ttbrs["kernel"].keys()).difference(set(ttbr1s.keys()))
        )
        tpsu = sorted(
            set(ttbr0s.keys()).intersection(set(self.data.ttbrs["user"].keys()))
        )
        fnsu = sorted(
            set(ttbr0s.keys()).difference(set(self.data.ttbrs["user"].keys()))
        )
        fpsu = sorted(
            set(self.data.ttbrs["user"].keys()).difference(set(ttbr0s.keys()))
        )

    # Show results
    table = PrettyTable()
    table.field_names = ["Address", "Found", "Mode", "First seen", "Last seen"]
    if SHORT.ttbcr_n == 0:
        kernel_regs = ttbr0s
        mode = "K/U"
    else:
        kernel_regs = ttbr1s
        mode = "K"

    for tp in sorted(tps):
        table.add_row(
            [hex(tp), "X", mode, kernel_regs[tp][1][0], kernel_regs[tp][1][1]]
        )

    for fn in sorted(fns):
        table.add_row(
            [hex(fn), "", mode, kernel_regs[fn][1][0], kernel_regs[fn][1][1]]
        )

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

    # User
    if SHORT.ttbcr_n != 0:
        for tp in sorted(tpsu):
            table.add_row([hex(tp), "X", "U", ttbr0s[tp][1][0], ttbr0s[tp][1][1]])

        for fn in sorted(fnsu):
            table.add_row([hex(fn), "", "U", ttbr0s[fn][1][0], ttbr0s[fn][1][1]])

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

    print(table)
    print(f"TP:{len(tps)} FN:{len(fns)} FP:{len(fps)}")
    if SHORT.ttbcr_n != 0:
        print(f"USER TP:{len(tpsu)} FN:{len(fnsu)} FP:{len(fpsu)}")

do_show_registers_gtruth(args)

Compare TTBCR register values found with the ground truth

Source code in mmushell/architectures/arm.py

def do_show_registers_gtruth(self, args):
    """Compare TTBCR register values found with the ground truth"""
    if not self.data.is_registers_found:
        logging.info("Please, find them first!")
        return

    # Check if the last value of TTBCR was found
    all_ttbcrs = (
        {}
    )  # QEMU export TTBCR inside various registers as TTBCR, TTBCR, TCR_S etc. due to it's capability to emulate different ARM/ARM64 systems
    for reg_name, value_data in self.gtruth.items():
        if "TCR" in reg_name or "TTBCR" in reg_name:
            for value, value_info in value_data.items():
                if value not in all_ttbcrs or (
                    value_info[1] > all_ttbcrs[value][1]
                ):
                    all_ttbcrs[value] = (value_info[0], value_info[1])

    last_ttbcr = TTBCR(
        sorted(all_ttbcrs.keys(), key=lambda x: all_ttbcrs[x][1], reverse=True)[0]
    )

    ttbcr_fields_equals = {}
    for value_found_obj in self.data.regs_values["TTBCR"]:
        ttbcr_fields_equals[value_found_obj] = value_found_obj.count_fields_equals(
            last_ttbcr
        )
    k_sorted = sorted(
        ttbcr_fields_equals.keys(),
        key=lambda x: ttbcr_fields_equals[x],
        reverse=True,
    )
    tcr_found = k_sorted[0]
    correct_fields_found = ttbcr_fields_equals[tcr_found]

    if correct_fields_found:
        print(f"Correct TTBCR value: {last_ttbcr}, Found: {tcr_found}")
        print("TTBCR fields found:... {}/2".format(correct_fields_found))
        print("FP: {}".format(str(len(self.data.regs_values["TTBCR"]) - 1)))
    else:
        print(f"Correct TTBCR value: {last_ttbcr}")
        print("TTBCR fields found:... 0/2")
        print("FP: {}".format(str(len(self.data.regs_values["TTBCR"]))))