From 89e87ff24c4fa80a9fd99e68275a5d1e3554eae4 Mon Sep 17 00:00:00 2001 From: Andrew Case Date: Sat, 1 Mar 2025 12:36:40 -0600 Subject: [PATCH] Fix crash-causing bugs. Add typing where possible. --- volatility3/framework/plugins/linux/kmsg.py | 90 ++++++++++++++------- 1 file changed, 59 insertions(+), 31 deletions(-) diff --git a/volatility3/framework/plugins/linux/kmsg.py b/volatility3/framework/plugins/linux/kmsg.py index 849060d3c..1069a312f 100644 --- a/volatility3/framework/plugins/linux/kmsg.py +++ b/volatility3/framework/plugins/linux/kmsg.py @@ -5,7 +5,7 @@ import re import logging from abc import ABC, abstractmethod from enum import Enum -from typing import Generator, Iterator, List, Tuple, Union +from typing import Generator, Iterator, List, Tuple, Optional from volatility3.framework import ( class_subclasses, @@ -73,7 +73,7 @@ class ABCKmsg(ABC): cls, context: interfaces.context.ContextInterface, config: interfaces.configuration.HierarchicalDict, - ) -> Iterator[Tuple[str, str, str, str, str]]: + ) -> Iterator[Tuple[str, str, str, Optional[str], str]]: """It calls each subclass symtab_checks() to test the required conditions to that specific kernel implementation. @@ -108,10 +108,12 @@ class ABCKmsg(ABC): break if kmsg_inst is None: - vollog.error("Unsupported kernel ring buffer implementation") + vollog.error( + "Unsupported kernel ring buffer implementation. Please file a bug on our issue tracker with your specific kernel version." + ) @abstractmethod - def run(self) -> Iterator[Tuple[str, str, str, str, str]]: + def run(self) -> Iterator[Tuple[str, str, str, Optional[str], str]]: """Walks through the specific kernel implementation. Returns: @@ -135,7 +137,7 @@ class ABCKmsg(ABC): bool: True if the kernel being analyzed fulfill the class requirements. """ - def get_string(self, addr: int, length: int) -> Union[str, None]: + def get_string(self, addr: int, length: int) -> Optional[str]: layer = self._context.layers[self.layer_name] if not layer.is_valid(addr, length): vollog.warning("Failed to read log record at address 0x%x", addr) @@ -161,21 +163,21 @@ class ABCKmsg(ABC): # obj could be log, printk_log or printk_info return self.nsec_to_sec_str(obj.ts_nsec) - def get_caller(self, obj): + def get_caller(self, obj) -> Optional[str]: # In some kernel versions, it's only available if CONFIG_PRINTK_CALLER is defined. # caller_id is a member of printk_log struct from 5.1 to the latest 5.9 # From kernels 5.10 on, it's a member of printk_info struct if obj.has_member("caller_id"): return self.get_caller_text(obj.caller_id) - else: - return renderers.NotAvailableValue() - def get_caller_text(self, caller_id): + return None + + def get_caller_text(self, caller_id) -> str: caller_name = "CPU" if caller_id & 0x80000000 else "Task" caller = f"{caller_name}({caller_id & ~0x80000000})" return caller - def get_prefix(self, obj) -> Tuple[int, int, str, str]: + def get_prefix(self, obj) -> Tuple[int, int, str, Optional[str]]: # obj could be log, printk_log or printk_info return ( obj.facility, @@ -213,6 +215,7 @@ class Kmsg_pre_3_5(ABCKmsg): def symtab_checks(cls, vmlinux) -> bool: return ( vmlinux.has_symbol("log_end") + and vmlinux.has_symbol("log_buf_len") and not vmlinux.has_symbol("log_first_idx") and not ( vmlinux.has_type("log") @@ -220,7 +223,7 @@ class Kmsg_pre_3_5(ABCKmsg): ) ) - def run(self) -> Iterator[Tuple[str, str, str, str, str]]: + def run(self) -> Iterator[Tuple[str, str, str, Optional[str], str]]: log_buf_ptr = self.vmlinux.object_from_symbol(symbol_name="log_buf") log_buf_len = self.vmlinux.object_from_symbol(symbol_name="log_buf_len") log_buf = utility.pointer_to_string(log_buf_ptr, count=log_buf_len) @@ -249,7 +252,7 @@ class Kmsg_pre_3_5(ABCKmsg): facility = level_facility >> 3 level_txt = self.get_level_text(level) facility_txt = self.get_facility_text(facility) - caller = renderers.NotAvailableValue() + caller = None yield facility_txt, level_txt, timestamp_str, caller, line @@ -266,10 +269,10 @@ class Kmsg_3_5_to_3_11(ABCKmsg): and vmlinux.has_symbol("log_first_idx") ) - def _get_log_struct_name(self): + def _get_log_struct_name(self) -> str: return "log" - def get_text_from_log(self, msg) -> Union[str, None]: + def get_text_from_log(self, msg) -> Optional[str]: log_struct_name = self._get_log_struct_name() log_struct_size = self.vmlinux.get_type(log_struct_name).size msg_offset = msg.vol.offset + log_struct_size @@ -283,7 +286,7 @@ class Kmsg_3_5_to_3_11(ABCKmsg): def get_dict_lines(self, msg) -> Generator[str, None, None]: if msg.dict_len == 0: - return None + return log_struct_name = self._get_log_struct_name() log_struct_size = self.vmlinux.get_type(log_struct_name).size @@ -293,12 +296,12 @@ class Kmsg_3_5_to_3_11(ABCKmsg): dict_data = layer.read(dict_offset, msg.dict_len) except exceptions.InvalidAddressException: vollog.debug("Unable to read kmsg dict from 0x%x", dict_offset) - return None + return for chunk in dict_data.split(b"\x00"): - yield " " + chunk.decode() + yield " " + chunk.decode(encoding="utf8", errors="replace") - def run(self) -> Iterator[Tuple[str, str, str, str, str]]: + def run(self) -> Iterator[Tuple[str, str, str, Optional[str], str]]: # First, the ring buffer size is determined in the kernel configuration # by CONFIG_LOG_BUF_SHIFT. This static buffer is held in the '__log_buf' # global variable, with 'log_buf' serving as a pointer to it. @@ -311,7 +314,13 @@ class Kmsg_3_5_to_3_11(ABCKmsg): # remains unused. Therefore, it is crucial to read from 'log_buf' rather # than '__log_buf'. - log_buf_ptr = self.vmlinux.object_from_symbol("log_buf") + # This can happen on kernels where log_buf is declared twice + try: + log_buf_ptr = self.vmlinux.object_from_symbol("log_buf") + except exceptions.InvalidAddressException: + vollog.debug("Unable to access `log_buf`. Bailing.") + return + log_buf_len = self.vmlinux.object_from_symbol("log_buf_len") log_first_idx = int(self.vmlinux.object_from_symbol("log_first_idx")) @@ -327,7 +336,10 @@ class Kmsg_3_5_to_3_11(ABCKmsg): while cur_idx < end_idx: msg_offset = log_buf_ptr + cur_idx # type: ignore - msg = self.vmlinux.object(object_type=log_struct_name, offset=msg_offset) + msg = self.vmlinux.object( + object_type=log_struct_name, offset=msg_offset, absolute=True + ) + try: if msg.len == 0: # As per kernel/printk.c: @@ -359,9 +371,14 @@ class Kmsg_3_11_to_5_10(Kmsg_3_5_to_3_11): @classmethod def symtab_checks(cls, vmlinux) -> bool: - return vmlinux.has_type("printk_log") + return ( + not vmlinux.has_type("printk_ringbuffer") + and vmlinux.has_type("printk_log") + and vmlinux.get_type("printk_log").has_member("ts_nsec") + and vmlinux.has_symbol("log_first_idx") + ) - def _get_log_struct_name(self): + def _get_log_struct_name(self) -> str: return "printk_log" @@ -412,9 +429,9 @@ class Kmsg_5_10_to_(ABCKmsg): @classmethod def symtab_checks(cls, vmlinux) -> bool: - return vmlinux.has_symbol("prb") + return vmlinux.has_symbol("prb") and vmlinux.has_type("printk_ringbuffer") - def get_text_from_data_ring(self, text_data_ring, desc, info) -> Union[str, None]: + def get_text_from_data_ring(self, text_data_ring, desc, info) -> Optional[str]: text_data_sz = text_data_ring.size_bits text_data_mask = 1 << text_data_sz @@ -423,7 +440,7 @@ class Kmsg_5_10_to_(ABCKmsg): # This record doesn't contain text if begin & 1: - return "" + return None # This means a wrap-around to the beginning of the buffer if begin > end: @@ -454,7 +471,7 @@ class Kmsg_5_10_to_(ABCKmsg): if dict_text: yield f" DEVICE={dict_text}" - def run(self) -> Iterator[Tuple[str, str, str, str, str]]: + def run(self) -> Iterator[Tuple[str, str, str, Optional[str], str]]: # static struct printk_ringbuffer *prb = &printk_rb_static; ringbuffers = self.vmlinux.object_from_symbol("prb").dereference() @@ -516,7 +533,7 @@ class Kmsg(interfaces.plugins.PluginInterface): _required_framework_version = (2, 6, 0) - _version = (1, 0, 2) + _version = (2, 0, 0) @classmethod def get_requirements(cls) -> List[interfaces.configuration.RequirementInterface]: @@ -528,17 +545,28 @@ class Kmsg(interfaces.plugins.PluginInterface): ), ] - def _generator(self) -> Iterator[Tuple[int, Tuple[str, str, str, str, str]]]: - for values in ABCKmsg.run_all(context=self.context, config=self.config): - yield (0, values) + def _generator( + self, + ) -> Iterator[Tuple[int, Tuple[str, str, str, Optional[str], str]]]: + for facility, level, timestamp, caller, line in ABCKmsg.run_all( + context=self.context, config=self.config + ): + yield 0, ( + facility, + level, + timestamp, + caller or renderers.NotAvailableValue(), + line, + ) def run(self): if not self.context.symbol_space.verify_table_versions( "dwarf2json", lambda version, _: (not version) or version > (0, 4, 1) ): - raise exceptions.SymbolSpaceError( + vollog.info( "Invalid symbol table, please ensure the ISF table produced by dwarf2json was produced using a version > 0.4.1" ) + return return renderers.TreeGrid( [