RISC-V

RISC-V

MMUShell

Bases: MMUShell

Source code in mmushell/architectures/riscv.py

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

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

    def do_parse_memory(self, args):
        """Parse memory to find tables"""
        if self.data.is_mem_parsed:
            logger.warning("Memory already parsed")
            return
        self.parse_memory()
        self.data.is_mem_parsed = True

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

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

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

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

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

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

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

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

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

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

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

                    else:
                        referenced_nxt.append(entry_obj.address)

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

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

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

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

        # Go back from PTLn up to top level, the particular form of PTLn permits to find PTL0
        logging.info("Go up the paging trees starting from data pages...")
        candidates = []
        already_explored = set()
        for page_addr in tqdm(self.data.data_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.extend(
                    self.radix_roots_from_data_page(
                        lvl,
                        addr,
                        self.data.reverse_map_pages,
                        self.data.reverse_map_tables,
                    )
                )
                already_explored.add(derived_address)
        candidates = list(
            set(candidates).intersection(self.data.page_tables["global"][0].keys())
        )
        candidates.sort()

        logger.info("Filter candidates...")
        satps = {}
        for candidate in tqdm(candidates):
            # Obtain radix tree infos
            consistency, pas = self.physpace(
                candidate, self.data.page_tables["global"], self.data.empty_tables
            )

            # Only consistent trees are valid
            if not consistency:
                continue

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

            vas = self.virtspace(candidate, 0, self.machine.mmu.top_prefix)
            satps[candidate] = RadixTree(candidate, 0, pas, vas)

        self.data.satps = satps
        self.data.is_radix_found = True

    def do_show_radix_trees(self, args):
        """Show found radix trees"""
        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)",
        ]
        table = PrettyTable()
        table.field_names = labels
        for satp in self.data.satps.values():
            table.add_row(satp.entry_resume_stringified())
        table.sortby = "Radix address"
        print(table)

do_find_radix_trees(args)

Reconstruct radix trees

Source code in mmushell/architectures/riscv.py

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

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

    # Go back from PTLn up to top level, the particular form of PTLn permits to find PTL0
    logging.info("Go up the paging trees starting from data pages...")
    candidates = []
    already_explored = set()
    for page_addr in tqdm(self.data.data_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.extend(
                self.radix_roots_from_data_page(
                    lvl,
                    addr,
                    self.data.reverse_map_pages,
                    self.data.reverse_map_tables,
                )
            )
            already_explored.add(derived_address)
    candidates = list(
        set(candidates).intersection(self.data.page_tables["global"][0].keys())
    )
    candidates.sort()

    logger.info("Filter candidates...")
    satps = {}
    for candidate in tqdm(candidates):
        # Obtain radix tree infos
        consistency, pas = self.physpace(
            candidate, self.data.page_tables["global"], self.data.empty_tables
        )

        # Only consistent trees are valid
        if not consistency:
            continue

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

        vas = self.virtspace(candidate, 0, self.machine.mmu.top_prefix)
        satps[candidate] = RadixTree(candidate, 0, pas, vas)

    self.data.satps = satps
    self.data.is_radix_found = True

do_parse_memory(args)

Parse memory to find tables

Source code in mmushell/architectures/riscv.py

def do_parse_memory(self, args):
    """Parse memory to find tables"""
    if self.data.is_mem_parsed:
        logger.warning("Memory already parsed")
        return
    self.parse_memory()
    self.data.is_mem_parsed = True

do_show_radix_trees(args)

Show found radix trees

Source code in mmushell/architectures/riscv.py

def do_show_radix_trees(self, args):
    """Show found radix trees"""
    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)",
    ]
    table = PrettyTable()
    table.field_names = labels
    for satp in self.data.satps.values():
        table.add_row(satp.entry_resume_stringified())
    table.sortby = "Radix address"
    print(table)

do_show_table(args)

Show MMU table. Usage: show_table ADDRESS [level]

Source code in mmushell/architectures/riscv.py

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

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

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

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

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

MMUShellGTruth

Bases: MMUShell

Source code in mmushell/architectures/riscv.py

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

        # Parse TP SATPs
        satp_tp = {}
        for satp in self.gtruth["SATP"]:
            new_satp = SATP(satp)
            if new_satp.address in satp_tp:
                continue

            # Validate SATP
            if new_satp.address not in self.data.page_tables["global"][0]:
                continue

            consistency, pas = self.physpace(
                new_satp.address,
                self.data.page_tables["global"],
                self.data.empty_tables,
            )
            if not consistency or (
                not pas.get_kernel_size() and not pas.get_user_size()
            ):
                continue
            satp_tp[new_satp.address] = new_satp

        # True positives, false negatives, false positives
        tps = set(satp_tp.keys()).intersection(set(self.data.satps.keys()))
        fns = set(satp_tp.keys()).difference(set(self.data.satps.keys()))
        fps = set(self.data.satps.keys()).difference(set(satp_tp.keys()))

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

        for fn in sorted(fns):
            table.add_row(
                [
                    hex(fn),
                    "",
                    self.gtruth["SATP"][satp_tp[fn].satp][0],
                    self.gtruth["SATP"][satp_tp[fn].satp][1],
                ]
            )

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

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

        # Export results for next analysis
        if len(args):
            from pickle import dump as dump_p

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

do_show_radix_trees_gtruth(args)

Compare found radix trees with the gound truth

Source code in mmushell/architectures/riscv.py

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

    # Parse TP SATPs
    satp_tp = {}
    for satp in self.gtruth["SATP"]:
        new_satp = SATP(satp)
        if new_satp.address in satp_tp:
            continue

        # Validate SATP
        if new_satp.address not in self.data.page_tables["global"][0]:
            continue

        consistency, pas = self.physpace(
            new_satp.address,
            self.data.page_tables["global"],
            self.data.empty_tables,
        )
        if not consistency or (
            not pas.get_kernel_size() and not pas.get_user_size()
        ):
            continue
        satp_tp[new_satp.address] = new_satp

    # True positives, false negatives, false positives
    tps = set(satp_tp.keys()).intersection(set(self.data.satps.keys()))
    fns = set(satp_tp.keys()).difference(set(self.data.satps.keys()))
    fps = set(self.data.satps.keys()).difference(set(satp_tp.keys()))

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

    for fn in sorted(fns):
        table.add_row(
            [
                hex(fn),
                "",
                self.gtruth["SATP"][satp_tp[fn].satp][0],
                self.gtruth["SATP"][satp_tp[fn].satp][1],
            ]
        )

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

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

    # Export results for next analysis
    if len(args):
        from pickle import dump as dump_p

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