From 5bb74829f6f76411ff1ef0a2b1596b9a23a94f42 Mon Sep 17 00:00:00 2001 From: hanmo1 Date: Sat, 11 Jul 2026 00:26:49 +0800 Subject: [PATCH] fix(bash): link executed scripts across files --- graphify/extractors/bash.py | 24 +++++++++-- tests/test_extract.py | 80 +++++++++++++++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 3 deletions(-) diff --git a/graphify/extractors/bash.py b/graphify/extractors/bash.py index 160ea032c..455c7eb81 100644 --- a/graphify/extractors/bash.py +++ b/graphify/extractors/bash.py @@ -62,6 +62,7 @@ def add_edge(src: str, tgt: str, relation: str, line: int, add_edge(file_nid, entry_nid, "contains", 1) _BASH_SOURCE_COMMANDS = frozenset({"source", "."}) + _BASH_SCRIPT_RUNNERS = frozenset({"bash", "sh", "zsh", "ksh", "dash"}) # Parent node types that mean a contained command is part of a substitution # or expansion, not a real function call. Token-level filtering misses # these because `$(build)` exposes `build` as a child command whose name @@ -161,11 +162,11 @@ def walk(node, parent_nid: str) -> None: cmd_name_node = node.children[0] if cmd_name_node: cmd = literal(cmd_name_node) + args = [c for c in node.children + if c.type in ("word", "string", "concatenation") + and c != cmd_name_node] if cmd in _BASH_SOURCE_COMMANDS and cmd not in defined_functions: # find the path argument (first word after command name) - args = [c for c in node.children - if c.type in ("word", "string", "concatenation") - and c != cmd_name_node] if args: raw = _read_text(args[0], source).strip().strip("'\"") line = node.start_point[0] + 1 @@ -184,6 +185,23 @@ def walk(node, parent_nid: str) -> None: if tgt_nid: add_edge(file_nid, tgt_nid, "imports", line, context="import") + elif cmd and cmd not in defined_functions: + raw = cmd if cmd.endswith(".sh") else None + if cmd in _BASH_SCRIPT_RUNNERS and args: + raw = literal(args[0]) + if raw and raw.endswith(".sh"): + resolved = (path.parent / raw).resolve() + if resolved.is_file(): + target_path = resolved + if not path.is_absolute(): + try: + target_path = resolved.relative_to(Path.cwd().resolve()) + except ValueError: + pass + caller_nid = entry_nid if parent_nid == file_nid else parent_nid + add_edge(caller_nid, _make_id(str(target_path)) + "__entry", + "calls", node.start_point[0] + 1, + context="script_invocation") return if t == "declaration_command": diff --git a/tests/test_extract.py b/tests/test_extract.py index 6247ec59c..934d6fbfa 100644 --- a/tests/test_extract.py +++ b/tests/test_extract.py @@ -1210,6 +1210,86 @@ def test_extract_bash_emits_source_imports_from(tmp_path): assert import_edges[0].get("context") == "import" +@pytest.mark.parametrize("command", ["./helpers.sh", "bash ./helpers.sh"]) +def test_extract_bash_emits_script_invocation_calls(tmp_path, command): + helpers = tmp_path / "helpers.sh" + helpers.write_text("#!/bin/bash\necho helper\n", encoding="utf-8") + script = tmp_path / "deploy.sh" + script.write_text(f"#!/bin/bash\n{command}\n", encoding="utf-8") + + result = extract_bash(script) + invocation = [ + edge for edge in result["edges"] + if edge.get("relation") == "calls" and edge.get("context") == "script_invocation" + ] + + assert invocation == [{ + "source": _make_id(str(script)) + "__entry", + "target": _make_id(str(helpers.resolve())) + "__entry", + "relation": "calls", + "confidence": "EXTRACTED", + "source_file": str(script), + "source_location": "L2", + "weight": 1.0, + "context": "script_invocation", + }] + + +def test_extract_bash_skips_missing_and_shadowed_script_invocations(tmp_path): + helpers = tmp_path / "helpers.sh" + helpers.write_text("#!/bin/bash\necho helper\n", encoding="utf-8") + script = tmp_path / "deploy.sh" + script.write_text( + "#!/bin/bash\n" + "bash() { echo custom; }\n" + "bash ./helpers.sh\n" + "./missing.sh\n", + encoding="utf-8", + ) + + result = extract_bash(script) + + assert not any(edge.get("context") == "script_invocation" for edge in result["edges"]) + + +def test_extract_bash_skips_dynamic_script_invocation(tmp_path): + helpers = tmp_path / "helpers.sh" + helpers.write_text("#!/bin/bash\necho helper\n", encoding="utf-8") + script = tmp_path / "deploy.sh" + script.write_text('#!/bin/bash\nbash "./$SCRIPT.sh"\n', encoding="utf-8") + + result = extract_bash(script) + + assert not any(edge.get("context") == "script_invocation" for edge in result["edges"]) + + +def test_extract_bash_relative_script_invocation_targets_existing_entrypoint(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + helpers = Path("helpers.sh") + helpers.write_text("#!/bin/bash\necho helper\n", encoding="utf-8") + script = Path("deploy.sh") + script.write_text("#!/bin/bash\n./helpers.sh\n", encoding="utf-8") + + result = extract([script, helpers], cache_root=tmp_path, parallel=False) + node_ids = {node["id"] for node in result["nodes"]} + invocation = next(edge for edge in result["edges"] if edge.get("context") == "script_invocation") + + assert invocation["target"] in node_ids + + +def test_extract_bash_attributes_script_invocation_to_function(tmp_path): + helpers = tmp_path / "helpers.sh" + helpers.write_text("#!/bin/bash\necho helper\n", encoding="utf-8") + script = tmp_path / "deploy.sh" + script.write_text("#!/bin/bash\ndeploy() { bash ./helpers.sh; }\n", encoding="utf-8") + + result = extract_bash(script) + deploy = next(node for node in result["nodes"] if node["label"] == "deploy()") + invocation = next(edge for edge in result["edges"] if edge.get("context") == "script_invocation") + + assert invocation["source"] == deploy["id"] + + def test_extract_bash_no_self_loops(): result = extract_bash(FIXTURES / "sample.sh") for e in result["edges"]: