Repository: mandiant/capa
File: scripts/bulk-process.py (Python)

Code before:

def get_capa_results(args):
    """
    run capa against the file at the given path, using the given rules.

    args is a tuple, containing:
      rules (capa.rules.RuleSet): the rules to match
      signatures (List[str]): list of file system paths to signature files
      format (str): the name of the sample file format
      os (str): the name of the operating system
      path (str): the file system path to the sample to process

    args is a tuple because i'm not quite sure how to unpack multiple arguments using `map`.

    returns an dict with two required keys:
      path (str): the file system path of the sample to process
      status (str): either "error" or "ok"

    when status == "error", then a human readable message is found in property "error".
    when status == "ok", then the capa results are found in the property "ok".

    the capa results are a dictionary with the following keys:
      meta (dict): the meta analysis results
      capabilities (dict): the matched capabilities and their result objects
    """
    rules, sigpaths, format, os, path = args
    should_save_workspace = os.environ.get("CAPA_SAVE_WORKSPACE") not in ("0", "no", "NO", "n", None)
    logger.info("computing capa results for: %s", path)
    try:
        extractor = capa.main.get_extractor(
            path, format, os, capa.main.BACKEND_VIV, sigpaths, should_save_workspace, disable_progress=True
        )
    except capa.main.UnsupportedFormatError:
        # i'm 100% sure if multiprocessing will reliably raise exceptions across process boundaries.
        # so instead, return an object with explicit success/failure status.
        #
        # if success, then status=ok, and results found in property "ok"
        # if error, then status=error, and human readable message in property "error"
        return {
            "path": path,
            "status": "error",
            "error": f"input file does not appear to be a PE file: {path}",
        }
    except capa.main.UnsupportedRuntimeError:
        return {
            "path": path,
            "status": "error",
            "error": "unsupported runtime or Python interpreter",
        }
    except Exception as e:
        return {

Review comment (suggestion):
```suggestion
    rules, sigpaths, format, os_, path = args
```

Code after:

def get_capa_results(args):
    """
    run capa against the file at the given path, using the given rules.

    args is a tuple, containing:
      rules (capa.rules.RuleSet): the rules to match
      signatures (List[str]): list of file system paths to signature files
      format (str): the name of the sample file format
      os (str): the name of the operating system
      path (str): the file system path to the sample to process

    args is a tuple because i'm not quite sure how to unpack multiple arguments using `map`.

    returns an dict with two required keys:
      path (str): the file system path of the sample to process
      status (str): either "error" or "ok"

    when status == "error", then a human readable message is found in property "error".
    when status == "ok", then the capa results are found in the property "ok".

    the capa results are a dictionary with the following keys:
      meta (dict): the meta analysis results
      capabilities (dict): the matched capabilities and their result objects
    """
    rules, sigpaths, format, os_, path = args
    should_save_workspace = os.environ.get("CAPA_SAVE_WORKSPACE") not in ("0", "no", "NO", "n", None)
    logger.info("computing capa results for: %s", path)
    try:
        extractor = capa.main.get_extractor(
            path, format, os_, capa.main.BACKEND_VIV, sigpaths, should_save_workspace, disable_progress=True
        )
    except capa.main.UnsupportedFormatError:
        # i'm 100% sure if multiprocessing will reliably raise exceptions across process boundaries.
        # so instead, return an object with explicit success/failure status.
        #
        # if success, then status=ok, and results found in property "ok"
        # if error, then status=error, and human readable message in property "error"
        return {
            "path": path,
            "status": "error",
            "error": f"input file does not appear to be a PE file: {path}",
        }
    except capa.main.UnsupportedRuntimeError:
        return {
            "path": path,
            "status": "error",
            "error": "unsupported runtime or Python interpreter",
        }
    except Exception as e:
        return {