AArch64

AArch64

MMUShell

Bases: MMUShell

Source code in mmushell/architectures/aarch64.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": defaultdict(dict), "kernel": defaultdict(dict)},
                data_pages={"user": [], "kernel": []},
                empty_tables={"user": [], "kernel": []},
                reverse_map_tables={"user": None, "kernel": None},
                reverse_map_pages={"user": None, "kernel": None},
                used_tcr=None,
                ttbrs=defaultdict(dict),
            )

    def reload_data_from_file(self, data_filename):
        super(MMUShell, self).reload_data_from_file(data_filename)

        # Reload TCR data and radix tree shape
        LONG.tcr = self.data.used_tcr
        self.do_set_tcr(str(LONG.tcr.value))

    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(
            65536, 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"] == "TCR_EL1" and it[1]["instruction"] == "MSR"
            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 = CPURegAArch64.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)

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

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

    def do_show_registers(self, args):
        """Show TCR 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_tcr(self, args):
        """Set TCR to be used"""
        args = args.split()
        if len(args) == 0:
            logging.error("Please use find_tables TCR_VALUE")
            return

        try:
            tcr_val = self.parse_int(args[0])
            tcr = TCR_EL1(tcr_val)
            if not tcr.valid:
                raise ValueError
        except ValueError:
            logger.warning("Invalid TCR value")
            return

        self.data.used_tcr = tcr

        # Set all MMU parameters
        LONG.tcr = tcr
        LONG.radix_levels = {}
        trees_struct = tcr.get_trees_struct()

        for mode in ["user", "kernel"]:
            granule = trees_struct[mode]["granule"]
            total_levels = trees_struct[mode]["total_levels"]
            top_table_size = trees_struct[mode]["top_table_size"]

            LONG.radix_levels[mode] = total_levels
            LONG.map_level_to_table_size[mode] = [top_table_size] + (
                [granule] * (total_levels - 1)
            )
            LONG.map_reserved_entries_to_levels[mode] = [
                [] for i in range(total_levels - 1)
            ] + [[ReservedEntry]]

            if granule == 4096:
                if total_levels == 1:
                    LONG.map_datapages_entries_to_levels[mode] = [PTPAGE_4KB]
                    LONG.map_ptr_entries_to_levels[mode] = [None]
                    LONG.map_entries_to_shifts[mode] = {PTPAGE_4KB: 12}
                elif total_levels == 2:
                    LONG.map_datapages_entries_to_levels[mode] = [
                        [PTBLOCK_L2_4KB],
                        [PTPAGE_4KB],
                    ]
                    LONG.map_ptr_entries_to_levels[mode] = [PTP_4KB_L0, None]
                    LONG.map_entries_to_shifts[mode] = {
                        PTP_4KB_L0: 21,
                        PTPAGE_4KB: 12,
                        PTBLOCK_L2_4KB: 21,
                    }
                elif total_levels == 3:
                    LONG.map_datapages_entries_to_levels[mode] = [
                        [PTBLOCK_L1_4KB],
                        [PTBLOCK_L2_4KB],
                        [PTPAGE_4KB],
                    ]
                    LONG.map_ptr_entries_to_levels[mode] = [
                        PTP_4KB_L0,
                        PTP_4KB_L1,
                        None,
                    ]
                    LONG.map_entries_to_shifts[mode] = {
                        PTP_4KB_L0: 30,
                        PTP_4KB_L1: 21,
                        PTPAGE_4KB: 12,
                        PTBLOCK_L2_4KB: 21,
                        PTBLOCK_L1_4KB: 30,
                    }
                else:
                    LONG.map_datapages_entries_to_levels[mode] = [
                        [None],
                        [PTBLOCK_L1_4KB],
                        [PTBLOCK_L2_4KB],
                        [PTPAGE_4KB],
                    ]
                    LONG.map_ptr_entries_to_levels[mode] = [
                        PTP_4KB_L0,
                        PTP_4KB_L1,
                        PTP_4KB_L2,
                        None,
                    ]
                    LONG.map_entries_to_shifts[mode] = {
                        PTP_4KB_L0: 39,
                        PTP_4KB_L1: 30,
                        PTP_4KB_L2: 21,
                        PTPAGE_4KB: 12,
                        PTBLOCK_L2_4KB: 21,
                        PTBLOCK_L1_4KB: 30,
                    }

            elif granule == 16384:
                if total_levels == 1:
                    LONG.map_datapages_entries_to_levels[mode] = [PTPAGE_16KB]
                    LONG.map_ptr_entries_to_levels[mode] = [None]
                    LONG.map_entries_to_shifts[mode] = {PTPAGE_16KB: 14}
                elif total_levels == 2:
                    LONG.map_datapages_entries_to_levels[mode] = [
                        [PTBLOCK_L2_16KB],
                        [PTPAGE_16KB],
                    ]
                    LONG.map_ptr_entries_to_levels[mode] = [PTP_16KB_L0, None]
                    LONG.map_entries_to_shifts[mode] = {
                        PTP_16KB_L0: 25,
                        PTPAGE_16KB: 14,
                        PTBLOCK_L2_16KB: 25,
                    }
                elif total_levels == 3:
                    LONG.map_datapages_entries_to_levels[mode] = [
                        [None],
                        [PTBLOCK_L2_16KB],
                        [PTPAGE_16KB],
                    ]
                    LONG.map_ptr_entries_to_levels[mode] = [
                        PTP_16KB_L0,
                        PTP_16KB_L1,
                        None,
                    ]
                    LONG.map_entries_to_shifts[mode] = {
                        PTP_16KB_L0: 36,
                        PTP_16KB_L1: 25,
                        PTPAGE_16KB: 14,
                        PTBLOCK_L2_16KB: 25,
                    }
                else:
                    LONG.map_datapages_entries_to_levels[mode] = [
                        [None],
                        [None],
                        [PTBLOCK_L2_16KB],
                        [PTPAGE_16KB],
                    ]
                    LONG.map_ptr_entries_to_levels[mode] = [
                        PTP_16KB_L0,
                        PTP_16KB_L2,
                        None,
                    ]
                    LONG.map_entries_to_shifts[mode] = {
                        PTP_16KB_L0: 47,
                        PTP_16KB_L1: 36,
                        PTP_16KB_L2: 25,
                        PTPAGE_16KB: 14,
                        PTBLOCK_L2_16KB: 25,
                    }

            else:
                if total_levels == 1:
                    LONG.map_datapages_entries_to_levels[mode] = [PTPAGE_64KB]
                    LONG.map_ptr_entries_to_levels[mode] = [None]
                    LONG.map_entries_to_shifts[mode] = {
                        PTPAGE_64KB: 16,
                        PTBLOCK_L2_64KB: 29,
                    }
                elif total_levels == 2:
                    LONG.map_datapages_entries_to_levels[mode] = [
                        [PTBLOCK_L2_64KB],
                        [PTPAGE_64KB],
                    ]
                    LONG.map_ptr_entries_to_levels[mode] = [PTP_64KB_L0, None]
                    LONG.map_entries_to_shifts[mode] = {
                        PTP_64KB_L0: 29,
                        PTPAGE_64KB: 16,
                        PTBLOCK_L2_16KB: 29,
                    }
                else:
                    LONG.map_datapages_entries_to_levels[mode] = [
                        [None],
                        [PTBLOCK_L2_64KB],
                        [PTPAGE_64KB],
                    ]
                    LONG.map_ptr_entries_to_levels[mode] = [
                        PTP_64KB_L0,
                        PTP_64KB_L1,
                        None,
                    ]
                    LONG.map_entries_to_shifts[mode] = {
                        PTP_64KB_L0: 42,
                        PTP_64KB_L1: 29,
                        PTPAGE_64KB: 16,
                        PTBLOCK_L2_64KB: 29,
                    }

    def do_find_tables(self, args):
        """Find MMU tables in memory"""
        if not self.data.used_tcr:
            logging.error("Please set a TCR register to use, using set_tcr TCR")
            return
        tcr = self.data.used_tcr

        # Delete all the previous table data
        if self.data.is_tables_found:
            self.data.page_tables = {
                "user": defaultdict(dict),
                "kernel": defaultdict(dict),
            }
            self.data.data_pages = {"user": [], "kernel": []}
            self.data.empty_tables = {"user": [], "kernel": []}
            self.data.reverse_map_tables = {}
            self.data.reverse_map_pages = {}

        # WORKAROUND: initialize here because unpickable!
        self.data.reverse_map_pages = {
            "kernel": defaultdict(_dummy_f),
            "user": defaultdict(_dummy_f),
        }
        self.data.reverse_map_tables = {
            "kernel": defaultdict(_dummy_f),
            "user": defaultdict(_dummy_f),
        }

        # Parse memory in chunk of 64KiB
        logger.info("Look for paging tables...")
        parallel_results = self.machine.apply_parallel(
            65536, self.machine.mmu.parse_parallel_frame, tcr=tcr
        )
        logger.info("Reaggregate threads data...")
        for result in parallel_results:
            page_tables, data_pages, empty_tables = result.get()

            for mode in ["user", "kernel"]:
                for level in range(self.machine.mmu.radix_levels[mode]):
                    self.data.page_tables[mode][level].update(page_tables[mode][level])

                self.data.data_pages[mode].extend(data_pages[mode])
                self.data.empty_tables[mode].extend(empty_tables[mode])

        for mode in ["user", "kernel"]:
            self.data.data_pages[mode] = set(self.data.data_pages[mode])
            self.data.empty_tables[mode] = set(self.data.empty_tables[mode])

        # Remove all tables which point to inexistent table of lower level
        logger.info("Reduce false positives...")
        for mode in ["user", "kernel"]:
            for lvl in range(self.machine.mmu.radix_levels[mode] - 1):
                ptr_class = self.machine.mmu.map_ptr_entries_to_levels[mode][lvl]
                referenced_nxt = []
                for table_addr in list(self.data.page_tables[mode][lvl].keys()):
                    for entry_obj in (
                        self.data.page_tables[mode][lvl][table_addr]
                        .entries[ptr_class]
                        .values()
                    ):
                        if (
                            entry_obj.address
                            not in self.data.page_tables[mode][lvl + 1]
                            and entry_obj.address not in self.data.empty_tables[mode]
                        ):
                            # Remove the table
                            self.data.page_tables[mode][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[mode][lvl + 1].keys()
                ).difference(referenced_nxt):
                    self.data.page_tables[mode][lvl + 1].pop(table_addr)

        logger.info("Fill reverse maps...")
        for mode in ["user", "kernel"]:
            for lvl in range(0, self.machine.mmu.radix_levels[mode]):
                ptr_class = self.machine.mmu.map_ptr_entries_to_levels[mode][lvl]
                page_classes = self.machine.mmu.map_datapages_entries_to_levels[mode][
                    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)

        # If kernel and user space use the same configuration, copy kernel data to user
        trees_struct = tcr.get_trees_struct()
        if trees_struct["kernel"] == trees_struct["user"]:
            self.data.page_tables["user"] = self.data.page_tables["kernel"]
            self.data.reverse_map_pages["user"] = self.data.reverse_map_pages["kernel"]
            self.data.reverse_map_tables["user"] = self.data.reverse_map_tables[
                "kernel"
            ]
            self.data.data_pages["user"] = self.data.data_pages["kernel"]
            self.data.empty_tables["user"] = self.data.empty_tables["kernel"]

        self.data.is_tables_found = True

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

        args = args.split()
        if len(args) < 2:
            logger.warning("Missing argument")
            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

        args[1] = args[1].lower()
        if args[1] not in ["kernel", "user"]:
            logger.warning("Mode must be kernel or user")
            return
        mode = args[1]

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

            trees_struct = LONG.tcr.get_trees_struct()
            valid_sizes = {"user": defaultdict(set), "kernel": defaultdict(set)}
            valid_sizes["kernel"][0].add(trees_struct["kernel"]["top_table_size"])
            valid_sizes["user"][0].add(trees_struct["user"]["top_table_size"])
            for i in range(1, trees_struct["kernel"]["total_levels"]):
                valid_sizes["kernel"][i].add(trees_struct["kernel"]["granule"])
            for i in range(1, trees_struct["user"]["total_levels"]):
                valid_sizes["user"][i].add(trees_struct["user"]["granule"])

            try:
                table_size = self.parse_int(args[3])
                if table_size not in valid_sizes[mode][lvl]:
                    logging.warning(
                        f"Size not allowed for choosen level! Valid sizes are:{valid_sizes[mode][lvl]}"
                    )
                    return
            except ValueError:
                logger.warning("Invalid size value")
                return
        else:
            table_size = 0x10000
            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, mode=mode
        )
        print(table_obj)
        print(f"Invalid entries: {invalids} Table levels: {pt_classes}")

    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]) and not len(
            self.data.page_tables["user"][0]
        ):
            logger.warning("OOPS... no tables in first level... Wrong MMU mode?")
            return

        ttbrs_candidates = {"kernel": [], "user": []}
        trees_struct = self.machine.mmu.tcr.get_trees_struct()

        # Collect opcodes
        opcode_classes = defaultdict(list)
        for opcode_addr, opcode_data in self.data.opcodes.items():
            opcode_classes[
                (opcode_data["instruction"], opcode_data["register"])
            ].append(opcode_addr)

        # Find all TTBR1_EL1 which contain interrupt related opcodes
        logging.info("Find TTBR1_EL1 candidates...")
        int_opcode_addrs = (
            opcode_classes[("MRS", "ESR_EL1")]
            + opcode_classes[("MRS", "FAR_EL1")]
            + opcode_classes[("MRS", "ELR_EL1")]
        )
        already_explored = set()
        for opcode_addr in int_opcode_addrs:
            derived_addresses = self.machine.mmu.derive_page_address(
                opcode_addr, mode="kernel"
            )
            for derived_address in derived_addresses:
                if derived_address in already_explored:
                    continue

                lvl, addr = derived_address
                ttbrs_candidates["kernel"].extend(
                    self.radix_roots_from_data_page(
                        lvl,
                        addr,
                        self.data.reverse_map_pages["kernel"],
                        self.data.reverse_map_tables["kernel"],
                    )
                )
                already_explored.add(derived_address)

        ttbrs_candidates["kernel"] = list(
            set(ttbrs_candidates["kernel"]).intersection(
                self.data.page_tables["kernel"][0].keys()
            )
        )

        # Filter kernel candidates for ERET and write on MMU registers
        logger.info("Filtering TTBR1_EL1 candidates...")
        mmu_w_opcode_addrs = (
            opcode_classes[("MSR", "TCR_EL1")] + opcode_classes[("MSR", "TTBR0_EL1")]
        )
        phy_cache = defaultdict(dict)
        ttbrs_filtered = {"kernel": {}, "user": {}}
        virt_cache = defaultdict(dict)
        for candidate in tqdm(ttbrs_candidates["kernel"]):
            # Calculate physpace and discard empty ones
            consistency, pas = self.physpace(
                candidate,
                self.data.page_tables["kernel"],
                self.data.empty_tables["kernel"],
                mode="kernel",
                hierarchical=True,
                cache=phy_cache,
            )

            # Discard inconsistent one
            if not consistency:
                continue

            # WARNING! We cannot filter for user_size = 0 due to TCR_EL1.E0PD1 !
            # Check if at least one MMU opcode in physical address space
            for opcode_addr in mmu_w_opcode_addrs:
                if pas.is_in_kernel_space(opcode_addr):
                    break
            else:
                continue

            # Check if at least one ERET opcode in physical address space
            for opcode_addr in opcode_classes[("ERET", "")]:
                if pas.is_in_kernel_space(opcode_addr):
                    break
            else:
                continue

            # At least a page must be writable by the kernel and not by user
            for perms in pas.space:
                if perms[1] and not perms[4]:
                    break
            else:
                continue

            vas = self.virtspace(
                candidate, mode="kernel", hierarchical=True, cache=virt_cache
            )
            radix_tree = RadixTree(
                candidate,
                trees_struct["kernel"]["total_levels"],
                pas,
                vas,
                kernel=True,
                user=False,
            )
            ttbrs_filtered["kernel"][candidate] = radix_tree

        # Find all TTBR0_EL1 which contain at least one RET instruction
        already_explored = set()
        virt_cache.clear()
        logging.info("Find TTBR0_EL1 candidates...")
        for opcode_addr in opcode_classes[("RET", "")]:
            derived_addresses = self.machine.mmu.derive_page_address(
                opcode_addr, mode="user"
            )
            for derived_address in derived_addresses:
                if derived_address in already_explored:
                    continue

                lvl, addr = derived_address
                ttbrs_candidates["user"].extend(
                    self.radix_roots_from_data_page(
                        lvl,
                        addr,
                        self.data.reverse_map_pages["user"],
                        self.data.reverse_map_tables["user"],
                    )
                )
                already_explored.add(derived_address)

        ttbrs_candidates["user"] = list(
            set(ttbrs_candidates["user"]).intersection(
                self.data.page_tables["user"][0].keys()
            )
        )

        logger.info("Filtering TTBR0_EL1 candidates...")
        phy_cache = defaultdict(dict)
        for candidate in tqdm(ttbrs_candidates["user"]):
            # Calculate physpace and discard empty ones
            consistency, pas = self.physpace(
                candidate,
                self.data.page_tables["user"],
                self.data.empty_tables["user"],
                mode="user",
                hierarchical=True,
                cache=phy_cache,
            )

            # Discard inconsistent one
            if not consistency:
                continue

            # At least a page must be R or W in usermode
            for perms in pas.space:
                if perms[3] or perms[4]:
                    break
            else:
                continue

            # Check if at least one BLR opcode in physical address space
            for opcode_addr in opcode_classes[("BLR", "")]:
                if opcode_addr in pas:
                    break
            else:
                continue

            vas = self.virtspace(
                candidate, mode="user", hierarchical=True, cache=virt_cache
            )
            radix_tree = RadixTree(
                candidate,
                trees_struct["user"]["total_levels"],
                pas,
                vas,
                kernel=False,
                user=True,
            )
            ttbrs_filtered["user"][candidate] = radix_tree

        self.data.ttbrs = 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",
            "Total levels",
            "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/aarch64.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]) and not len(
        self.data.page_tables["user"][0]
    ):
        logger.warning("OOPS... no tables in first level... Wrong MMU mode?")
        return

    ttbrs_candidates = {"kernel": [], "user": []}
    trees_struct = self.machine.mmu.tcr.get_trees_struct()

    # Collect opcodes
    opcode_classes = defaultdict(list)
    for opcode_addr, opcode_data in self.data.opcodes.items():
        opcode_classes[
            (opcode_data["instruction"], opcode_data["register"])
        ].append(opcode_addr)

    # Find all TTBR1_EL1 which contain interrupt related opcodes
    logging.info("Find TTBR1_EL1 candidates...")
    int_opcode_addrs = (
        opcode_classes[("MRS", "ESR_EL1")]
        + opcode_classes[("MRS", "FAR_EL1")]
        + opcode_classes[("MRS", "ELR_EL1")]
    )
    already_explored = set()
    for opcode_addr in int_opcode_addrs:
        derived_addresses = self.machine.mmu.derive_page_address(
            opcode_addr, mode="kernel"
        )
        for derived_address in derived_addresses:
            if derived_address in already_explored:
                continue

            lvl, addr = derived_address
            ttbrs_candidates["kernel"].extend(
                self.radix_roots_from_data_page(
                    lvl,
                    addr,
                    self.data.reverse_map_pages["kernel"],
                    self.data.reverse_map_tables["kernel"],
                )
            )
            already_explored.add(derived_address)

    ttbrs_candidates["kernel"] = list(
        set(ttbrs_candidates["kernel"]).intersection(
            self.data.page_tables["kernel"][0].keys()
        )
    )

    # Filter kernel candidates for ERET and write on MMU registers
    logger.info("Filtering TTBR1_EL1 candidates...")
    mmu_w_opcode_addrs = (
        opcode_classes[("MSR", "TCR_EL1")] + opcode_classes[("MSR", "TTBR0_EL1")]
    )
    phy_cache = defaultdict(dict)
    ttbrs_filtered = {"kernel": {}, "user": {}}
    virt_cache = defaultdict(dict)
    for candidate in tqdm(ttbrs_candidates["kernel"]):
        # Calculate physpace and discard empty ones
        consistency, pas = self.physpace(
            candidate,
            self.data.page_tables["kernel"],
            self.data.empty_tables["kernel"],
            mode="kernel",
            hierarchical=True,
            cache=phy_cache,
        )

        # Discard inconsistent one
        if not consistency:
            continue

        # WARNING! We cannot filter for user_size = 0 due to TCR_EL1.E0PD1 !
        # Check if at least one MMU opcode in physical address space
        for opcode_addr in mmu_w_opcode_addrs:
            if pas.is_in_kernel_space(opcode_addr):
                break
        else:
            continue

        # Check if at least one ERET opcode in physical address space
        for opcode_addr in opcode_classes[("ERET", "")]:
            if pas.is_in_kernel_space(opcode_addr):
                break
        else:
            continue

        # At least a page must be writable by the kernel and not by user
        for perms in pas.space:
            if perms[1] and not perms[4]:
                break
        else:
            continue

        vas = self.virtspace(
            candidate, mode="kernel", hierarchical=True, cache=virt_cache
        )
        radix_tree = RadixTree(
            candidate,
            trees_struct["kernel"]["total_levels"],
            pas,
            vas,
            kernel=True,
            user=False,
        )
        ttbrs_filtered["kernel"][candidate] = radix_tree

    # Find all TTBR0_EL1 which contain at least one RET instruction
    already_explored = set()
    virt_cache.clear()
    logging.info("Find TTBR0_EL1 candidates...")
    for opcode_addr in opcode_classes[("RET", "")]:
        derived_addresses = self.machine.mmu.derive_page_address(
            opcode_addr, mode="user"
        )
        for derived_address in derived_addresses:
            if derived_address in already_explored:
                continue

            lvl, addr = derived_address
            ttbrs_candidates["user"].extend(
                self.radix_roots_from_data_page(
                    lvl,
                    addr,
                    self.data.reverse_map_pages["user"],
                    self.data.reverse_map_tables["user"],
                )
            )
            already_explored.add(derived_address)

    ttbrs_candidates["user"] = list(
        set(ttbrs_candidates["user"]).intersection(
            self.data.page_tables["user"][0].keys()
        )
    )

    logger.info("Filtering TTBR0_EL1 candidates...")
    phy_cache = defaultdict(dict)
    for candidate in tqdm(ttbrs_candidates["user"]):
        # Calculate physpace and discard empty ones
        consistency, pas = self.physpace(
            candidate,
            self.data.page_tables["user"],
            self.data.empty_tables["user"],
            mode="user",
            hierarchical=True,
            cache=phy_cache,
        )

        # Discard inconsistent one
        if not consistency:
            continue

        # At least a page must be R or W in usermode
        for perms in pas.space:
            if perms[3] or perms[4]:
                break
        else:
            continue

        # Check if at least one BLR opcode in physical address space
        for opcode_addr in opcode_classes[("BLR", "")]:
            if opcode_addr in pas:
                break
        else:
            continue

        vas = self.virtspace(
            candidate, mode="user", hierarchical=True, cache=virt_cache
        )
        radix_tree = RadixTree(
            candidate,
            trees_struct["user"]["total_levels"],
            pas,
            vas,
            kernel=False,
            user=True,
        )
        ttbrs_filtered["user"][candidate] = radix_tree

    self.data.ttbrs = 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/aarch64.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(
        65536, 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"] == "TCR_EL1" and it[1]["instruction"] == "MSR"
        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 = CPURegAArch64.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)

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

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

do_find_tables(args)

Find MMU tables in memory

Source code in mmushell/architectures/aarch64.py

def do_find_tables(self, args):
    """Find MMU tables in memory"""
    if not self.data.used_tcr:
        logging.error("Please set a TCR register to use, using set_tcr TCR")
        return
    tcr = self.data.used_tcr

    # Delete all the previous table data
    if self.data.is_tables_found:
        self.data.page_tables = {
            "user": defaultdict(dict),
            "kernel": defaultdict(dict),
        }
        self.data.data_pages = {"user": [], "kernel": []}
        self.data.empty_tables = {"user": [], "kernel": []}
        self.data.reverse_map_tables = {}
        self.data.reverse_map_pages = {}

    # WORKAROUND: initialize here because unpickable!
    self.data.reverse_map_pages = {
        "kernel": defaultdict(_dummy_f),
        "user": defaultdict(_dummy_f),
    }
    self.data.reverse_map_tables = {
        "kernel": defaultdict(_dummy_f),
        "user": defaultdict(_dummy_f),
    }

    # Parse memory in chunk of 64KiB
    logger.info("Look for paging tables...")
    parallel_results = self.machine.apply_parallel(
        65536, self.machine.mmu.parse_parallel_frame, tcr=tcr
    )
    logger.info("Reaggregate threads data...")
    for result in parallel_results:
        page_tables, data_pages, empty_tables = result.get()

        for mode in ["user", "kernel"]:
            for level in range(self.machine.mmu.radix_levels[mode]):
                self.data.page_tables[mode][level].update(page_tables[mode][level])

            self.data.data_pages[mode].extend(data_pages[mode])
            self.data.empty_tables[mode].extend(empty_tables[mode])

    for mode in ["user", "kernel"]:
        self.data.data_pages[mode] = set(self.data.data_pages[mode])
        self.data.empty_tables[mode] = set(self.data.empty_tables[mode])

    # Remove all tables which point to inexistent table of lower level
    logger.info("Reduce false positives...")
    for mode in ["user", "kernel"]:
        for lvl in range(self.machine.mmu.radix_levels[mode] - 1):
            ptr_class = self.machine.mmu.map_ptr_entries_to_levels[mode][lvl]
            referenced_nxt = []
            for table_addr in list(self.data.page_tables[mode][lvl].keys()):
                for entry_obj in (
                    self.data.page_tables[mode][lvl][table_addr]
                    .entries[ptr_class]
                    .values()
                ):
                    if (
                        entry_obj.address
                        not in self.data.page_tables[mode][lvl + 1]
                        and entry_obj.address not in self.data.empty_tables[mode]
                    ):
                        # Remove the table
                        self.data.page_tables[mode][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[mode][lvl + 1].keys()
            ).difference(referenced_nxt):
                self.data.page_tables[mode][lvl + 1].pop(table_addr)

    logger.info("Fill reverse maps...")
    for mode in ["user", "kernel"]:
        for lvl in range(0, self.machine.mmu.radix_levels[mode]):
            ptr_class = self.machine.mmu.map_ptr_entries_to_levels[mode][lvl]
            page_classes = self.machine.mmu.map_datapages_entries_to_levels[mode][
                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)

    # If kernel and user space use the same configuration, copy kernel data to user
    trees_struct = tcr.get_trees_struct()
    if trees_struct["kernel"] == trees_struct["user"]:
        self.data.page_tables["user"] = self.data.page_tables["kernel"]
        self.data.reverse_map_pages["user"] = self.data.reverse_map_pages["kernel"]
        self.data.reverse_map_tables["user"] = self.data.reverse_map_tables[
            "kernel"
        ]
        self.data.data_pages["user"] = self.data.data_pages["kernel"]
        self.data.empty_tables["user"] = self.data.empty_tables["kernel"]

    self.data.is_tables_found = True

do_set_tcr(args)

Set TCR to be used

Source code in mmushell/architectures/aarch64.py

def do_set_tcr(self, args):
    """Set TCR to be used"""
    args = args.split()
    if len(args) == 0:
        logging.error("Please use find_tables TCR_VALUE")
        return

    try:
        tcr_val = self.parse_int(args[0])
        tcr = TCR_EL1(tcr_val)
        if not tcr.valid:
            raise ValueError
    except ValueError:
        logger.warning("Invalid TCR value")
        return

    self.data.used_tcr = tcr

    # Set all MMU parameters
    LONG.tcr = tcr
    LONG.radix_levels = {}
    trees_struct = tcr.get_trees_struct()

    for mode in ["user", "kernel"]:
        granule = trees_struct[mode]["granule"]
        total_levels = trees_struct[mode]["total_levels"]
        top_table_size = trees_struct[mode]["top_table_size"]

        LONG.radix_levels[mode] = total_levels
        LONG.map_level_to_table_size[mode] = [top_table_size] + (
            [granule] * (total_levels - 1)
        )
        LONG.map_reserved_entries_to_levels[mode] = [
            [] for i in range(total_levels - 1)
        ] + [[ReservedEntry]]

        if granule == 4096:
            if total_levels == 1:
                LONG.map_datapages_entries_to_levels[mode] = [PTPAGE_4KB]
                LONG.map_ptr_entries_to_levels[mode] = [None]
                LONG.map_entries_to_shifts[mode] = {PTPAGE_4KB: 12}
            elif total_levels == 2:
                LONG.map_datapages_entries_to_levels[mode] = [
                    [PTBLOCK_L2_4KB],
                    [PTPAGE_4KB],
                ]
                LONG.map_ptr_entries_to_levels[mode] = [PTP_4KB_L0, None]
                LONG.map_entries_to_shifts[mode] = {
                    PTP_4KB_L0: 21,
                    PTPAGE_4KB: 12,
                    PTBLOCK_L2_4KB: 21,
                }
            elif total_levels == 3:
                LONG.map_datapages_entries_to_levels[mode] = [
                    [PTBLOCK_L1_4KB],
                    [PTBLOCK_L2_4KB],
                    [PTPAGE_4KB],
                ]
                LONG.map_ptr_entries_to_levels[mode] = [
                    PTP_4KB_L0,
                    PTP_4KB_L1,
                    None,
                ]
                LONG.map_entries_to_shifts[mode] = {
                    PTP_4KB_L0: 30,
                    PTP_4KB_L1: 21,
                    PTPAGE_4KB: 12,
                    PTBLOCK_L2_4KB: 21,
                    PTBLOCK_L1_4KB: 30,
                }
            else:
                LONG.map_datapages_entries_to_levels[mode] = [
                    [None],
                    [PTBLOCK_L1_4KB],
                    [PTBLOCK_L2_4KB],
                    [PTPAGE_4KB],
                ]
                LONG.map_ptr_entries_to_levels[mode] = [
                    PTP_4KB_L0,
                    PTP_4KB_L1,
                    PTP_4KB_L2,
                    None,
                ]
                LONG.map_entries_to_shifts[mode] = {
                    PTP_4KB_L0: 39,
                    PTP_4KB_L1: 30,
                    PTP_4KB_L2: 21,
                    PTPAGE_4KB: 12,
                    PTBLOCK_L2_4KB: 21,
                    PTBLOCK_L1_4KB: 30,
                }

        elif granule == 16384:
            if total_levels == 1:
                LONG.map_datapages_entries_to_levels[mode] = [PTPAGE_16KB]
                LONG.map_ptr_entries_to_levels[mode] = [None]
                LONG.map_entries_to_shifts[mode] = {PTPAGE_16KB: 14}
            elif total_levels == 2:
                LONG.map_datapages_entries_to_levels[mode] = [
                    [PTBLOCK_L2_16KB],
                    [PTPAGE_16KB],
                ]
                LONG.map_ptr_entries_to_levels[mode] = [PTP_16KB_L0, None]
                LONG.map_entries_to_shifts[mode] = {
                    PTP_16KB_L0: 25,
                    PTPAGE_16KB: 14,
                    PTBLOCK_L2_16KB: 25,
                }
            elif total_levels == 3:
                LONG.map_datapages_entries_to_levels[mode] = [
                    [None],
                    [PTBLOCK_L2_16KB],
                    [PTPAGE_16KB],
                ]
                LONG.map_ptr_entries_to_levels[mode] = [
                    PTP_16KB_L0,
                    PTP_16KB_L1,
                    None,
                ]
                LONG.map_entries_to_shifts[mode] = {
                    PTP_16KB_L0: 36,
                    PTP_16KB_L1: 25,
                    PTPAGE_16KB: 14,
                    PTBLOCK_L2_16KB: 25,
                }
            else:
                LONG.map_datapages_entries_to_levels[mode] = [
                    [None],
                    [None],
                    [PTBLOCK_L2_16KB],
                    [PTPAGE_16KB],
                ]
                LONG.map_ptr_entries_to_levels[mode] = [
                    PTP_16KB_L0,
                    PTP_16KB_L2,
                    None,
                ]
                LONG.map_entries_to_shifts[mode] = {
                    PTP_16KB_L0: 47,
                    PTP_16KB_L1: 36,
                    PTP_16KB_L2: 25,
                    PTPAGE_16KB: 14,
                    PTBLOCK_L2_16KB: 25,
                }

        else:
            if total_levels == 1:
                LONG.map_datapages_entries_to_levels[mode] = [PTPAGE_64KB]
                LONG.map_ptr_entries_to_levels[mode] = [None]
                LONG.map_entries_to_shifts[mode] = {
                    PTPAGE_64KB: 16,
                    PTBLOCK_L2_64KB: 29,
                }
            elif total_levels == 2:
                LONG.map_datapages_entries_to_levels[mode] = [
                    [PTBLOCK_L2_64KB],
                    [PTPAGE_64KB],
                ]
                LONG.map_ptr_entries_to_levels[mode] = [PTP_64KB_L0, None]
                LONG.map_entries_to_shifts[mode] = {
                    PTP_64KB_L0: 29,
                    PTPAGE_64KB: 16,
                    PTBLOCK_L2_16KB: 29,
                }
            else:
                LONG.map_datapages_entries_to_levels[mode] = [
                    [None],
                    [PTBLOCK_L2_64KB],
                    [PTPAGE_64KB],
                ]
                LONG.map_ptr_entries_to_levels[mode] = [
                    PTP_64KB_L0,
                    PTP_64KB_L1,
                    None,
                ]
                LONG.map_entries_to_shifts[mode] = {
                    PTP_64KB_L0: 42,
                    PTP_64KB_L1: 29,
                    PTPAGE_64KB: 16,
                    PTBLOCK_L2_64KB: 29,
                }

do_show_radix_trees(args)

Show radix trees found

Source code in mmushell/architectures/aarch64.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",
        "Total levels",
        "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 TCR values found

Source code in mmushell/architectures/aarch64.py

def do_show_registers(self, args):
    """Show TCR 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 MMU table at chosen address. Usage: show_table ADDRESS (user, kernel) [level size]

