Fix review issues from #1375 (quantumstrand correctness/regression fixes)

floss/cache.py

@@ -94,9 +94,14 @@ def get_cache_dir() -> Path:
     return Path(user_cache_dir("floss"))
 
 
-def compute_key(sha256: str, version: str) -> str:
-    """The cache key: content-addressed sample hash + FLOSS version."""
-    return f"{sha256}-{version}"
+def compute_key(sha256: str, version: str, format: str) -> str:
+    """The cache key: content-addressed sample hash + analysis format + version.
+
+    The analysis format is part of the key so interpreting the same sample bytes
+    under a different ``--format`` (e.g. sc32 vs sc64) never serves a stale
+    document.
+    """
+    return f"{sha256}-{format}-{version}"
 
 
 def cache_file_path(cache_dir: Path, key: str) -> Path:

floss/enrich.py

@@ -29,8 +29,14 @@ def layout_encoding_to_string_encoding(encoding: str) -> StringEncoding:
 
 
 def is_structured_layout(layout_name: str) -> bool:
-    """True when compute_layout produced PE/ELF/Mach-O (not the binary fallback)."""
+    """True when compute_layout produced PE/ELF/Mach-O (not the binary fallback).
+
+    An XOR-obfuscated PE/ELF header appends `` (XOR decoded with key: 0x...)``
+    to the layout name, so strip any parenthetical suffix before matching.
+    """
     name = layout_name.lower()
+    if " (" in name:
+        name = name.split(" (", 1)[0]
     return name in ("pe", "elf") or name.startswith("macho")
 
 

floss/layout/base.py

