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

AGENTS.md

@@ -3,3 +3,4 @@
 - To set up the development environment, make sure that a venv is created and the pre-commit and pre-push hooks are installed, see `.pre-commit-config.yaml`
 - All lints, formatters and tests in `.github/workflows` **must** pass before making a PR. Enforce this strictly.
 - The `floss/` folder has the main functionality, while `scripts/` has auxiliary plugins and scripts. Docs are in `doc/`.
+- Result caching is on by default. When reproducing or diffing output across commits, disable it with `FLOSS_CACHE_ENABLE=0` (or `FLOSS_CACHE_DIR` to point it at a scratch dir) — otherwise a run may serve a cached document written by earlier code.

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 = "auto") -> 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/cli.py

@@ -378,14 +378,6 @@ def make_parser():
     logging_group.add_argument(
         "-q", "--quiet", action="store_true", help="disable all status output on STDOUT except fatal errors"
     )
-    logging_group.add_argument(
-        "-y",
-        "--yes",
-        dest="prompt_deobfuscation",
-        action="store_false",
-        default=True,
-        help="do not prompt to enable string deobfuscation (defaults to not running it)",
-    )
     logging_group.add_argument(
         "--color",
         type=str,

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

@@ -149,16 +149,15 @@ def main(argv=None) -> int:
             if len(values) > 1 and StringType.ALL.value in values:
                 parser.error("%s: 'all' cannot be combined with other string types" % flag)
         if args.summary:
-            # 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.
-            for flag, values in (
-                ("--string-type", args.enabled_string_types),
-                ("--no-string-type", args.disabled_string_types),
-            ):
-                for value in values:
-                    if value != StringType.STATIC.value:
-                        parser.error("%s: '--summary' only covers static strings and does not take %s" % (flag, value))
+            if args.analyze_functions:
+                parser.error(
+                    "--summary only covers static strings, which --analyze-functions does not show; "
+                    "these flags cannot be combined"
+                )
+            if args.enabled_string_types or args.disabled_string_types:
+                # --summary is its own static-only view; reject any string-type
+                # selection rather than accepting shadow args
+                parser.error("--summary only covers static strings and does not take --string-type/--no-string-type")
         if args.max_strings is not None and args.max_strings <= 0:
             parser.error("--max-strings must be a positive integer")
         for pattern in args.queries:
@@ -280,7 +279,6 @@ def main(argv=None) -> int:
         large_file=args.large_file,
         quiet=args.quiet,
         verbose=args.verbose,
-        prompt_deobfuscation=args.prompt_deobfuscation,
         cache_dir=cache_dir,
     )
 

floss/pipeline.py

@@ -104,10 +104,7 @@ class Options:
     signatures: Optional[Path] = None
     large_file: bool = False
     quiet: bool = False
-    disable_progress: bool = False
     verbose: int = Verbosity.DEFAULT
-    # when True, prompt on TTY for deobfuscation on language binaries
-    prompt_deobfuscation: bool = True
     # analysis cache directory; None disables caching (default in the CLI is
     # the platform cache directory via floss.cache.get_cache_dir())
     cache_dir: Optional[Path] = None
@@ -323,7 +320,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):
@@ -390,32 +387,6 @@ def analyze(options: Options) -> Optional[ResultDocument]:
         analysis.enable_tight_strings = False
         analysis.enable_decoded_strings = False
 
-    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 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
-
     # in order of expected run time, fast to slow
     # 1. static strings (done above for language ID; layout-aware replace below when enabled)
     #  a) includes language-specific strings, if applicable
@@ -524,7 +495,7 @@ def analyze(options: Options) -> Optional[ResultDocument]:
                 text="analyzing program",
                 spinner="simpleDots",
                 stream=sys.stderr,
-                enabled=not (options.quiet or options.disable_progress),
+                enabled=not options.quiet,
             ):
                 with results.metadata.runtime.measure_and_set_time("vivisect"):
                     vw = load_vw(sample, options.format, sigpaths, should_save_workspace)
@@ -542,7 +513,7 @@ def analyze(options: Options) -> Optional[ResultDocument]:
                 raise PipelineError(e.args[0], exit_code=-1)
 
             decoding_function_features, library_functions = find_decoding_function_features(
-                vw, selected_functions, disable_progress=options.quiet or options.disable_progress
+                vw, selected_functions, disable_progress=options.quiet
             )
             results.analysis.functions.library = len(library_functions)
 
@@ -563,7 +534,7 @@ def analyze(options: Options) -> Optional[ResultDocument]:
                     funcs,
                     options.min_length,
                     verbosity=options.verbose,
-                    disable_progress=options.quiet or options.disable_progress,
+                    disable_progress=options.quiet,
                 )
                 results.analysis.functions.analyzed_stack_strings = len(funcs)
 
@@ -575,7 +546,7 @@ def analyze(options: Options) -> Optional[ResultDocument]:
                     tightloop_functions,
                     min_length=options.min_length,
                     verbosity=options.verbose,
-                    disable_progress=options.quiet or options.disable_progress,
+                    disable_progress=options.quiet,
                 )
                 results.analysis.functions.analyzed_tight_strings = len(tightloop_functions)
 
@@ -607,7 +578,7 @@ def analyze(options: Options) -> Optional[ResultDocument]:
                     fvas_to_emulate,
                     options.min_length,
                     verbosity=options.verbose,
-                    disable_progress=options.quiet or options.disable_progress,
+                    disable_progress=options.quiet,
                 )
                 results.analysis.functions.analyzed_decoded_strings = len(fvas_to_emulate)
 

floss/render/layout.py

@@ -81,11 +81,10 @@ def render_string_string(s: ResultString, tag_rules: TagRules) -> Text:
         raise ValueError("string should be hidden")
 
     # render like json, but strip the leading/trailing quote marks.
-    # this means that whitespace characters like \t and \n will be rendered as such,
-    # which ensures that the rendered string will be a single line.
+    # this means that whitespace characters like \t, \n, and \r are rendered as
+    # literal escape sequences, which keeps the rendered string on a single line
+    # and matches the escaping done by sanitize() in the classic views.
     rendered_string = json.dumps(s.string)[1:-1]
-    if "\\t" in rendered_string:
-        rendered_string = rendered_string.replace("\\t", "    ")
     return make_span(rendered_string, style=string_style)
 
 

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

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

tests/test_cache.py

@@ -190,7 +190,11 @@ def test_get_cache_dir_default(monkeypatch):
 
 def test_compute_key():
     sha256 = "a" * 64
-    assert floss.cache.compute_key(sha256, "1.0") == f"{sha256}-1.0"
+    # auto is the default analysis format
+    assert floss.cache.compute_key(sha256, "1.0") == f"{sha256}-auto-1.0"
+    assert floss.cache.compute_key(sha256, "1.0") == f"{sha256}-auto-1.0"
+    # the analysis format is part of the key: sc32 and sc64 never collide
+    assert floss.cache.compute_key(sha256, "1.0", "sc32") != floss.cache.compute_key(sha256, "1.0", "sc64")
 
 
 def test_cache_file_path(tmp_path):