Source code in mmushell/architectures/aarch64.py

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

    args = args.split()
    if len(args) < 2:
        logger.warning("Missing argument")
        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

    args[1] = args[1].lower()
    if args[1] not in ["kernel", "user"]:
        logger.warning("Mode must be kernel or user")
        return
    mode = args[1]

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

        trees_struct = LONG.tcr.get_trees_struct()
        valid_sizes = {"user": defaultdict(set), "kernel": defaultdict(set)}
        valid_sizes["kernel"][0].add(trees_struct["kernel"]["top_table_size"])
        valid_sizes["user"][0].add(trees_struct["user"]["top_table_size"])
        for i in range(1, trees_struct["kernel"]["total_levels"]):
            valid_sizes["kernel"][i].add(trees_struct["kernel"]["granule"])
        for i in range(1, trees_struct["user"]["total_levels"]):
            valid_sizes["user"][i].add(trees_struct["user"]["granule"])

        try:
            table_size = self.parse_int(args[3])
            if table_size not in valid_sizes[mode][lvl]:
                logging.warning(
                    f"Size not allowed for choosen level! Valid sizes are:{valid_sizes[mode][lvl]}"
                )
                return
        except ValueError:
            logger.warning("Invalid size value")
            return
    else:
        table_size = 0x10000
        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, mode=mode
    )
    print(table_obj)
    print(f"Invalid entries: {invalids} Table levels: {pt_classes}")

