Script Analysis: Bash Support#3121
Conversation
Revives and supersedes old PR mandiant#1080. Resolves merge conflicts and brings up to date with current master.
There was a problem hiding this comment.
Code Review
This pull request introduces Tree-Sitter script analysis support to capa, enabling the extraction of features from script languages such as C#, Python, Bash, and HTML. It adds a new script extractor backend, defines a new "script" format, introduces FileOffsetRangeAddress and ScriptLanguage features, and updates rules and freeze serialization to support them. Several issues and improvement opportunities were identified in the review: language detection in get_language should prioritize file extensions for accuracy and efficiency; get_idalib_extractor in tests must remain a context manager to prevent resource leaks; duplicate definitions of FORMAT_SCRIPT should be cleaned up; parse_integer and parse_int need to correctly handle signed numbers; a potential UnboundLocalError in capa/helpers.py should be avoided by using a with statement; set operations in rule indexing can be optimized; and missing assertions in the Tree-Sitter tests should be added to ensure proper validation.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| def get_language(path: Path) -> str: | ||
| try: | ||
| with path.open("rb") as f: | ||
| buf = f.read() | ||
| return get_language_ts(buf) | ||
| except ValueError: | ||
| return get_language_from_ext(str(path)) |
There was a problem hiding this comment.
The current implementation of get_language runs get_language_ts before checking the file extension. Since tree-sitter parsers (especially for permissive languages like Bash) can parse many simple scripts of other languages without producing ERROR nodes, get_language_ts will often incorrectly identify Python or C# files as Bash.
Additionally, checking the file extension first is much more efficient as it avoids reading the file contents from disk when the extension is already recognized.
We should check the file extension first, and only fall back to get_language_ts if the extension is unrecognized.
| def get_language(path: Path) -> str: | |
| try: | |
| with path.open("rb") as f: | |
| buf = f.read() | |
| return get_language_ts(buf) | |
| except ValueError: | |
| return get_language_from_ext(str(path)) | |
| def get_language(path: Path) -> str: | |
| try: | |
| return get_language_from_ext(str(path)) | |
| except ValueError: | |
| try: | |
| with path.open("rb") as f: | |
| buf = f.read() | |
| return get_language_ts(buf) | |
| except ValueError: | |
| raise ValueError(f"unsupported file type or extension for {path}") |
| def get_idalib_extractor(path: Path): | ||
| import capa.features.extractors.ida.idalib as idalib | ||
|
|
||
| has_idalib = getattr(idalib, "has_idalib", None) | ||
| load_idalib = getattr(idalib, "load_idalib", None) | ||
|
|
||
| if has_idalib is None or not has_idalib(): | ||
| raise RuntimeError("cannot find IDA idalib module.") | ||
|
|
||
| if load_idalib is None or not load_idalib(): | ||
| raise RuntimeError("failed to load IDA idalib module.") | ||
|
|
||
| import idapro | ||
| import ida_auto | ||
|
|
||
| import capa.features.extractors.ida.extractor | ||
|
|
||
| logger.debug("idalib: opening database...") | ||
|
|
||
| idapro.enable_console_messages(False) | ||
|
|
||
| # we set the primary and secondary Lumina servers to 0.0.0.0 to disable Lumina, | ||
| # which sometimes provides bad names, including overwriting names from debug info. | ||
| # | ||
| # use -R to load resources, which can help us embedded PE files. | ||
| # | ||
| # return values from open_database: | ||
| # 0 - Success | ||
| # 2 - User cancelled or 32-64 bit conversion failed | ||
| # 4 - Database initialization failed | ||
| # -1 - Generic errors (database already open, auto-analysis failed, etc.) | ||
| # -2 - User cancelled operation | ||
| ret = idapro.open_database( | ||
| str(path), run_auto_analysis=True, args="-Olumina:host=0.0.0.0 -Osecondary_lumina:host=0.0.0.0 -R" | ||
| ) | ||
| if ret != 0: | ||
| raise RuntimeError("failed to analyze input file") | ||
|
|
||
| logger.debug("idalib: waiting for analysis...") | ||
| ida_auto.auto_wait() | ||
| logger.debug("idalib: opened database.") | ||
|
|
||
| extractor = capa.features.extractors.ida.extractor.IdaFeatureExtractor() | ||
| fixup_idalib(path, extractor) | ||
| return extractor |
There was a problem hiding this comment.
In the previous implementation, get_idalib_extractor was a context manager that safely closed the IDA database in a finally block. In this new implementation, it is a regular function that returns the extractor but never closes the database. This will leave the database files open and locked, leading to resource leaks and potential test failures in a full test suite run.
We should restore the context manager implementation to ensure proper cleanup.
@contextlib.contextmanager
def get_idalib_extractor(path: Path):
import capa.features.extractors.ida.idalib as idalib
has_idalib = getattr(idalib, "has_idalib", None)
load_idalib = getattr(idalib, "load_idalib", None)
if has_idalib is None or not has_idalib():
raise RuntimeError("cannot find IDA idalib module.")
if load_idalib is None or not load_idalib():
raise RuntimeError("failed to load IDA idalib module.")
import idapro
import ida_auto
import capa.features.extractors.ida.extractor
logger.debug("idalib: opening database...")
idapro.enable_console_messages(False)
# we set the primary and secondary Lumina servers to 0.0.0.0 to disable Lumina,
# which sometimes provides bad names, including overwriting names from debug info.
#
# use -R to load resources, which can help us embedded PE files.
#
# return values from open_database:
# 0 - Success
# 2 - User cancelled or 32-64 bit conversion failed
# 4 - Database initialization failed
# -1 - Generic errors (database already open, auto-analysis failed, etc.)
# -2 - User cancelled operation
ret = idapro.open_database(
str(path), run_auto_analysis=True, args="-Olumina:host=0.0.0.0 -Osecondary_lumina:host=0.0.0.0 -R"
)
if ret != 0:
raise RuntimeError("failed to analyze input file")
logger.debug("idalib: waiting for analysis...")
ida_auto.auto_wait()
logger.debug("idalib: opened database.")
extractor = capa.features.extractors.ida.extractor.IdaFeatureExtractor()
fixup_idalib(path, extractor)
try:
yield extractor
finally:
logger.debug("closing database...")
idapro.close_database(save=False)
logger.debug("closed database.")| FORMAT_SCRIPT = "script" | ||
| FORMAT_DOTNET = "dotnet" | ||
| VALID_FORMAT = (FORMAT_PE, FORMAT_ELF, FORMAT_DOTNET) | ||
| FORMAT_SCRIPT = "script" | ||
| VALID_FORMAT = (FORMAT_PE, FORMAT_ELF, FORMAT_DOTNET, FORMAT_SCRIPT) |
There was a problem hiding this comment.
The constant FORMAT_SCRIPT is defined twice in this block (on line 498 and line 500). We should remove the duplicate definition.
| FORMAT_SCRIPT = "script" | |
| FORMAT_DOTNET = "dotnet" | |
| VALID_FORMAT = (FORMAT_PE, FORMAT_ELF, FORMAT_DOTNET) | |
| FORMAT_SCRIPT = "script" | |
| VALID_FORMAT = (FORMAT_PE, FORMAT_ELF, FORMAT_DOTNET, FORMAT_SCRIPT) | |
| FORMAT_SCRIPT = "script" | |
| FORMAT_DOTNET = "dotnet" | |
| VALID_FORMAT = (FORMAT_PE, FORMAT_ELF, FORMAT_DOTNET, FORMAT_SCRIPT) |
| def parse_integer(self, integer: str) -> int: | ||
| integer = integer.lower() | ||
| for suffix in self.integer_suffixes: | ||
| if integer.endswith(suffix): | ||
| integer = integer[: -len(suffix)] | ||
| break | ||
| for prefix, base in self.integer_prefixes: | ||
| if integer.startswith(prefix): | ||
| return int(integer, base) | ||
| return int(integer) |
There was a problem hiding this comment.
The parse_integer method does not handle negative or positive signs (e.g., -0x10 or +0x10) when checking for prefixes like 0x. This causes negative hexadecimal, octal, or binary literals to fail parsing and be ignored.
We should extract and handle the sign before checking the prefix.
| def parse_integer(self, integer: str) -> int: | |
| integer = integer.lower() | |
| for suffix in self.integer_suffixes: | |
| if integer.endswith(suffix): | |
| integer = integer[: -len(suffix)] | |
| break | |
| for prefix, base in self.integer_prefixes: | |
| if integer.startswith(prefix): | |
| return int(integer, base) | |
| return int(integer) | |
| def parse_integer(self, integer: str) -> int: | |
| integer = integer.lower() | |
| sign = 1 | |
| if integer.startswith("-"): | |
| sign = -1 | |
| integer = integer[1:] | |
| elif integer.startswith("+"): | |
| integer = integer[1:] | |
| for suffix in self.integer_suffixes: | |
| if integer.endswith(suffix): | |
| integer = integer[: -len(suffix)] | |
| break | |
| for prefix, base in self.integer_prefixes: | |
| if integer.startswith(prefix): | |
| return sign * int(integer, base) | |
| return sign * int(integer) |
| # Create a temporary file and redirect stdout to it | ||
| tfile = tempfile.TemporaryFile(mode="w+b") | ||
| _redirect_stdout(tfile.fileno()) | ||
| # Yield to caller, then redirect stdout back to the saved fd | ||
| yield | ||
| _redirect_stdout(saved_stdout_fd) | ||
| # Copy contents of temporary file to the given stream | ||
| tfile.flush() | ||
| tfile.seek(0, io.SEEK_SET) | ||
| stream.write(tfile.read()) | ||
| finally: | ||
| tfile.close() | ||
| os.close(saved_stdout_fd) |
There was a problem hiding this comment.
If tempfile.TemporaryFile raises an exception (e.g., due to permission issues or out of disk space), the finally block will execute. However, since tfile was never successfully assigned, calling tfile.close() will raise an UnboundLocalError, masking the original exception and making debugging difficult.
Using a with statement is the standard, safe, and Pythonic way to manage the temporary file lifecycle here.
| # Create a temporary file and redirect stdout to it | |
| tfile = tempfile.TemporaryFile(mode="w+b") | |
| _redirect_stdout(tfile.fileno()) | |
| # Yield to caller, then redirect stdout back to the saved fd | |
| yield | |
| _redirect_stdout(saved_stdout_fd) | |
| # Copy contents of temporary file to the given stream | |
| tfile.flush() | |
| tfile.seek(0, io.SEEK_SET) | |
| stream.write(tfile.read()) | |
| finally: | |
| tfile.close() | |
| os.close(saved_stdout_fd) | |
| with tempfile.TemporaryFile(mode="w+b") as tfile: | |
| _redirect_stdout(tfile.fileno()) | |
| # Yield to caller, then redirect stdout back to the saved fd | |
| yield | |
| _redirect_stdout(saved_stdout_fd) | |
| # Copy contents of temporary file to the given stream | |
| tfile.flush() | |
| tfile.seek(0, io.SEEK_SET) | |
| stream.write(tfile.read()) | |
| finally: | |
| os.close(saved_stdout_fd) |
| def parse_int(s: str) -> int: | ||
| if s.startswith(("0x", "-0x")): | ||
| if s.startswith("0x"): | ||
| return int(s, 0x10) | ||
| else: | ||
| return int(s, 10) |
There was a problem hiding this comment.
Removing "-0x" from the prefix check breaks parsing of negative hexadecimal values in rules (e.g., -0x10). We should keep support for negative hex values.
| def parse_int(s: str) -> int: | |
| if s.startswith(("0x", "-0x")): | |
| if s.startswith("0x"): | |
| return int(s, 0x10) | |
| else: | |
| return int(s, 10) | |
| def parse_int(s: str) -> int: | |
| if s.startswith(("0x", "-0x")): | |
| return int(s, 0x10) | |
| else: | |
| return int(s, 10) |
| new_candidates: list[str] = [] | ||
| for new_feature in new_features: | ||
| for candidate_name in feature_index.rules_by_feature.get(new_feature, ()): | ||
| # Deduplicate candidate rules at two levels: | ||
| # 1. Globally: Ensure we don't re-queue rules already evaluated or waiting in `candidate_rule_names`. | ||
| # 2. Locally: The `new_candidates` set prevents duplicates if a rule is triggered | ||
| # by both the matched rule name and its namespace in the same pass. | ||
| if candidate_name not in candidate_rule_names: | ||
| new_candidates.add(candidate_name) | ||
| new_candidates.extend( | ||
| rule_name | ||
| for rule_name in feature_index.rules_by_feature.get(new_feature, ()) | ||
| if rule_name not in candidate_rule_names | ||
| ) | ||
|
|
||
| if new_candidates: | ||
| candidate_rule_names.update(new_candidates) | ||
| candidate_rules.extend([self.rules[rule_name] for rule_name in new_candidates]) | ||
| candidate_rules.extend([self.rules[rule_name] for rule_name in set(new_candidates)]) |
There was a problem hiding this comment.
Using a list for new_candidates and then converting it to a set at the end is less efficient and more complex than keeping new_candidates as a set from the beginning.
| new_candidates: list[str] = [] | |
| for new_feature in new_features: | |
| for candidate_name in feature_index.rules_by_feature.get(new_feature, ()): | |
| # Deduplicate candidate rules at two levels: | |
| # 1. Globally: Ensure we don't re-queue rules already evaluated or waiting in `candidate_rule_names`. | |
| # 2. Locally: The `new_candidates` set prevents duplicates if a rule is triggered | |
| # by both the matched rule name and its namespace in the same pass. | |
| if candidate_name not in candidate_rule_names: | |
| new_candidates.add(candidate_name) | |
| new_candidates.extend( | |
| rule_name | |
| for rule_name in feature_index.rules_by_feature.get(new_feature, ()) | |
| if rule_name not in candidate_rule_names | |
| ) | |
| if new_candidates: | |
| candidate_rule_names.update(new_candidates) | |
| candidate_rules.extend([self.rules[rule_name] for rule_name in new_candidates]) | |
| candidate_rules.extend([self.rules[rule_name] for rule_name in set(new_candidates)]) | |
| new_candidates: set[str] = set() | |
| for new_feature in new_features: | |
| new_candidates.update( | |
| rule_name | |
| for rule_name in feature_index.rules_by_feature.get(new_feature, ()) | |
| if rule_name not in candidate_rule_names | |
| ) | |
| if new_candidates: | |
| candidate_rule_names.update(new_candidates) | |
| candidate_rules.extend([self.rules[rule_name] for rule_name in new_candidates]) |
| def do_test_ts_extractor_engine_get_assigned_property_names( | ||
| engine: TreeSitterExtractorEngine, root_node: Node, expected: List[str] | ||
| ): | ||
| assert len(list(engine.get_processed_property_names(root_node))) == len(expected) | ||
| for (node, _name), _expected_name in zip(engine.get_processed_property_names(root_node), expected): | ||
| assert isinstance(node, Node) | ||
| do_test_ts_base_engine_get_address(engine, node) |
There was a problem hiding this comment.
The loop in do_test_ts_extractor_engine_get_assigned_property_names unpacks _name and _expected_name but never asserts that they are equal. This means the test is not actually verifying that the correct property names are extracted.
We should add an assertion to verify the property names.
| def do_test_ts_extractor_engine_get_assigned_property_names( | |
| engine: TreeSitterExtractorEngine, root_node: Node, expected: List[str] | |
| ): | |
| assert len(list(engine.get_processed_property_names(root_node))) == len(expected) | |
| for (node, _name), _expected_name in zip(engine.get_processed_property_names(root_node), expected): | |
| assert isinstance(node, Node) | |
| do_test_ts_base_engine_get_address(engine, node) | |
| def do_test_ts_extractor_engine_get_assigned_property_names( | |
| engine: TreeSitterExtractorEngine, root_node: Node, expected: List[str] | |
| ): | |
| assert len(list(engine.get_processed_property_names(root_node))) == len(expected) | |
| for (node, name), expected_name in zip(engine.get_processed_property_names(root_node), expected): | |
| assert isinstance(node, Node) | |
| assert name == expected_name | |
| do_test_ts_base_engine_get_address(engine, node) |
83a97b2 to
600659a
Compare
|
It's currently hard to review this pr though since there are thousands of changes with many from the other branch. |


Work in Progress
I have created this branch by keeping script-feature branch as base to avoid code duplication. I will rebase this pr once that pr will merged successfully.
Temporary PR: only referencing Bash related changes (saniyafatima07#1)
Checklist