Update attribute parser and importer

Author: christoph2

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.8.3
+current_version = 0.8.4
 commit = True
 tag = False
 

objutils/__init__.py

@@ -13,7 +13,7 @@
 The first parameter is always the codec name.
 """
 
-__version__ = "0.8.3"
+__version__ = "0.8.4"
 
 __all__ = [
     "Image",

objutils/dwarf/traverser.py

@@ -132,8 +132,40 @@ class AttributeParser:
     #    Little = 0
     #    Big = 1
 
-    def __init__(self, session):
-        self.session = session
+    def __init__(self, session_or_path, *, import_if_needed: bool = True, force_import: bool = False, quiet: bool = True):
+        """
+        Create an AttributeParser.
+
+        Parameters
+        ----------
+        session_or_path:
+            Either an existing SQLAlchemy session (backward compatible) or a path to an ELF/.prgdb file.
+            If a path is given, the corresponding program database will be opened (and imported if needed).
+        import_if_needed: bool
+            When a path to an ELF is provided, import DWARF into a sibling .prgdb if it doesn't exist yet.
+        force_import: bool
+            Force re-import when creating the database from an ELF path.
+        quiet: bool
+            Suppress non-error output during on-demand import when a path is provided.
+        """
+        # Lazy import to avoid heavy module import/cycles at module import time.
+        from objutils.elf import open_program_database  # local import by design
+
+        # Determine whether we received a session or a filesystem path.
+        if hasattr(session_or_path, "query"):
+            # Assume it's a SQLAlchemy session (backward compatible path)
+            self.session = session_or_path
+            self._model = None
+        else:
+            # Treat as a path (str or Path-like)
+            db_model = open_program_database(
+                session_or_path,
+                import_if_needed=import_if_needed,
+                force_import=force_import,
+                quiet=quiet,
+            )
+            self._model = db_model
+            self.session = db_model.session
         self.type_stack: set[int] = set()
         self.parsed_types: dict = {}
         self.att_types: dict = defaultdict(set)
@@ -161,6 +193,82 @@ def __init__(self, session):
         self.stack_machine = factory.stack_machine
         self.dwarf_expression = factory.dwarf_expression
 
+    @lru_cache(maxsize=64 * 1024)
+    def type_tree(self, obj: Union[int, model.DebugInformationEntry, DIEAttribute]) -> dict[str, Any] | CircularReference:
+        """Return a fully traversed type tree as a dict.
+
+        Accepts one of:
+        - a DIE offset (absolute),
+        - a DIE instance that has a DW_AT_type attribute,
+        - a DIEAttribute instance (DW_AT_type) whose value references a type.
+
+        The returned dictionary contains:
+        - tag: DWARF tag name for the type DIE
+        - attrs: non-structural attributes with values; nested "type" attributes
+                 are resolved recursively into dicts
+        - children: list of child DIE dicts (e.g., members, enumerators, subranges)
+
+        Circular references are represented by CircularReference(tag, name).
+        """
+        # Case 1: already an absolute DIE offset
+        if isinstance(obj, int):
+            return self.parse_type(obj)
+
+        # Case 2: attribute object (expected to be DW_AT_type)
+        if isinstance(obj, DIEAttribute):
+            # Try to resolve relative ref forms to absolute offset using the parent DIE if available
+            parent: Optional[model.DebugInformationEntry] = getattr(obj, "entry", None)
+            off = self._resolve_type_offset(obj, parent)
+            if off is None:
+                return {"tag": "<invalid>", "attrs": {}}
+            return self.parse_type(off)
+
+        # Case 3: a DIE that should have a DW_AT_type attribute
+        if hasattr(obj, "attributes_map") or hasattr(obj, "attributes"):
+            die = obj  # type: ignore[assignment]
+            type_attr = self._get_attr(die, "type")
+            if type_attr is None:
+                return {"tag": "<no-type>", "attrs": {}}
+            off = self._resolve_type_offset(type_attr, die)
+            if off is None:
+                return {"tag": "<invalid>", "attrs": {}}
+            return self.parse_type(off)
+
+        # Fallback
+        return {"tag": "<unsupported>", "attrs": {}}
+
+    def _resolve_type_offset(
+        self,
+        type_attr: DIEAttribute,
+        context_die: Optional[model.DebugInformationEntry],
+    ) -> Optional[int]:
+        """Resolve a DW_AT_type attribute's value to an absolute DIE offset.
+
+        Handles CU-relative reference forms by adding the DIE's cu_start.
+        Returns None if the attribute cannot be interpreted as an integer offset.
+        """
+        raw = getattr(type_attr, "raw_value", None)
+        try:
+            off = int(raw) if raw is not None else None
+        except Exception:
+            off = None
+        if off is None:
+            return None
+        try:
+            frm = getattr(type_attr, "form", None)
+            if frm in (
+                getattr(AttributeForm, "DW_FORM_ref1", None),
+                getattr(AttributeForm, "DW_FORM_ref2", None),
+                getattr(AttributeForm, "DW_FORM_ref4", None),
+                getattr(AttributeForm, "DW_FORM_ref8", None),
+                getattr(AttributeForm, "DW_FORM_ref_udata", None),
+            ):
+                base = getattr(context_die, "cu_start", 0) if context_die is not None else 0
+                off += int(base or 0)
+        except Exception:
+            pass
+        return off
+
     # The cache lives per-instance because "self" participates in the key.
     @lru_cache(maxsize=8192)
     def get_die(self, offset: int) -> model.DebugInformationEntry | None:

objutils/elf/__init__.py

@@ -254,7 +254,7 @@ def fetch(
         self,
         sections: str = None,
         name_pattern: str = None,
-        symbol_list: str = None,  # comma-separated list of symbols
+        symbol_list: typing.Optional[list[str]] = None,  # comma-separated list of symbols
         bindings: str = None,
         access: str = None,
         types_str: str = None,
@@ -315,7 +315,7 @@ def fetch(
                     flt.append(defs.SymbolType.STT_TLS)
                 query = query.filter(model.Elf_Symbol.st_type.in_(flt))
         if symbol_list:
-            name_flt = frozenset(symbol_list.split(","))
+            name_flt = frozenset(symbol_list)
             query = query.filter(model.Elf_Symbol.symbol_name.in_(name_flt))
         query = query.order_by(model.Elf_Symbol.section_name)
         if order_by_value:
@@ -1061,3 +1061,153 @@ def create_image(
         if callback:
             callback("stop", None)
         return img
+
+
+def import_dwarf_to_db(
+    elf_path: str,
+    out_db: str | None,
+    *,
+    quiet: bool = False,
+    verbose: bool = False,
+    run_lines: bool = True,
+    run_pubnames: bool = True,
+    run_aranges: bool = True,
+    run_mac: bool = False,
+    force: bool = False,
+) -> int:
+    """Import DWARF data from ELF into a .prgdb database file.
+
+    Returns an exit code compatible with the CLI script.
+    """
+    # Local import to avoid potential circular import at module import time.
+    from objutils.dwarf import DwarfProcessor
+
+    def _print(msg: str):
+        if not quiet:
+            print(msg)
+
+    elf_p = Path(elf_path)
+    if not elf_p.exists() or not elf_p.is_file():
+        _print(f"ELF file not found: {elf_path}")
+        return 2
+
+    default_db_path = elf_p.with_suffix(model.DB_EXTENSION)
+    try:
+        if force:
+            if default_db_path.exists():
+                try:
+                    default_db_path.unlink()
+                except Exception:
+                    pass
+            if out_db:
+                outp = Path(out_db)
+                if outp.exists():
+                    try:
+                        outp.unlink()
+                    except Exception:
+                        pass
+    except Exception:
+        pass
+
+    try:
+        ep = ElfParser(str(elf_p))
+    except Exception as e:
+        _print(f"Failed to open ELF file '{elf_path}': {e}")
+        return 2
+
+    if verbose:
+        _print(str(ep))
+
+    try:
+        dp = DwarfProcessor(ep)
+    except TypeError as te:
+        _print(f"No DWARF sections available in '{elf_path}': {te}")
+        return 1
+
+    if run_pubnames:
+        try:
+            dp.pubnames()
+        except Exception as e:
+            if verbose:
+                _print(f"Warning: pubnames failed: {e}")
+    if run_aranges:
+        try:
+            dp.aranges()
+        except Exception as e:
+            if verbose:
+                _print(f"Warning: aranges failed: {e}")
+    if run_lines:
+        try:
+            dp.do_lines()
+        except Exception as e:
+            if verbose:
+                _print(f"Warning: do_lines failed: {e}")
+
+    try:
+        dp.do_dbg_info()
+    except Exception as e:
+        _print(f"Error while parsing .debug_info: {e}")
+        return 3
+
+    if run_mac:
+        try:
+            dp.do_mac_info()
+        except Exception as e:
+            if verbose:
+                _print(f"Warning: do_mac_info failed: {e}")
+
+    try:
+        if out_db:
+            src_db = default_db_path
+            dst_db = Path(out_db)
+            if str(dst_db.resolve()) != str(src_db.resolve()):
+                if not src_db.exists():
+                    with model.Model(str(dst_db)) as _mdb:  # type: ignore[attr-defined]
+                        pass
+                else:
+                    try:
+                        ep.db.close()
+                    except Exception:
+                        pass
+                    import shutil as _shutil
+
+                    _shutil.copyfile(str(src_db), str(dst_db))
+                    _print(f"Wrote database: {dst_db}")
+            else:
+                _print(f"Database available at: {src_db}")
+        else:
+            _print(f"Database available at: {default_db_path}")
+    except Exception as e:
+        _print(f"Failed to write/copy database: {e}")
+        return 4
+
+    return 0
+
+
+def open_program_database(
+    path: str | os.PathLike,
+    *,
+    import_if_needed: bool = True,
+    force_import: bool = False,
+    quiet: bool = True,
+) -> model.Model:
+    """Open a program database (.prgdb) or derive it from an ELF file.
+
+    If `path` points to a .prgdb file, open and return it. If it points to an ELF
+    file (or a .prgdb does not yet exist next to the ELF), optionally import
+    DWARF to create the database and return the opened model.
+    """
+    p = Path(path)
+    if p.suffix.lower() == model.DB_EXTENSION:
+        return model.Model(str(p))
+
+    # Treat as ELF input; determine the sibling .prgdb
+    elf_path = p
+    db_path = elf_path.with_suffix(model.DB_EXTENSION)
+    if not db_path.exists():
+        if not import_if_needed:
+            raise FileNotFoundError(str(db_path))
+        rc = import_dwarf_to_db(str(elf_path), str(db_path), quiet=quiet, force=force_import)
+        if rc != 0:
+            raise RuntimeError(f"Failed to import DWARF from '{elf_path}' (rc={rc})")
+    return model.Model(str(db_path))