MMUShellGTruth

Bases: MMUShell

Source code in mmushell/architectures/aarch64.py

class MMUShellGTruth(MMUShell):
    def do_show_registers_gtruth(self, args):
        """Compare TCR 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 TCR was found
        all_tcrs = {}
        for reg_name, value_data in self.gtruth.items():
            if "TCR_EL1" in reg_name:
                for value, value_info in value_data.items():
                    if value not in all_tcrs or (value_info[1] > all_tcrs[value][1]):
                        all_tcrs[value] = (value_info[0], value_info[1])

        last_tcr = TCR_EL1(
            sorted(all_tcrs.keys(), key=lambda x: all_tcrs[x][1], reverse=True)[0]
        )

        tcr_fields_equals = {}
        for value_found_obj in self.data.regs_values["TCR_EL1"]:
            tcr_fields_equals[value_found_obj] = value_found_obj.count_fields_equals(
                last_tcr
            )
        k_sorted = sorted(
            tcr_fields_equals.keys(), key=lambda x: tcr_fields_equals[x], reverse=True
        )
        if not k_sorted:
            print(f"Correct TCR_EL1 value: {last_tcr}")
            print("TCR_EL1 fields found:... 0/4")
            print("FP: {}".format(str(len(self.data.regs_values["TCR_EL1"]))))
            return
        else:
            tcr_found = k_sorted[0]
            correct_fields_found = tcr_fields_equals[tcr_found]
            print(f"Correct TCR_EL1 value: {last_tcr}, Found: {tcr_found}")
            print("TCR_EL1 fields found:... {}/4".format(correct_fields_found))
            print("FP: {}".format(str(len(self.data.regs_values["TCR_EL1"]) - 1)))

    def do_show_radix_trees_gtruth(self, args):
        """Compare radix trees found 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)

        # Collect opcodes
        opcode_classes = defaultdict(list)
        for opcode_addr, opcode_data in self.data.opcodes.items():
            opcode_classes[
                (opcode_data["instruction"], opcode_data["register"])
            ].append(opcode_addr)

        # Kernel radix trees
        # Filtering using the same criteria used by the algorithm, however we test only candidates which are possible
        # False Negatives beacuse the interection must always pass the check!
        mmu_w_opcode_addrs = (
            opcode_classes[("MSR", "TCR_EL1")]
            + opcode_classes[("MSR", "TTBR0_EL1")]
            + opcode_classes[("MSR", "TTBR1_EL1")]
        )

        kernel_radix_trees = (
            False  # Some AArch64 machines do not have TTBR1_EL1 but only TTBR0_EL1
        )
        for key in ["TTBR1_EL1", "TTBR1_EL1_S"]:
            for value, data in tqdm(self.gtruth.get(key, {}).items()):
                ttbr = TTBR1_EL1(value)

                try:
                    for addr_r, data_r in ttbr1s.items():
                        ttbr_r, dates_r = data_r
                        if ttbr.is_mmu_equivalent_to(ttbr_r):
                            if data[0] < dates_r[0]:
                                ttbr1s[addr_r][1][0] = data[0]
                            if data[1] > dates_r[1]:
                                ttbr1s[addr_r][1][1] = data[1]
                            raise UserWarning
                except UserWarning:
                    continue

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

                ttbr1s[ttbr.address] = [ttbr, list(data)]
                kernel_radix_trees = True

        if kernel_radix_trees:
            tps = sorted(
                set(ttbr1s.keys()).intersection(set(self.data.ttbrs["kernel"].keys()))
            )
            fps = sorted(
                set(self.data.ttbrs["kernel"].keys()).difference(set(ttbr1s.keys()))
            )
            fns_candidates = set(ttbr1s.keys()).difference(
                set(self.data.ttbrs["kernel"].keys())
            )

            fns = []
            # Check False negatives
            for candidate in tqdm(fns_candidates):
                # Calculate physpace and discard empty ones
                consistency, pas = self.physpace(
                    candidate,
                    self.data.page_tables["kernel"],
                    self.data.empty_tables["kernel"],
                    mode="kernel",
                    hierarchical=True,
                    cache=ttbr1_phy_cache,
                )

                # Discard inconsistent one
                if not consistency:
                    continue

                # Check if at least one ERET opcode in physical address space
                for opcode_addr in opcode_classes[("ERET", "")]:
                    if pas.is_in_kernel_space(opcode_addr):
                        break
                else:
                    continue

                # WARNING! We cannot filter for user_size = 0 due to TCR_EL1.E0PD1 !
                # Check if at least one MMU opcode in physical address space
                for opcode_addr in mmu_w_opcode_addrs:
                    if pas.is_in_kernel_space(opcode_addr):
                        break
                else:
                    continue

                fns.append(candidate)
            fns.sort()

        # User radix trees
        for key in ["TTBR0_EL1", "TTBR0_EL1_S"]:
            for value, data in tqdm(self.gtruth.get(key, {}).items()):
                ttbr = TTBR0_EL1(value)

                try:
                    for addr_r, data_r in ttbr0s.items():
                        ttbr_r, dates_r = data_r
                        if ttbr.is_mmu_equivalent_to(ttbr_r):
                            if data[0] < dates_r[0]:
                                ttbr0s[addr_r][1][0] = data[0]
                            if data[1] > dates_r[1]:
                                ttbr0s[addr_r][1][1] = data[1]
                            raise UserWarning
                except UserWarning:
                    continue

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

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

        # If not TTBR1 uses TTBR0 as TTBR0+TTBR1
        user_ttbrs = list(self.data.ttbrs["user"])
        if not kernel_radix_trees:
            user_ttbrs.extend(self.data.ttbrs["kernel"].keys())

        tpsu = sorted(set(ttbr0s.keys()).intersection(set(user_ttbrs)))
        fpsu = sorted(set(user_ttbrs).difference(set(ttbr0s.keys())))
        fnsu_candidates = set(ttbr0s.keys()).difference(set(user_ttbrs))

        # Filter FN
        fnsu = []
        for candidate in tqdm(fnsu_candidates):
            # Calculate physpace and discard empty ones
            consistency, pas = self.physpace(
                candidate,
                self.data.page_tables["user"],
                self.data.empty_tables["user"],
                mode="user",
                hierarchical=True,
                cache=ttbr0_phy_cache,
            )

            # Discard inconsistent one
            if not consistency:
                continue

            # At least a page must be R or W in usermode
            for perms in pas.space:
                if perms[3] or perms[4]:
                    break
            else:
                continue

            # Check if at least one BLR opcode in physical address space
            for opcode_addr in opcode_classes[("BLR", "")]:
                if opcode_addr in pas:
                    break
            else:
                continue

            # Check if at least one RET opcode in physical address space
            for opcode_addr in opcode_classes[("RET", "")]:
                if opcode_addr in pas:
                    break
            else:
                continue

            fnsu.append(candidate)
        fnsu.sort()

        # Show results
        table = PrettyTable()
        table.field_names = ["Address", "Found", "Mode", "First seen", "Last seen"]
        kernel_regs = ttbr1s

        if kernel_radix_trees:
            umode = "U"
            for tp in sorted(tps):
                table.add_row(
                    [hex(tp), "X", "K", kernel_regs[tp][1][0], kernel_regs[tp][1][1]]
                )

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

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

        # User
        for tp in sorted(tpsu):
            table.add_row([hex(tp), "X", umode, ttbr0s[tp][1][0], ttbr0s[tp][1][1]])

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

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

        print(table)
        if kernel_radix_trees:
            print(f"TP:{len(tps)} FN:{len(fns)} FP:{len(fps)}")
            print(f"USER TP:{len(tpsu)} FN:{len(fnsu)} FP:{len(fpsu)}")
        else:
            print(f"TP:{len(tpsu)} FN:{len(fnsu)} FP:{len(fpsu)}")