@@ -235,16 +235,20 @@ def mark_structures(self, structures: Optional[Tuple[Dict[int, Structure], ...]]
         such as a PE file and all its data.
         """
         if structures:
-            for string in self.strings:
-                for structures_by_address in structures:
-                    structure = structures_by_address.get(string.offset)
-                    if structure:
-                        string.structure = structure.name
-                        break
+            self._mark_string_structures(structures)
 
         for child in self.children:
             child.mark_structures(structures=structures, **kwargs)
 
+    def _mark_string_structures(self, structures) -> None:
+        """attach the first matching structure name to this node's own strings."""
+        for string in self.strings:
+            for structures_by_address in structures:
+                structure = structures_by_address.get(string.offset)
+                if structure:
+                    string.structure = structure.name
+                    break
+
 
 class SectionLayout(Layout):
     model_config = ConfigDict(arbitrary_types_allowed=True)
@@ -291,6 +295,9 @@ def check_is_code_tagger(s: ExtractedString) -> Sequence[Tag]:
         super().tag_strings(taggers)
 
     def mark_structures(self, structures=(), **kwargs):
+        # apply the PE structures to this node's own strings too (e.g. strings
+        # in the header gap that are attached to the root layout node)
+        self._mark_string_structures((structures or ()) + (self.structures_by_address,))
         for child in self.children:
             if isinstance(child, (SectionLayout, SegmentLayout)):
                 # expected child of a PE
@@ -332,6 +339,9 @@ def check_is_code_tagger(s: ExtractedString) -> Sequence[Tag]:
         super().tag_strings(taggers)
 
     def mark_structures(self, structures: Optional[Tuple[Dict[int, Structure], ...]] = (), **kwargs):
+        # apply the ELF structures to this node's own strings too (the ELF
+        # header/program-header gap strings attach to the root layout node)
+        self._mark_string_structures((structures or ()) + (self.structures_by_address,))
         for child in self.children:
             if isinstance(child, (SectionLayout, SegmentLayout)):
                 child.mark_structures(structures=(structures or ()) + (self.structures_by_address,), **kwargs)

floss/main.py

@@ -152,6 +152,13 @@ def main(argv=None) -> int:
             # the summary's layout-derived sections cover static strings only, so
             # a non-static string-type selection is a shadow arg that does nothing.
             # reject it instead of silently ignoring it.
+            if args.analyze_functions:
+                # the summary is static-only, --analyze-functions cannot show
+                # static, so the combination leaves nothing to analyze
+                parser.error(
+                    "--summary only covers static strings, which --analyze-functions does not show; "
+                    "these flags cannot be combined"
+                )
             for flag, values in (
                 ("--string-type", args.enabled_string_types),
                 ("--no-string-type", args.disabled_string_types),

floss/pipeline.py

@@ -323,7 +323,7 @@ def analyze(options: Options) -> Optional[ResultDocument]:
     if options.cache_dir is not None and not options.analyze_functions and floss.cache.cache_enabled():
         # --analyze-functions changes which functions are analyzed, so it is a miss:
         # a hit would wrongly return a full-function document.
-        cache_key = floss.cache.compute_key(results.metadata.sha256, __version__)
+        cache_key = floss.cache.compute_key(results.metadata.sha256, __version__, options.format)
         if not floss.cache.cache_refresh():
             cached = floss.cache.load(options.cache_dir, cache_key, results.metadata.sha256, __version__)
             if cached is not None and floss.cache.covers(cached, analysis, options.min_length):
@@ -393,25 +393,34 @@ def analyze(options: Options) -> Optional[ResultDocument]:
     enabled_string_types = options.enabled_string_types or []
     disabled_string_types = options.disabled_string_types or []
     if results.metadata.language not in ("", "unknown"):
-        if not enabled_string_types and not disabled_string_types and options.prompt_deobfuscation:
-            # when stdout is redirected, such as in 'floss foo.exe | less' use default prompt values
-            if sys.stdout.isatty():
-                try:
-                    prompt = input("Do you want to enable string deobfuscation? (this could take a long time) [y/N] ")
-                except KeyboardInterrupt:
-                    raise PipelineError("aborted by user", exit_code=130)
-                except EOFError:
-                    raise PipelineError("aborted by user", exit_code=1)
-            else:
-                prompt = "n"
+        if not enabled_string_types and not disabled_string_types:
+            if options.prompt_deobfuscation:
+                # when stdout is redirected, such as in 'floss foo.exe | less' use default prompt values
+                if sys.stdout.isatty():
+                    try:
+                        prompt = input(
+                            "Do you want to enable string deobfuscation? (this could take a long time) [y/N] "
+                        )
+                    except KeyboardInterrupt:
+                        raise PipelineError("aborted by user", exit_code=130)
+                    except EOFError:
+                        raise PipelineError("aborted by user", exit_code=1)
+                else:
+                    prompt = "n"
 
-            if prompt.lower() == "y":
-                logger.info("enabled string deobfuscation")
-                analysis.enable_stack_strings = True
-                analysis.enable_tight_strings = True
-                analysis.enable_decoded_strings = True
+                if prompt.lower() == "y":
+                    logger.info("enabled string deobfuscation")
+                    analysis.enable_stack_strings = True
+                    analysis.enable_tight_strings = True
+                    analysis.enable_decoded_strings = True
+                else:
+                    logger.info("disabled string deobfuscation")
+                    analysis.enable_stack_strings = False
+                    analysis.enable_tight_strings = False
+                    analysis.enable_decoded_strings = False
             else:
-                logger.info("disabled string deobfuscation")
+                # -y/--yes: never prompt, default to not running deobfuscation
+                logger.info("string deobfuscation disabled (-y/--yes)")
                 analysis.enable_stack_strings = False
                 analysis.enable_tight_strings = False
                 analysis.enable_decoded_strings = False

floss/tags/__init__.py

@@ -29,6 +29,26 @@ def data_root() -> pathlib.Path:
     return pathlib.Path(__file__).resolve().parent / "data"
 
 
+# the first line of a Git LFS pointer file; used to detect unpulled databases
+LFS_POINTER_PREFIX = b"version https://git-lfs.github.com/"
+
+
+def ensure_not_lfs_pointer(path: pathlib.Path) -> None:
+    """Raise a clear error when a tag database file is an unpulled Git LFS pointer.
+
+    Without ``git lfs pull`` the LFS-tracked database files are tiny text
+    pointers, which the loaders otherwise fail on with confusing gzip/msgspec
+    errors.
+    """
+    try:
+        with path.open("rb") as f:
+            head = f.read(len(LFS_POINTER_PREFIX))
+    except OSError:
+        return
+    if head == LFS_POINTER_PREFIX:
+        raise ValueError(f"Git LFS pointer detected in {path.name}; please run `git lfs pull`")
+
+
 from floss.tags.engine import (
     Tagger,
     load_databases,

floss/tags/expert.py

@@ -27,7 +27,7 @@
 
 import msgspec
 
-from floss.tags import data_root
+from floss.tags import data_root, ensure_not_lfs_pointer
 
 
 class ExpertRule(msgspec.Struct):
@@ -78,6 +78,7 @@ def from_file(cls, path: pathlib.Path) -> "ExpertStringDatabase":
         substring_rules: List[ExpertRule] = []
         regex_rules: List[Tuple[ExpertRule, re.Pattern]] = []
 
+        ensure_not_lfs_pointer(path)
         decoder = msgspec.json.Decoder(type=ExpertRule)
         buf = path.read_bytes()
         for line in buf.split(b"\n"):

floss/tags/gp.py

@@ -28,7 +28,7 @@
 
 import msgspec
 
-from floss.tags import data_root
+from floss.tags import data_root, ensure_not_lfs_pointer
 
 Encoding = Literal["ascii"] | Literal["utf-16le"] | Literal["unknown"]
 # header | gap | overlay
@@ -135,11 +135,9 @@ def __contains__(self, other: bytes | str) -> bool:
     def from_file(cls, path: pathlib.Path) -> "StringHashDatabase":
         string_hashes: Set[bytes] = set()
 
+        ensure_not_lfs_pointer(path)
         buf = path.read_bytes()
 
-        if buf.startswith(b"version https://git-lfs.github.com/"):
-            raise ValueError(f"Git LFS pointer detected in {path.name}; please run `git lfs pull`")
-
         for i in range(0, len(buf), 8):
             string_hashes.add(buf[i : i + 8])
 

floss/tags/oss.py

@@ -25,7 +25,7 @@
 
 import msgspec
 
-from floss.tags import data_root
+from floss.tags import data_root, ensure_not_lfs_pointer
 
 
 class OpenSourceString(msgspec.Struct):
@@ -47,6 +47,7 @@ def __len__(self) -> int:
     @classmethod
     def from_file(cls, path: pathlib.Path) -> "OpenSourceStringDatabase":
         metadata_by_string: Dict[str, OpenSourceString] = {}
+        ensure_not_lfs_pointer(path)
         decoder = msgspec.json.Decoder(type=OpenSourceString)
         for line in gzip.decompress(path.read_bytes()).split(b"\n"):
             if not line:

floss/tags/winapi.py

@@ -23,7 +23,7 @@
 from typing import Set, Sequence
 from dataclasses import dataclass
 
-from floss.tags import data_root
+from floss.tags import data_root, ensure_not_lfs_pointer
 
 
 @dataclass
@@ -39,11 +39,13 @@ def from_dir(cls, path: pathlib.Path) -> "WindowsApiStringDatabase":
         dll_names: Set[str] = set()
         api_names: Set[str] = set()
 
+        ensure_not_lfs_pointer(path / "dlls.txt.gz")
         for line in gzip.decompress((path / "dlls.txt.gz").read_bytes()).decode("utf-8").splitlines():
             if not line:
                 continue
             dll_names.add(line)
 
+        ensure_not_lfs_pointer(path / "apis.txt.gz")
         for line in gzip.decompress((path / "apis.txt.gz").read_bytes()).decode("utf-8").splitlines():
             if not line:
                 continue

scripts/tags/extract_strings.py

@@ -263,7 +263,7 @@ def main():
     if max_len == -1:
         if args.libs:
             max_len = MAX_LEN_PES
-        elif args.libs:
+        elif args.pes:
             max_len = MAX_LEN_LIBS
         else:
             raise ValueError("unknown extraction type")