mirror of
https://github.com/vee1e/flare-floss.git
synced 2026-09-01 09:49:59 +00:00
471 lines
16 KiB
Python
471 lines
16 KiB
Python
import os
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import floss.cache
|
|
from floss.results import (
|
|
Strings,
|
|
Analysis,
|
|
Metadata,
|
|
AddressType,
|
|
StackString,
|
|
TightString,
|
|
ResultLayout,
|
|
ResultString,
|
|
StaticString,
|
|
DecodedString,
|
|
ResultDocument,
|
|
StringEncoding,
|
|
)
|
|
from floss.version import __version__
|
|
|
|
|
|
def make_doc(
|
|
sha256="a" * 64,
|
|
version=__version__,
|
|
min_length=4,
|
|
enable_static=True,
|
|
enable_stack=False,
|
|
enable_tight=False,
|
|
enable_decoded=False,
|
|
enable_language=False,
|
|
enable_layout=True,
|
|
enable_tags=True,
|
|
):
|
|
layout = ResultLayout(name="pe", offset=0, length=8) if enable_layout else None
|
|
return ResultDocument(
|
|
metadata=Metadata(
|
|
file_path="sample.exe",
|
|
md5="0" * 32,
|
|
sha1="0" * 40,
|
|
sha256=sha256,
|
|
version=version,
|
|
min_length=min_length,
|
|
),
|
|
analysis=Analysis(
|
|
enable_static_strings=enable_static,
|
|
enable_stack_strings=enable_stack,
|
|
enable_tight_strings=enable_tight,
|
|
enable_decoded_strings=enable_decoded,
|
|
enable_language_strings=enable_language,
|
|
enable_layout=enable_layout,
|
|
enable_tags=enable_tags,
|
|
),
|
|
strings=Strings(
|
|
static_strings=[StaticString(string="hello", offset=0, encoding=StringEncoding.ASCII)],
|
|
),
|
|
layout=layout,
|
|
)
|
|
|
|
|
|
def make_full_doc():
|
|
layout = ResultLayout(
|
|
name="pe",
|
|
offset=0,
|
|
length=8,
|
|
strings=[ResultString(string="layout", offset=0, size=6, encoding="ASCII", tags=["#winapi"])],
|
|
)
|
|
return ResultDocument(
|
|
metadata=Metadata(file_path="sample.exe", sha256="a" * 64, version=__version__, min_length=4),
|
|
analysis=Analysis(
|
|
enable_static_strings=True,
|
|
enable_stack_strings=True,
|
|
enable_tight_strings=True,
|
|
enable_decoded_strings=True,
|
|
enable_language_strings=True,
|
|
enable_layout=True,
|
|
enable_tags=True,
|
|
),
|
|
strings=Strings(
|
|
static_strings=[StaticString(string="hello", offset=0, encoding=StringEncoding.ASCII, tags=["#common"])],
|
|
stack_strings=[
|
|
StackString(
|
|
function=0,
|
|
string="stack",
|
|
encoding=StringEncoding.ASCII,
|
|
program_counter=0,
|
|
stack_pointer=0,
|
|
original_stack_pointer=0,
|
|
offset=0,
|
|
frame_offset=0,
|
|
)
|
|
],
|
|
tight_strings=[
|
|
TightString(
|
|
function=0,
|
|
string="tight",
|
|
encoding=StringEncoding.ASCII,
|
|
program_counter=0,
|
|
stack_pointer=0,
|
|
original_stack_pointer=0,
|
|
offset=0,
|
|
frame_offset=0,
|
|
)
|
|
],
|
|
decoded_strings=[
|
|
DecodedString(
|
|
address=0,
|
|
address_type=AddressType.STACK,
|
|
string="decoded",
|
|
encoding=StringEncoding.ASCII,
|
|
decoded_at=0,
|
|
decoding_routine=0,
|
|
)
|
|
],
|
|
language_strings=[StaticString(string="gostring", offset=0, encoding=StringEncoding.UTF8, tags=["#go"])],
|
|
language_strings_missed=[StaticString(string="missed", offset=0, encoding=StringEncoding.UTF8)],
|
|
),
|
|
layout=layout,
|
|
)
|
|
|
|
|
|
def wanted(
|
|
enable_static=True,
|
|
enable_stack=False,
|
|
enable_tight=False,
|
|
enable_decoded=False,
|
|
enable_language=False,
|
|
enable_layout=True,
|
|
enable_tags=True,
|
|
):
|
|
return Analysis(
|
|
enable_static_strings=enable_static,
|
|
enable_stack_strings=enable_stack,
|
|
enable_tight_strings=enable_tight,
|
|
enable_decoded_strings=enable_decoded,
|
|
enable_language_strings=enable_language,
|
|
enable_layout=enable_layout,
|
|
enable_tags=enable_tags,
|
|
)
|
|
|
|
|
|
def test_cache_enabled_default(monkeypatch):
|
|
monkeypatch.delenv(floss.cache.ENV_CACHE_ENABLE, raising=False)
|
|
assert floss.cache.cache_enabled()
|
|
|
|
|
|
@pytest.mark.parametrize("value", ("0", "false", "no", "n", ""))
|
|
def test_cache_enabled_disabled(monkeypatch, value):
|
|
monkeypatch.setenv(floss.cache.ENV_CACHE_ENABLE, value)
|
|
assert not floss.cache.cache_enabled()
|
|
|
|
|
|
@pytest.mark.parametrize("value", ("False", "NO", "N", "FALSE"))
|
|
def test_cache_enabled_disabled_case_insensitive(monkeypatch, value):
|
|
monkeypatch.setenv(floss.cache.ENV_CACHE_ENABLE, value)
|
|
assert not floss.cache.cache_enabled()
|
|
|
|
|
|
def test_cache_refresh_default(monkeypatch):
|
|
monkeypatch.delenv(floss.cache.ENV_CACHE_REFRESH, raising=False)
|
|
assert not floss.cache.cache_refresh()
|
|
|
|
|
|
@pytest.mark.parametrize("value", ("1", "true", "yes", "y"))
|
|
def test_cache_refresh_enabled(monkeypatch, value):
|
|
monkeypatch.setenv(floss.cache.ENV_CACHE_REFRESH, value)
|
|
assert floss.cache.cache_refresh()
|
|
|
|
|
|
@pytest.mark.parametrize("value", ("True", "YES", "Y", "TRUE"))
|
|
def test_cache_refresh_enabled_case_insensitive(monkeypatch, value):
|
|
monkeypatch.setenv(floss.cache.ENV_CACHE_REFRESH, value)
|
|
assert floss.cache.cache_refresh()
|
|
|
|
|
|
def test_get_cache_dir_override(monkeypatch, tmp_path):
|
|
monkeypatch.setenv(floss.cache.ENV_CACHE_DIR, str(tmp_path))
|
|
assert floss.cache.get_cache_dir() == tmp_path
|
|
|
|
|
|
def test_get_cache_dir_default(monkeypatch):
|
|
from platformdirs import user_cache_dir
|
|
|
|
monkeypatch.delenv(floss.cache.ENV_CACHE_DIR, raising=False)
|
|
assert floss.cache.get_cache_dir() == Path(user_cache_dir("floss"))
|
|
|
|
|
|
def test_compute_key():
|
|
sha256 = "a" * 64
|
|
# 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):
|
|
assert floss.cache.cache_file_path(tmp_path, "key") == tmp_path / "key.json"
|
|
|
|
|
|
def test_store_and_load_roundtrip(tmp_path):
|
|
doc = make_doc()
|
|
key = floss.cache.compute_key(doc.metadata.sha256, __version__)
|
|
|
|
assert floss.cache.store(tmp_path, key, doc)
|
|
assert (tmp_path / f"{key}.json").is_file()
|
|
|
|
loaded = floss.cache.load(tmp_path, key, doc.metadata.sha256, __version__)
|
|
assert loaded is not None
|
|
assert loaded.metadata.sha256 == doc.metadata.sha256
|
|
assert loaded.metadata.version == __version__
|
|
assert loaded.strings.static_strings[0].string == "hello"
|
|
|
|
|
|
def test_store_writes_json_schema(tmp_path):
|
|
doc = make_doc()
|
|
key = floss.cache.compute_key(doc.metadata.sha256, __version__)
|
|
|
|
floss.cache.store(tmp_path, key, doc)
|
|
payload = json.loads((tmp_path / f"{key}.json").read_text())
|
|
assert set(("metadata", "analysis", "strings")) <= set(payload.keys())
|
|
|
|
|
|
def test_load_missing(tmp_path):
|
|
key = floss.cache.compute_key("a" * 64, __version__)
|
|
assert floss.cache.load(tmp_path, key, "a" * 64, __version__) is None
|
|
|
|
|
|
def test_load_drops_invalid_json(tmp_path):
|
|
key = floss.cache.compute_key("a" * 64, __version__)
|
|
(tmp_path / f"{key}.json").write_text("{not json")
|
|
|
|
assert floss.cache.load(tmp_path, key, "a" * 64, __version__) is None
|
|
assert not (tmp_path / f"{key}.json").exists()
|
|
|
|
|
|
def test_load_drops_checksum_mismatch(tmp_path):
|
|
doc = make_doc(sha256="a" * 64)
|
|
key = floss.cache.compute_key("a" * 64, __version__)
|
|
floss.cache.store(tmp_path, key, doc)
|
|
|
|
assert floss.cache.load(tmp_path, key, "b" * 64, __version__) is None
|
|
assert not (tmp_path / f"{key}.json").exists()
|
|
|
|
|
|
def test_load_ignores_unlink_failure_on_invalid_entry(tmp_path, monkeypatch):
|
|
key = floss.cache.compute_key("a" * 64, __version__)
|
|
(tmp_path / f"{key}.json").write_text("{not json")
|
|
|
|
def raising_unlink(self, *args, **kwargs):
|
|
raise PermissionError("held open by another process")
|
|
|
|
monkeypatch.setattr(Path, "unlink", raising_unlink)
|
|
assert floss.cache.load(tmp_path, key, "a" * 64, __version__) is None
|
|
|
|
|
|
def test_load_ignores_unlink_failure_on_stale_entry(tmp_path, monkeypatch):
|
|
doc = make_doc(sha256="a" * 64)
|
|
key = floss.cache.compute_key("a" * 64, __version__)
|
|
floss.cache.store(tmp_path, key, doc)
|
|
|
|
def raising_unlink(self, *args, **kwargs):
|
|
raise PermissionError("held open by another process")
|
|
|
|
monkeypatch.setattr(Path, "unlink", raising_unlink)
|
|
assert floss.cache.load(tmp_path, key, "b" * 64, __version__) is None
|
|
|
|
|
|
def test_store_skips_when_replace_fails(tmp_path, monkeypatch):
|
|
doc = make_doc()
|
|
key = floss.cache.compute_key(doc.metadata.sha256, __version__)
|
|
|
|
def raising_replace(src, dst):
|
|
raise PermissionError("destination held open by antivirus")
|
|
|
|
monkeypatch.setattr(os, "replace", raising_replace)
|
|
assert floss.cache.store(tmp_path, key, doc) is False
|
|
assert not (tmp_path / f"{key}.json").exists()
|
|
# the temporary file is cleaned up
|
|
assert list(tmp_path.glob("*.tmp")) == []
|
|
|
|
|
|
def test_store_skips_when_mkdir_fails(tmp_path, monkeypatch):
|
|
doc = make_doc()
|
|
key = floss.cache.compute_key(doc.metadata.sha256, __version__)
|
|
cache_dir = tmp_path / "cache"
|
|
|
|
def raising_mkdir(self, *args, **kwargs):
|
|
raise PermissionError("no write access to cache directory")
|
|
|
|
monkeypatch.setattr(Path, "mkdir", raising_mkdir)
|
|
assert floss.cache.store(cache_dir, key, doc) is False
|
|
|
|
|
|
def test_store_skips_when_mkstemp_fails(tmp_path, monkeypatch):
|
|
doc = make_doc()
|
|
key = floss.cache.compute_key(doc.metadata.sha256, __version__)
|
|
|
|
def raising_mkstemp(*args, **kwargs):
|
|
raise OSError("no space left on device")
|
|
|
|
monkeypatch.setattr(tempfile, "mkstemp", raising_mkstemp)
|
|
assert floss.cache.store(tmp_path, key, doc) is False
|
|
assert not (tmp_path / f"{key}.json").exists()
|
|
|
|
|
|
def test_load_skips_when_isfile_fails(tmp_path, monkeypatch):
|
|
doc = make_doc()
|
|
key = floss.cache.compute_key(doc.metadata.sha256, __version__)
|
|
|
|
def raising_is_file(self, *args, **kwargs):
|
|
raise PermissionError("no read access to cache directory")
|
|
|
|
monkeypatch.setattr(Path, "is_file", raising_is_file)
|
|
assert floss.cache.load(tmp_path, key, doc.metadata.sha256, __version__) is None
|
|
|
|
|
|
def test_load_drops_version_mismatch(tmp_path):
|
|
doc = make_doc(version="9.9.9")
|
|
key = floss.cache.compute_key(doc.metadata.sha256, __version__)
|
|
floss.cache.store(tmp_path, key, doc)
|
|
|
|
assert floss.cache.load(tmp_path, key, doc.metadata.sha256, __version__) is None
|
|
assert not (tmp_path / f"{key}.json").exists()
|
|
|
|
|
|
def test_store_skips_when_locked(tmp_path):
|
|
doc = make_doc()
|
|
key = floss.cache.compute_key(doc.metadata.sha256, __version__)
|
|
lock_path = tmp_path / f"{key}.lock"
|
|
|
|
fd = floss.cache._acquire_lock(lock_path)
|
|
assert fd is not None
|
|
try:
|
|
assert floss.cache.store(tmp_path, key, doc) is False
|
|
assert not (tmp_path / f"{key}.json").exists()
|
|
finally:
|
|
floss.cache._release_lock(fd, lock_path)
|
|
|
|
|
|
def test_covers_matches():
|
|
doc = make_doc()
|
|
assert floss.cache.covers(doc, wanted(), 4)
|
|
|
|
|
|
def test_covers_missing_requested_type_is_miss():
|
|
doc = make_doc(enable_static=True, enable_stack=False)
|
|
assert not floss.cache.covers(doc, wanted(enable_stack=True), 4)
|
|
|
|
|
|
def test_covers_min_length_below_stored_is_miss():
|
|
doc = make_doc(min_length=8)
|
|
assert not floss.cache.covers(doc, wanted(), 4)
|
|
|
|
|
|
def test_covers_layout_enabled_mismatch_is_miss():
|
|
doc = make_doc(enable_layout=False)
|
|
assert not floss.cache.covers(doc, wanted(enable_layout=True), 4)
|
|
|
|
|
|
def test_covers_layout_disabled_is_hit():
|
|
# a cached layout can satisfy a no-layout request: materialize() drops it
|
|
doc = make_doc(enable_layout=True)
|
|
assert floss.cache.covers(doc, wanted(enable_layout=False), 4)
|
|
|
|
|
|
def test_covers_tags_enabled_mismatch_is_miss():
|
|
doc = make_doc(enable_tags=False)
|
|
assert not floss.cache.covers(doc, wanted(enable_tags=True), 4)
|
|
|
|
|
|
def test_covers_tags_disabled_is_hit():
|
|
# a tags-enabled document satisfies a no-tags request: materialize redacts
|
|
doc = make_doc(enable_tags=True)
|
|
assert floss.cache.covers(doc, wanted(enable_tags=False), 4)
|
|
|
|
|
|
def test_materialize_sets_file_path_and_filters(tmp_path):
|
|
doc = make_doc(min_length=4)
|
|
key = floss.cache.compute_key(doc.metadata.sha256, __version__)
|
|
floss.cache.store(tmp_path, key, doc)
|
|
loaded = floss.cache.load(tmp_path, key, doc.metadata.sha256, __version__)
|
|
assert loaded is not None
|
|
|
|
sample = tmp_path / "run" / "sample.exe"
|
|
mat = floss.cache.materialize(loaded, sample, wanted(), 8)
|
|
assert mat.metadata.file_path == str(sample)
|
|
# "hello" is 5 chars, below the requested -n 8
|
|
assert mat.strings.static_strings == []
|
|
|
|
|
|
def test_materialize_drops_layout_when_disabled(tmp_path):
|
|
doc = make_doc(enable_layout=True)
|
|
key = floss.cache.compute_key(doc.metadata.sha256, __version__)
|
|
floss.cache.store(tmp_path, key, doc)
|
|
loaded = floss.cache.load(tmp_path, key, doc.metadata.sha256, __version__)
|
|
assert loaded is not None
|
|
assert loaded.layout is not None
|
|
|
|
mat = floss.cache.materialize(loaded, tmp_path / "sample.exe", wanted(enable_layout=False), 4)
|
|
assert mat.layout is None
|
|
|
|
|
|
def test_materialize_clears_disabled_string_types():
|
|
mat = floss.cache.materialize(make_full_doc(), Path("sample.exe"), wanted(), 4)
|
|
assert mat.analysis.enable_static_strings is True
|
|
assert mat.strings.static_strings
|
|
|
|
disabled = Analysis(
|
|
enable_static_strings=False,
|
|
enable_stack_strings=False,
|
|
enable_tight_strings=False,
|
|
enable_decoded_strings=False,
|
|
enable_language_strings=False,
|
|
)
|
|
mat = floss.cache.materialize(make_full_doc(), Path("sample.exe"), disabled, 4)
|
|
assert mat.strings.static_strings == []
|
|
assert mat.strings.stack_strings == []
|
|
assert mat.strings.tight_strings == []
|
|
assert mat.strings.decoded_strings == []
|
|
assert mat.strings.language_strings == []
|
|
assert mat.strings.language_strings_missed == []
|
|
# static strings are disabled, so the layout that holds them is gone too
|
|
assert mat.layout is None
|
|
|
|
|
|
def test_materialize_clears_only_disabled_types():
|
|
disabled_stack = Analysis(enable_stack_strings=False)
|
|
mat = floss.cache.materialize(make_full_doc(), Path("sample.exe"), disabled_stack, 4)
|
|
assert mat.strings.stack_strings == []
|
|
assert mat.strings.static_strings
|
|
assert mat.strings.tight_strings
|
|
assert mat.strings.decoded_strings
|
|
assert mat.layout is not None
|
|
|
|
|
|
def test_materialize_syncs_layout_and_tag_flags():
|
|
doc = make_full_doc()
|
|
mat = floss.cache.materialize(doc, Path("sample.exe"), Analysis(enable_layout=False, enable_tags=False), 4)
|
|
assert mat.analysis.enable_layout is False
|
|
assert mat.analysis.enable_tags is False
|
|
|
|
|
|
def test_materialize_updates_metadata_min_length():
|
|
# a -n 6 hit against a -n 4 cache entry must report the requested length
|
|
mat = floss.cache.materialize(make_full_doc(), Path("sample.exe"), wanted(), 6)
|
|
assert mat.metadata.min_length == 6
|
|
assert all(len(s.string) >= 6 for s in mat.strings.static_strings)
|
|
|
|
|
|
def test_materialize_clears_tags_when_disabled():
|
|
doc = make_full_doc()
|
|
assert doc.strings.static_strings[0].tags == ["#common"]
|
|
assert doc.layout.strings[0].tags == ["#winapi"]
|
|
|
|
# Analysis(enable_tags=False) keeps every string type enabled, tags off
|
|
mat = floss.cache.materialize(doc, Path("sample.exe"), Analysis(enable_tags=False), 4)
|
|
assert mat.layout is not None
|
|
assert mat.strings.static_strings[0].tags == []
|
|
assert mat.strings.language_strings[0].tags == []
|
|
assert mat.layout.strings[0].tags == []
|
|
|
|
|
|
def test_materialize_keeps_tags_when_enabled():
|
|
mat = floss.cache.materialize(make_full_doc(), Path("sample.exe"), wanted(), 4)
|
|
assert mat.layout is not None
|
|
assert mat.strings.static_strings[0].tags == ["#common"]
|
|
assert mat.layout.strings[0].tags == ["#winapi"]
|