do_show_radix_trees_gtruth(args)

Compare radix trees found with the ground truth

Source code in mmushell/architectures/aarch64.py

def do_show_radix_trees_gtruth(self, args):
    """Compare radix trees found 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)

    # Collect opcodes
    opcode_classes = defaultdict(list)
    for opcode_addr, opcode_data in self.data.opcodes.items():
        opcode_classes[
            (opcode_data["instruction"], opcode_data["register"])
        ].append(opcode_addr)

    # Kernel radix trees
    # Filtering using the same criteria used by the algorithm, however we test only candidates which are possible
    # False Negatives beacuse the interection must always pass the check!
    mmu_w_opcode_addrs = (
        opcode_classes[("MSR", "TCR_EL1")]
        + opcode_classes[("MSR", "TTBR0_EL1")]
        + opcode_classes[("MSR", "TTBR1_EL1")]
    )

    kernel_radix_trees = (
        False  # Some AArch64 machines do not have TTBR1_EL1 but only TTBR0_EL1
    )
    for key in ["TTBR1_EL1", "TTBR1_EL1_S"]:
        for value, data in tqdm(self.gtruth.get(key, {}).items()):
            ttbr = TTBR1_EL1(value)

            try:
                for addr_r, data_r in ttbr1s.items():
                    ttbr_r, dates_r = data_r
                    if ttbr.is_mmu_equivalent_to(ttbr_r):
                        if data[0] < dates_r[0]:
                            ttbr1s[addr_r][1][0] = data[0]
                        if data[1] > dates_r[1]:
                            ttbr1s[addr_r][1][1] = data[1]
                        raise UserWarning
            except UserWarning:
                continue

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

            ttbr1s[ttbr.address] = [ttbr, list(data)]
            kernel_radix_trees = True

    if kernel_radix_trees:
        tps = sorted(
            set(ttbr1s.keys()).intersection(set(self.data.ttbrs["kernel"].keys()))
        )
        fps = sorted(
            set(self.data.ttbrs["kernel"].keys()).difference(set(ttbr1s.keys()))
        )
        fns_candidates = set(ttbr1s.keys()).difference(
            set(self.data.ttbrs["kernel"].keys())
        )

        fns = []
        # Check False negatives
        for candidate in tqdm(fns_candidates):
            # Calculate physpace and discard empty ones
            consistency, pas = self.physpace(
                candidate,
                self.data.page_tables["kernel"],
                self.data.empty_tables["kernel"],
                mode="kernel",
                hierarchical=True,
                cache=ttbr1_phy_cache,
            )

            # Discard inconsistent one
            if not consistency:
                continue

            # Check if at least one ERET opcode in physical address space
            for opcode_addr in opcode_classes[("ERET", "")]:
                if pas.is_in_kernel_space(opcode_addr):
                    break
            else:
                continue

            # WARNING! We cannot filter for user_size = 0 due to TCR_EL1.E0PD1 !
            # Check if at least one MMU opcode in physical address space
            for opcode_addr in mmu_w_opcode_addrs:
                if pas.is_in_kernel_space(opcode_addr):
                    break
            else:
                continue

            fns.append(candidate)
        fns.sort()

    # User radix trees
    for key in ["TTBR0_EL1", "TTBR0_EL1_S"]:
        for value, data in tqdm(self.gtruth.get(key, {}).items()):
            ttbr = TTBR0_EL1(value)

            try:
                for addr_r, data_r in ttbr0s.items():
                    ttbr_r, dates_r = data_r
                    if ttbr.is_mmu_equivalent_to(ttbr_r):
                        if data[0] < dates_r[0]:
                            ttbr0s[addr_r][1][0] = data[0]
                        if data[1] > dates_r[1]:
                            ttbr0s[addr_r][1][1] = data[1]
                        raise UserWarning
            except UserWarning:
                continue

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

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

    # If not TTBR1 uses TTBR0 as TTBR0+TTBR1
    user_ttbrs = list(self.data.ttbrs["user"])
    if not kernel_radix_trees:
        user_ttbrs.extend(self.data.ttbrs["kernel"].keys())

    tpsu = sorted(set(ttbr0s.keys()).intersection(set(user_ttbrs)))
    fpsu = sorted(set(user_ttbrs).difference(set(ttbr0s.keys())))
    fnsu_candidates = set(ttbr0s.keys()).difference(set(user_ttbrs))

    # Filter FN
    fnsu = []
    for candidate in tqdm(fnsu_candidates):
        # Calculate physpace and discard empty ones
        consistency, pas = self.physpace(
            candidate,
            self.data.page_tables["user"],
            self.data.empty_tables["user"],
            mode="user",
            hierarchical=True,
            cache=ttbr0_phy_cache,
        )

        # Discard inconsistent one
        if not consistency:
            continue

        # At least a page must be R or W in usermode
        for perms in pas.space:
            if perms[3] or perms[4]:
                break
        else:
            continue

        # Check if at least one BLR opcode in physical address space
        for opcode_addr in opcode_classes[("BLR", "")]:
            if opcode_addr in pas:
                break
        else:
            continue

        # Check if at least one RET opcode in physical address space
        for opcode_addr in opcode_classes[("RET", "")]:
            if opcode_addr in pas:
                break
        else:
            continue

        fnsu.append(candidate)
    fnsu.sort()

    # Show results
    table = PrettyTable()
    table.field_names = ["Address", "Found", "Mode", "First seen", "Last seen"]
    kernel_regs = ttbr1s

    if kernel_radix_trees:
        umode = "U"
        for tp in sorted(tps):
            table.add_row(
                [hex(tp), "X", "K", kernel_regs[tp][1][0], kernel_regs[tp][1][1]]
            )

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

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

    # User
    for tp in sorted(tpsu):
        table.add_row([hex(tp), "X", umode, ttbr0s[tp][1][0], ttbr0s[tp][1][1]])

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

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

    print(table)
    if kernel_radix_trees:
        print(f"TP:{len(tps)} FN:{len(fns)} FP:{len(fps)}")
        print(f"USER TP:{len(tpsu)} FN:{len(fnsu)} FP:{len(fpsu)}")
    else:
        print(f"TP:{len(tpsu)} FN:{len(fnsu)} FP:{len(fpsu)}")

do_show_registers_gtruth(args)

Compare TCR values found with the ground truth

Source code in mmushell/architectures/aarch64.py

def do_show_registers_gtruth(self, args):
    """Compare TCR 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 TCR was found
    all_tcrs = {}
    for reg_name, value_data in self.gtruth.items():
        if "TCR_EL1" in reg_name:
            for value, value_info in value_data.items():
                if value not in all_tcrs or (value_info[1] > all_tcrs[value][1]):
                    all_tcrs[value] = (value_info[0], value_info[1])

    last_tcr = TCR_EL1(
        sorted(all_tcrs.keys(), key=lambda x: all_tcrs[x][1], reverse=True)[0]
    )

    tcr_fields_equals = {}
    for value_found_obj in self.data.regs_values["TCR_EL1"]:
        tcr_fields_equals[value_found_obj] = value_found_obj.count_fields_equals(
            last_tcr
        )
    k_sorted = sorted(
        tcr_fields_equals.keys(), key=lambda x: tcr_fields_equals[x], reverse=True
    )
    if not k_sorted:
        print(f"Correct TCR_EL1 value: {last_tcr}")
        print("TCR_EL1 fields found:... 0/4")
        print("FP: {}".format(str(len(self.data.regs_values["TCR_EL1"]))))
        return
    else:
        tcr_found = k_sorted[0]
        correct_fields_found = tcr_fields_equals[tcr_found]
        print(f"Correct TCR_EL1 value: {last_tcr}, Found: {tcr_found}")
        print("TCR_EL1 fields found:... {}/4".format(correct_fields_found))
        print("FP: {}".format(str(len(self.data.regs_values["TCR_EL1"]) - 1)))