Skip to content

Detect strings written to global memory #1335

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
amancslab wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Detect strings written to global memory #1335

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 27 additions & 3 deletions floss/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@
get_functions_without_tightloops,
)
from floss.logging_ import TRACE, DebugLevel
from floss.stackstrings import extract_stackstrings
from floss.stackstrings import extract_stack_and_global_strings
from floss.tightstrings import extract_tightstrings
from floss.string_decoder import decode_strings
from floss.language.identify import Language, identify_language_and_version
Expand Down Expand Up @@ -769,13 +769,14 @@ def main(argv=None) -> int:
# and should be caught by the tightstrings extraction below
selected_functions = get_functions_without_tightloops(decoding_function_features)

results.strings.stack_strings = extract_stackstrings(
results.strings.stack_strings, written_global_strings = extract_stack_and_global_strings(
vw,
selected_functions,
args.min_length,
verbosity=args.verbose,
disable_progress=args.quiet or args.disable_progress,
)
results.strings.decoded_strings.extend(written_global_strings)
results.analysis.functions.analyzed_stack_strings = len(selected_functions)
results.metadata.runtime.stack_strings = get_runtime_diff(interim)
interim = time()
Expand Down Expand Up @@ -814,13 +815,36 @@ def main(argv=None) -> int:
logger.debug(" - 0x%x: score: %.3f, xrefs to: %d", fva, score, xrefs_to)

# TODO filter out strings decoded in library function or function only called by library function(s)
results.strings.decoded_strings = decode_strings(
decoded_strings = decode_strings(
vw,
fvas_to_emulate,
args.min_length,
verbosity=args.verbose,
disable_progress=args.quiet or args.disable_progress,
)

seen_decoded_strings = {
(
decoded_string.address,
decoded_string.address_type,
decoded_string.string,
decoded_string.encoding,
)
for decoded_string in results.strings.decoded_strings
}

for decoded_string in decoded_strings:
key = (
decoded_string.address,
decoded_string.address_type,
decoded_string.string,
decoded_string.encoding,
)
if key in seen_decoded_strings:
continue

seen_decoded_strings.add(key)
results.strings.decoded_strings.append(decoded_string)
results.analysis.functions.analyzed_decoded_strings = len(fvas_to_emulate)
results.metadata.runtime.decoded_strings = get_runtime_diff(interim)

Expand Down
147 changes: 137 additions & 10 deletions floss/stackstrings.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,27 +13,78 @@
# limitations under the License.


from typing import Set, List, Optional
from typing import Set, List, Tuple, Optional
from dataclasses import dataclass

import tqdm
import viv_utils
import envi.archs.i386
import envi.archs.amd64
import visgraph.pathcore as vg_path # type: ignore[import-untyped]
import viv_utils.emulator_drivers

import floss.utils
import floss.strings
from floss.utils import getPointerSize, extract_strings
from floss.render import Verbosity
from floss.results import StackString
from floss.results import AddressType, StackString, DecodedString

logger = floss.logging_.getLogger(__name__)
MAX_STACK_SIZE = 0x10000

MIN_NUMBER_OF_MOVS = 5


def get_written_global_memory(emu, initial_sp: int) -> List[Tuple[int, bytes]]:
"""
Return contiguous non-stack memory ranges written during emulation.

Only addresses belonging to the input workspace are included. This
excludes the emulated stack and temporary heap allocations.
"""
stack_start = 0
stack_end = 0

for mapva, mapsize, _permissions, _name in emu.getMemoryMaps():
if mapva <= initial_sp < mapva + mapsize:
stack_start = mapva
stack_end = mapva + mapsize
break

written_addresses: Set[int] = set()

for path_node in vg_path.getAllPaths(emu.path):
for _parent, _children, node_data in path_node:
for _opva, refva, written_data in node_data.get("writelog", []) or []:
for address in range(refva, refva + len(written_data)):
if stack_start <= address < stack_end:
continue

if emu.vw.isValidPointer(address):
written_addresses.add(address)

if not written_addresses:
return []

regions: List[Tuple[int, bytes]] = []
sorted_addresses = sorted(written_addresses)
region_start = sorted_addresses[0]
previous = sorted_addresses[0]

for address in sorted_addresses[1:]:
if address != previous + 1:
size = previous - region_start + 1
regions.append((region_start, emu.readMemory(region_start, size)))
region_start = address

previous = address

size = previous - region_start + 1
regions.append((region_start, emu.readMemory(region_start, size)))
Comment on lines +56 to +83

@gemini-code-assist gemini-code-assist Bot Jul 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Correctness, Performance, and Robustness Issues in Global Memory Extraction

There are three major issues with the current implementation of get_written_global_memory:

  1. Correctness (False Positives/Corrupted Strings): Using vg_path.getAllPaths(emu.path) retrieves the writelogs of all branches in the path tree, including sibling branches that were backtracked. Since those backtracked branches are not part of the current execution path, their writes are not in the emulator's memory. Reading those addresses from the current emulator memory state will read uninitialized/unrelated data, leading to incorrect string extraction.
  2. Performance (Quadratic Complexity): getAllPaths traverses the entire path tree, which can be extremely large in full coverage emulation, leading to quadratic complexity and severe performance degradation as emulation progresses.
  3. Robustness (Potential Crashes): emu.readMemory can raise exceptions (e.g., SegmentationViolation or generic Exception) if memory is unmapped or protected. Wrapping it in a try...except block prevents FLOSS from crashing during emulation.

Solution

Using vg_path.getPathToNode(emu.path, emu.path) correctly retrieves only the nodes along the current active execution path, which is both correct and significantly faster. Additionally, wrapping emu.readMemory in a try...except block ensures robustness.

    for _nodeid, _parentid, node_data in vg_path.getPathToNode(emu.path, emu.path):
        for _opva, refva, written_data in node_data.get("writelog", []) or []:
            for address in range(refva, refva + len(written_data)):
                if stack_start <= address < stack_end:
                    continue

                if emu.vw.isValidPointer(address):
                    written_addresses.add(address)

    if not written_addresses:
        return []

    regions: List[Tuple[int, bytes]] = []
    sorted_addresses = sorted(written_addresses)
    region_start = sorted_addresses[0]
    previous = sorted_addresses[0]

    for address in sorted_addresses[1:]:
        if address != previous + 1:
            size = previous - region_start + 1
            try:
                regions.append((region_start, emu.readMemory(region_start, size)))
            except Exception as e:
                logger.debug("failed to read memory at 0x%x (size: %d): %s", region_start, size, e)
            region_start = address

        previous = address

    size = previous - region_start + 1
    try:
        regions.append((region_start, emu.readMemory(region_start, size)))
    except Exception as e:
        logger.debug("failed to read memory at 0x%x (size: %d): %s", region_start, size, e)


return regions


@dataclass(frozen=True)
class CallContext:
"""
Expand All @@ -44,16 +95,44 @@ class CallContext:
sp: the current stack counter
init_sp: the initial stack counter at start of function
stack_memory: the active stack frame contents
global_memory: non-stack memory written before this context
pre_ctx_strings: strings identified before this context
"""

pc: int
sp: int
init_sp: int
stack_memory: bytes
global_memory: List[Tuple[int, bytes]]
pre_ctx_strings: Optional[Set[str]]


def extract_global_strings_from_context(
ctx: CallContext,
function_va: int,
min_length: int,
seen: Set[str],
) -> List[DecodedString]:
"""Extract strings from global memory written before a checkpoint."""
global_strings: List[DecodedString] = []

for region_address, region_data in ctx.global_memory:
for extracted_string in extract_strings(region_data, min_length, seen):
decoded_string = DecodedString(
address=region_address + extracted_string.offset,
address_type=AddressType.GLOBAL,
string=extracted_string.string,
encoding=extracted_string.encoding,
decoded_at=ctx.pc,
decoding_routine=function_va,
)

seen.add(decoded_string.string)
global_strings.append(decoded_string)

return global_strings


class StackstringContextMonitor(viv_utils.emulator_drivers.Monitor):
"""
Observes emulation and extracts the active stack frame contents:
Expand Down Expand Up @@ -97,7 +176,15 @@ def get_call_context(self, emu, va, pre_ctx_strings: Optional[Set[str]] = None)
stack_buf = emu.readMemory(stack_top, stack_size)
# would probably be an optimization here to strip garbage bytes, however, then we cannot easily track
# the correct frame offset
ctx = CallContext(va, stack_top, stack_bottom, stack_buf, pre_ctx_strings)
global_memory = get_written_global_memory(emu, self._init_sp)
ctx = CallContext(
va,
stack_top,
stack_bottom,
stack_buf,
global_memory,
pre_ctx_strings,
)
return ctx

# overrides emulator_drivers.Monitor
Expand Down Expand Up @@ -160,9 +247,13 @@ def get_basic_block_ends(vw):
return index


def extract_stackstrings(
vw, selected_functions, min_length, verbosity=Verbosity.DEFAULT, disable_progress=False
) -> List[StackString]:
def extract_stack_and_global_strings(
vw,
selected_functions,
min_length,
verbosity=Verbosity.DEFAULT,
disable_progress=False,
) -> Tuple[List[StackString], List[DecodedString]]:
"""
Extracts the stackstrings from functions in the given workspace.

Expand All @@ -174,21 +265,39 @@ def extract_stackstrings(
"""
logger.info("extracting stackstrings from %d functions", len(selected_functions))

stack_strings = list()
stack_strings: List[StackString] = []
global_strings: List[DecodedString] = []
bb_ends = get_basic_block_ends(vw)

pb = floss.utils.get_progress_bar(
selected_functions, disable_progress, desc="extracting stackstrings", unit=" functions"
selected_functions,
disable_progress,
desc="extracting stackstrings",
unit=" functions",
)
with tqdm.contrib.logging.logging_redirect_tqdm(), floss.utils.redirecting_print_to_tqdm():
with (
tqdm.contrib.logging.logging_redirect_tqdm(),
floss.utils.redirecting_print_to_tqdm(),
):
for fva in pb:
seen: Set[str] = floss.utils.get_referenced_strings(vw, fva)
logger.debug("extracting stackstrings from function 0x%x", fva)
ctxs = extract_call_contexts(vw, fva, bb_ends)
for n, ctx in enumerate(ctxs, 1):
logger.trace(
"extracting stackstrings at checkpoint: 0x%x stacksize: 0x%x", ctx.pc, ctx.init_sp - ctx.sp
"extracting stackstrings at checkpoint: 0x%x stacksize: 0x%x",
ctx.pc,
ctx.init_sp - ctx.sp,
)
for decoded_string in extract_global_strings_from_context(
ctx,
fva,
min_length,
seen,
):
floss.results.log_result(decoded_string, verbosity)
global_strings.append(decoded_string)

for s in extract_strings(ctx.stack_memory, min_length, seen):
frame_offset = (ctx.init_sp - ctx.sp) - s.offset - getPointerSize(vw)
ss = StackString(
Expand All @@ -204,4 +313,22 @@ def extract_stackstrings(
floss.results.log_result(ss, verbosity)
seen.add(s.string)
stack_strings.append(ss)
return stack_strings, global_strings


def extract_stackstrings(
vw,
selected_functions,
min_length,
verbosity=Verbosity.DEFAULT,
disable_progress=False,
) -> List[StackString]:
"""Extract stack strings while preserving the existing public API."""
stack_strings, _global_strings = extract_stack_and_global_strings(
vw,
selected_functions,
min_length,
verbosity=verbosity,
disable_progress=disable_progress,
)
return stack_strings
7 changes: 5 additions & 2 deletions tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,10 +46,13 @@ def extract_strings(vw):
yield s_decoded.string

no_tightloop_functions = get_functions_without_tightloops(decoding_function_features)
for s_stack in stackstrings.extract_stackstrings(
stack_strings, global_strings = stackstrings.extract_stack_and_global_strings(
vw, no_tightloop_functions, MIN_STRING_LENGTH, disable_progress=True
):
)
for s_stack in stack_strings:
yield s_stack.string
for s_global in global_strings:
yield s_global.string

tightloop_functions = get_functions_with_tightloops(decoding_function_features)
for s_tight in tightstrings.extract_tightstrings(vw, tightloop_functions, MIN_STRING_LENGTH, disable_progress=True):
Expand Down