flare-floss/scripts/analysis/bulk_analyze.py

# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import sys
import logging
import pathlib
import argparse
import subprocess

from floss.cli import set_log_config
from floss.layout.extract import MIN_STR_LEN

# TODO: full deobf when those strings are first-class in layout output)
logger = logging.getLogger("floss.bulk")


def main():
    parser = argparse.ArgumentParser(description="Bulk analyze a directory of binaries with floss.")
    parser.add_argument("input_directory", type=pathlib.Path, help="Directory containing binaries to analyze.")
    parser.add_argument("output_directory", type=pathlib.Path, help="Directory to write JSON results to.")
    parser.add_argument(
        "-n",
        "--minimum-length",
        dest="min_length",
        type=int,
        default=MIN_STR_LEN,
        help="Minimum string length.",
    )
    parser.add_argument(
        "--save-rendered",
        action="store_true",
        help="Save the rendered output to a .txt file in the output directory.",
    )
    parser.add_argument(
        "--reprocess",
        action="store_true",
        help="Reprocess files even if the output files already exist.",
    )

    logging_group = parser.add_argument_group("logging arguments")
    logging_group.add_argument("-d", "--debug", action="store_true", help="Enable debugging output on STDERR.")
    logging_group.add_argument(
        "-q", "--quiet", action="store_true", help="Disable all status output except fatal errors."
    )
    args = parser.parse_args()

    set_log_config(args.debug, args.quiet)

    if not args.input_directory.is_dir():
        logger.error("Input path %s is not a directory.", args.input_directory)
        return 1

    args.output_directory.mkdir(parents=True, exist_ok=True)

    for file_path in args.input_directory.rglob("*"):
        if not file_path.is_file():
            continue

        relative_path = file_path.relative_to(args.input_directory)
        output_dir_for_file = args.output_directory / relative_path.parent
        output_dir_for_file.mkdir(parents=True, exist_ok=True)

        json_output_path = output_dir_for_file / f"{file_path.name}.json"
        rendered_output_path = output_dir_for_file / f"{file_path.name}.txt"

        should_analyze = not json_output_path.exists() or args.reprocess
        should_render = args.save_rendered and (not rendered_output_path.exists() or args.reprocess)

        if not should_analyze and not should_render:
            logger.info("Skipping file, all required outputs already exist: %s", file_path)
            continue

        if should_analyze:
            logger.info("Analyzing file: %s", file_path)
            cmd = [
                sys.executable,
                "-m",
                "floss.main",
                str(file_path),
                "--no-string-type",
                "stack",
                "tight",
                "decoded",
                "--json",
                "-n",
                str(args.min_length),
            ]
            if args.quiet:
                cmd.append("--quiet")
            if args.debug:
                cmd.append("--debug")

            try:
                result = subprocess.run(cmd, check=False, capture_output=True, text=True, encoding="utf-8")
                if result.returncode == 0:
                    with json_output_path.open("w", encoding="utf-8") as f:
                        f.write(result.stdout)
                    logger.info("Wrote JSON output to %s", json_output_path)

                    if should_render:
                        cmd_render = [
                            sys.executable,
                            "-m",
                            "floss.main",
                            str(json_output_path),
                        ]
                        if args.quiet:
                            cmd_render.append("--quiet")
                        if args.debug:
                            cmd_render.append("--debug")

                        result_render = subprocess.run(
                            cmd_render, check=False, capture_output=True, text=True, encoding="utf-8"
                        )
                        if result_render.returncode == 0:
                            with rendered_output_path.open("w", encoding="utf-8") as f:
                                f.write(result_render.stdout)
                            logger.info("Wrote rendered output to %s", rendered_output_path)
                        else:
                            logger.error(
                                "Failed to render file %s from JSON, exited with code %d",
                                file_path,
                                result_render.returncode,
                            )
                            if result_render.stdout:
                                logger.error("stdout:\n%s", result_render.stdout)
                            if result_render.stderr:
                                logger.error("stderr:\n%s", result_render.stderr)
                else:
                    logger.error("Failed to analyze file %s, exited with code %d", file_path, result.returncode)
                    if result.stdout:
                        logger.error("stdout:\n%s", result.stdout)
                    if result.stderr:
                        logger.error("stderr:\n%s", result.stderr)
            except Exception as e:
                logger.error("Failed to run analysis subprocess for file %s: %s", file_path, e, exc_info=True)

        elif should_render:
            logger.info("Generating rendered output from existing JSON for: %s", file_path)
            cmd = [
                sys.executable,
                "-m",
                "floss.main",
                str(json_output_path),
            ]
            if args.quiet:
                cmd.append("--quiet")
            if args.debug:
                cmd.append("--debug")

            try:
                result = subprocess.run(cmd, check=False, capture_output=True, text=True, encoding="utf-8")
                if result.returncode == 0:
                    with rendered_output_path.open("w", encoding="utf-8") as f:
                        f.write(result.stdout)
                    logger.info("Wrote rendered output to %s", rendered_output_path)
                else:
                    logger.error(
                        "Failed to generate rendered output for %s, exited with code %d", file_path, result.returncode
                    )
                    if result.stdout:
                        logger.error("stdout:\n%s", result.stdout)
                    if result.stderr:
                        logger.error("stderr:\n%s", result.stderr)
            except Exception as e:
                logger.error("Failed to run rendering subprocess for file %s: %s", file_path, e, exc_info=True)

    return 0


if __name__ == "__main__":
    sys.exit(main())