From d4c815189285d796e1f6e25499fd0a29e69005fe Mon Sep 17 00:00:00 2001 From: yashrajbasav Date: Thu, 15 Oct 2026 14:17:44 +0530 Subject: [PATCH 1/4] fix(scan): require execution context for decoded shell payloads Signed-off-by: yashrajbasav --- .../analyzers/static_patterns_supply_chain.py | 45 +++++++++- tests/unit/test_patterns_new.py | 88 +++++++++++++++++-- 2 files changed, 125 insertions(+), 8 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py index 27f8f0864..2768d9810 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py +++ b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py @@ -330,10 +330,14 @@ def _decoded_literal_xor_calls(content: str) -> list[tuple[int, str]]: re.IGNORECASE, ) _SC3_SHELL_DECODE_COMMAND = re.compile(_SC3_SHELL_DECODER, re.IGNORECASE) +_SC3_DECODED_DATA_SINK = re.compile( + r"(?:[ \t]{1,16}-[A-Za-z0-9-]{1,16}){0,4}[ \t]{0,16}" + r"\|(?!\|)&?[ \t]{0,16}sha256sum(?=$|[\s;&|)`])" +) def _decoded_shell_literal_payloads(content: str) -> list[tuple[int, str]]: - """Decode literal payloads fed to a shell decoder. + """Decode literal payloads whose decoder output reaches an executor. Only an inline echo/printf argument or a here-string is decoded, as base64, base32 or (for xxd) hex; an ``openssl enc`` cipher is not applied. @@ -341,9 +345,48 @@ def _decoded_shell_literal_payloads(content: str) -> list[tuple[int, str]]: so a stray byte cannot hide the command. Returns ``(offset, text)`` pairs in source order. """ + if not any( + pattern.search(content) for pattern in (_SC3_PIPED_LITERAL, _SC3_HERE_STRING_LITERAL) + ): + return [] + pipeline_executor = re.compile(SC3_SHELL_PATTERNS[0][0], re.IGNORECASE | re.MULTILINE) + substitution_ranges: list[tuple[int, int]] = [] + substitution_ends: dict[int, int | None] = {} + backtick_ends: dict[int, int | None] = {} + for executor in re.finditer(_SC3_SUBSTITUTION_EXECUTOR, content, re.MULTILINE): + backtick = content[executor.end() - 1] == "`" + start = executor.end() - (1 if backtick else 2) + limit = min(len(content), start + _MAX_SHELL_LITERAL_CHARS) + skip = _skip_backtick_substitution if backtick else _skip_command_substitution + end = skip( + content, + start, + limit, + end_cache=backtick_ends if backtick else substitution_ends, + ) + # The parse limit cannot prove later decoder output is outside this wrapper. + body_end = len(content) if end is None else end - 1 + if substitution_ranges and executor.end() <= substitution_ranges[-1][1]: + previous_start, previous_end = substitution_ranges[-1] + substitution_ranges[-1] = (previous_start, max(previous_end, body_end)) + else: + substitution_ranges.append((executor.end(), body_end)) + if end is None: + break + substitution_starts = [start for start, _ in substitution_ranges] decoded: list[tuple[int, str]] = [] for pattern in (_SC3_PIPED_LITERAL, _SC3_HERE_STRING_LITERAL): for match in pattern.finditer(content): + decoder_start = match.start("decoder") + index = bisect_right(substitution_starts, decoder_start) - 1 + in_execution_substitution = index >= 0 and decoder_start < substitution_ranges[index][1] + if _SC3_DECODED_DATA_SINK.match(content, match.end()) is not None: + continue + if ( + not in_execution_substitution + and pipeline_executor.match(content, decoder_start) is None + ): + continue literal = match.group("literal") decoder = match.group("decoder").lower() try: diff --git a/tests/unit/test_patterns_new.py b/tests/unit/test_patterns_new.py index ca37fb85b..a97f0b87d 100644 --- a/tests/unit/test_patterns_new.py +++ b/tests/unit/test_patterns_new.py @@ -2700,6 +2700,12 @@ def test_sc3_shell_decode_to_interpreter(self, command: str) -> None: [ pytest.param(f"echo {_SHELL_FETCH_B64} | base64 -d > payload.sh", id="to_file"), pytest.param(f"echo {_SHELL_FETCH_B64} | base64 -d | sha256sum", id="sha256sum"), + pytest.param(f"echo {_SHELL_FETCH_B32} | base32 -d | sha256sum", id="base32_data"), + pytest.param(f"echo {_SHELL_FETCH_HEX} | xxd -r -p | sha256sum", id="hex_data"), + pytest.param( + f"echo {_SHELL_FETCH_B64} | openssl base64 -d -A | sha256sum", id="openssl_data" + ), + pytest.param(f"base64 -d <<< '{_SHELL_FETCH_B64}' | sha256sum", id="here_string_data"), pytest.param(f"echo {_SHELL_FETCH_B64} | base64 -d | ssh host 'cat > p'", id="ssh"), pytest.param( 'echo "$TOKEN" | cut -d. -f2 | base64 -d | python3 -m json.tool', id="json_tool" @@ -2736,10 +2742,31 @@ def test_sc3_shell_decode_to_interpreter(self, command: str) -> None: pytest.param("tar -xzf - -C . < <(base64 -d bundle.b64)", id="dot_argument"), ], ) - def test_shell_decode_without_execution_is_not_sc3(self, command: str) -> None: + def test_shell_decode_without_execution_is_not_sc2_or_sc3(self, command: str) -> None: findings = sc_mod.analyze(command + "\n", "setup.sh", "shell") - assert not [f for f in findings if f.rule_id == "SC3"] + assert not [f for f in findings if f.rule_id in {"SC2", "SC3"}] + + @pytest.mark.parametrize("separator", ["\n", "; ", " && "]) + @pytest.mark.parametrize("data_first", [True, False]) + def test_only_executed_literal_produces_sc2(self, separator: str, data_first: bool) -> None: + data = f"echo {_SHELL_FETCH_B64} | base64 -d | sha256sum" + executed = "echo ZWNobyBoZWxsbw== | base64 -d | sh" + commands = [data, executed] if data_first else [executed, data] + + findings = sc_mod.analyze(separator.join(commands), "setup.sh", "shell") + + assert any(f.rule_id == "SC3" for f in findings) + assert not any(f.rule_id == "SC2" for f in findings) + + @pytest.mark.parametrize("wrapper", ['eval "$(%s)"', "eval `%s`", "bash <(%s)"]) + def test_execution_substitution_keeps_data_only_literal_out_of_sc2(self, wrapper: str) -> None: + body = f"echo {_SHELL_FETCH_B64} | base64 -d | sha256sum; echo ZWNobyBoZWxsbwo= | base64 -d" + + findings = sc_mod.analyze(wrapper % body, "setup.sh", "shell") + + assert any(f.rule_id == "SC3" for f in findings) + assert not any(f.rule_id == "SC2" for f in findings) @pytest.mark.parametrize( "command", @@ -2755,6 +2782,39 @@ def test_shell_decode_without_execution_is_not_sc3(self, command: str) -> None: pytest.param(f"echo {_SHELL_FETCH_HEX} | xxd -r -p | sh", id="echo_hex"), pytest.param(f"sh <(xxd -r -p <<< {_SHELL_FETCH_HEX})", id="here_string_hex"), pytest.param(f"echo -n {_SHELL_FETCH_B64} | openssl base64 -d -A | sh", id="openssl"), + pytest.param(f'eval "$(echo {_SHELL_FETCH_B64} | base64 -d)"', id="eval"), + pytest.param(f"eval `echo {_SHELL_FETCH_B64} | base64 -d`", id="eval_backticks"), + pytest.param(f'bash -c "$(echo {_SHELL_FETCH_B64} | base64 -d)"', id="bash_c"), + pytest.param(f"source <(base64 -d <<< '{_SHELL_FETCH_B64}')", id="source"), + pytest.param(f"bash < <(echo {_SHELL_FETCH_B64} | base64 -d)", id="redirected_process"), + pytest.param( + f'bash <<< "$(echo {_SHELL_FETCH_B64} | base64 -d)"', id="here_string_sub" + ), + pytest.param( + f'eval "$(echo ZWNobyBoZWxsbwo= | base64 -d; echo {_SHELL_FETCH_B64} | base64 -d)"', + id="second_decoder_in_substitution", + ), + pytest.param( + f"echo ZWNobyBoZWxsbw== | base64 -d | echo {_SHELL_FETCH_B64} | base64 -d | sh", + id="overlapping_decoded_pipeline", + ), + pytest.param( + f'eval "$(echo {_SHELL_FETCH_B64} | base64 -d | cat)"', + id="substitution_with_filter", + ), + pytest.param( + f'eval "$(echo {_SHELL_FETCH_B64} | base64 -d > >(cat))"', + id="substitution_with_redirected_filter", + ), + pytest.param( + f'eval "$(echo {_SHELL_FETCH_B64} | base64 -d > payload.sh; cat payload.sh)"', + id="substitution_with_file_round_trip", + ), + pytest.param( + "eval \"$(echo ZWNobyBoZWxsbwo= | base64 -d; $(printf ''); " + f'echo {_SHELL_FETCH_B64} | base64 -d)"', + id="second_decoder_after_nested_substitution", + ), ], ) def test_sc2_shell_literal_decoded_command(self, command: str) -> None: @@ -2765,6 +2825,16 @@ def test_sc2_shell_literal_decoded_command(self, command: str) -> None: assert decoded[0].severity == Severity.HIGH assert decoded[0].location.start_line == 2 + @pytest.mark.parametrize("wrapper", ['eval "$(%s)"', "bash <(%s)"]) + def test_maximum_literal_retains_sc2_in_execution_substitution(self, wrapper: str) -> None: + script = (_SHELL_FETCH + "\n#").ljust(12_288) + payload = base64.b64encode(script.encode()).decode() + assert len(payload) == sc_mod._MAX_SHELL_LITERAL_CHARS + + findings = sc_mod.analyze(wrapper % f"echo {payload} | base64 -d", "setup.sh", "shell") + + assert any(f.rule_id == "SC2" and f.matched_text.startswith(_SHELL_FETCH) for f in findings) + def test_sc3_benign_decoded_payload_is_not_sc2(self) -> None: payload = base64.b64encode(b"echo hello").decode() findings = sc_mod.analyze(f"echo {payload} | base64 -d | sh\n", "setup.sh", "shell") @@ -2866,20 +2936,24 @@ def test_decoded_shell_literal_payloads_reports_literal_offsets(self) -> None: ("content", "expected"), [ pytest.param( - f"echo {_SHELL_FETCH_HEX.upper()}0 | xxd -r -p", _SHELL_FETCH, id="hex_odd_nibble" + f"echo {_SHELL_FETCH_HEX.upper()}0 | xxd -r -p | sh", + _SHELL_FETCH, + id="hex_odd_nibble", ), pytest.param( - f"base32 --decode <<< '{_SHELL_FETCH_B32.rstrip('=')}'", + f"base32 --decode <<< '{_SHELL_FETCH_B32.rstrip('=')}' | sh", _SHELL_FETCH, id="base32_unpadded", ), pytest.param( - f'echo \\"{_SHELL_FETCH_B64}\\" | base64 -d', _SHELL_FETCH, id="json_escaped_quotes" + f'echo \\"{_SHELL_FETCH_B64}\\" | base64 -d | sh', + _SHELL_FETCH, + id="json_escaped_quotes", ), pytest.param( "echo " + base64.b64encode(_SHELL_FETCH.encode() + b" #\xff").decode() - + " | base64 -d", + + " | base64 -d | sh", _SHELL_FETCH + " #�", id="invalid_utf8", ), @@ -2904,7 +2978,7 @@ def test_decoded_shell_literal_payloads_decodes_variants( ], ) def test_decoded_shell_literal_payloads_skips_unusable_literals(self, content: str) -> None: - assert sc_mod._decoded_shell_literal_payloads(content) == [] + assert sc_mod._decoded_shell_literal_payloads(content + " | sh") == [] @pytest.mark.parametrize( "content", From 06b431819bd868414816659559569bea5500225c Mon Sep 17 00:00:00 2001 From: yashrajbasav Date: Thu, 15 Oct 2026 14:17:45 +0530 Subject: [PATCH 2/4] fix(artifacts): preserve disk-container error provenance Signed-off-by: yashrajbasav --- src/skillspector/nested_artifacts.py | 45 ++++++++++++++----------- src/skillspector/nodes/build_context.py | 7 ++-- tests/nodes/test_nested_artifacts.py | 37 ++++++++++++++++++++ 3 files changed, 64 insertions(+), 25 deletions(-) diff --git a/src/skillspector/nested_artifacts.py b/src/skillspector/nested_artifacts.py index 14a2c9c1f..abfb44aae 100644 --- a/src/skillspector/nested_artifacts.py +++ b/src/skillspector/nested_artifacts.py @@ -128,6 +128,10 @@ class NestedInspectionResult: # Expected extensions and format mismatches are not content recognition. recognized_zip_paths: set[str] = field(default_factory=set) ledger_events: list[InspectionLedgerEvent] = field(default_factory=list) + # Keep the actual event objects: disk and virtual paths can share both a + # spelling and a work ID, so neither identifies which input produced it. + disk_container_events: list[InspectionLedgerEvent] = field(default_factory=list) + _active_disk_path: str | None = None uncompressed_bytes: int = 0 # Exceptions can target a top-level container before a virtual artifact row # exists. Preserve their canonical disposition so build_context can apply @@ -621,27 +625,26 @@ def _exception( limit_seconds: float | None = None, ) -> None: _mark_inventory_exception(result, path=path, reason=reason) - result.ledger_events.append( - ledger_event( - outcome=( - LedgerOutcome.PARTIAL - if reason in _PARTIAL_INVENTORY_REASONS - else LedgerOutcome.SKIPPED - ), - record_type=LedgerRecordType.SYSTEM, - phase="nested_artifact_inspection", - path=path, - reason=reason, - observed_bytes=observed_bytes, - limit_bytes=limit_bytes, - observed_artifacts=observed_artifacts, - limit_artifacts=limit_artifacts, - observed_depth=observed_depth, - limit_depth=limit_depth, - observed_seconds=observed_seconds, - limit_seconds=limit_seconds, - ) + event = ledger_event( + outcome=( + LedgerOutcome.PARTIAL if reason in _PARTIAL_INVENTORY_REASONS else LedgerOutcome.SKIPPED + ), + record_type=LedgerRecordType.SYSTEM, + phase="nested_artifact_inspection", + path=path, + reason=reason, + observed_bytes=observed_bytes, + limit_bytes=limit_bytes, + observed_artifacts=observed_artifacts, + limit_artifacts=limit_artifacts, + observed_depth=observed_depth, + limit_depth=limit_depth, + observed_seconds=observed_seconds, + limit_seconds=limit_seconds, ) + result.ledger_events.append(event) + if path == result._active_disk_path: + result.disk_container_events.append(event) def _time_exception( @@ -1233,6 +1236,7 @@ def inspect_nested_artifacts( last_checked_at=started_at, ) for path in dict.fromkeys(components): + result._active_disk_path = path if budget.halted: break if budget.expired(): @@ -1336,6 +1340,7 @@ def inspect_nested_artifacts( result=result, ) + result._active_disk_path = None result.components = list(dict.fromkeys(result.components)) result.uncompressed_bytes = budget.uncompressed_bytes _apply_inventory_overrides(result) diff --git a/src/skillspector/nodes/build_context.py b/src/skillspector/nodes/build_context.py index 30ed35dab..88a38fea2 100644 --- a/src/skillspector/nodes/build_context.py +++ b/src/skillspector/nodes/build_context.py @@ -3366,14 +3366,11 @@ def mark_excluded_nested_metadata( " Distinct descendant members may still be analyzed." ) prework_events.append(event) + disk_container_event_ids = {id(event) for event in nested.disk_container_events} for event in nested.ledger_events: path = event["path"] if path in reserved_paths: - # These reasons are emitted only for top-level disk inputs. - if path in nested.outer_metadata and event.get("reason_code") in { - LedgerReason.ARCHIVE_FORMAT_MISMATCH, - LedgerReason.ARCHIVE_MALFORMED, - }: + if id(event) in disk_container_event_ids: continue event["path"] = member_containers.get(path, path.rsplit("!/", 1)[0]) event["work_id"] = inspection_work_id( diff --git a/tests/nodes/test_nested_artifacts.py b/tests/nodes/test_nested_artifacts.py index 36e8212a9..e24e233e0 100644 --- a/tests/nodes/test_nested_artifacts.py +++ b/tests/nodes/test_nested_artifacts.py @@ -1545,6 +1545,43 @@ def test_collision_preserves_disk_container_errors( assert result["execution_successful"] is False +@pytest.mark.parametrize("unsafe_source", ["disk", "member", "both"]) +def test_collision_preserves_unsafe_member_event_source(tmp_path: Path, unsafe_source: str) -> None: + from skillspector.graph import graph + + (tmp_path / "SKILL.md").write_text("---\nname: disk-error\n---\nA helper.\n") + (tmp_path / "bundle.zip!").mkdir() + path = "bundle.zip!/inner.zip" + disk_members = {"disk.txt": b"Disk note.\n"} + archive_members = {"archive.txt": b"Archive note.\n"} + if unsafe_source in {"disk", "both"}: + disk_members["../unsafe.txt"] = b"A rejected disk entry.\n" + if unsafe_source in {"member", "both"}: + archive_members["../unsafe.txt"] = b"A rejected archive entry.\n" + disk_bytes = _zip_bytes(disk_members) + (tmp_path / path).write_bytes(disk_bytes) + _write_archive(tmp_path / "bundle.zip", {"inner.zip": _zip_bytes(archive_members)}) + + result = graph.invoke({"skill_path": str(tmp_path), "use_llm": False}) + + events = [ + event + for event in result["inspection_ledger"] + if event.get("reason_code") == LedgerReason.ARCHIVE_UNSAFE_MEMBER_PATH + ] + expected_paths = ( + {path, "bundle.zip"} + if unsafe_source == "both" + else {path if unsafe_source == "disk" else "bundle.zip"} + ) + assert len(events) == len(expected_paths) + assert {event["path"] for event in events} == expected_paths + for event in events: + assert ("withheld archive member" in event["message"]) is (event["path"] == "bundle.zip") + assert result["raw_file_cache"][path] == disk_bytes + assert result["execution_successful"] is False + + @pytest.mark.parametrize("disk_is_archive", [False, True]) def test_collision_keeps_distinct_archive_descendants( tmp_path: Path, disk_is_archive: bool From 89f28e036100bbed7ea283953239dbaf3c4c8487 Mon Sep 17 00:00:00 2001 From: yashrajbasav Date: Thu, 15 Oct 2026 14:35:39 +0530 Subject: [PATCH 3/4] fix(scan): preserve conservative checks outside checksum sinks Signed-off-by: yashrajbasav --- .../analyzers/static_patterns_supply_chain.py | 48 +++-------- tests/unit/test_patterns_new.py | 82 +++++++++++++++---- 2 files changed, 77 insertions(+), 53 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py index 2768d9810..044611f59 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py +++ b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py @@ -334,57 +334,31 @@ def _decoded_literal_xor_calls(content: str) -> list[tuple[int, str]]: r"(?:[ \t]{1,16}-[A-Za-z0-9-]{1,16}){0,4}[ \t]{0,16}" r"\|(?!\|)&?[ \t]{0,16}sha256sum(?=$|[\s;&|)`])" ) +_SC3_CHECKSUM_OVERRIDE = re.compile( + r"\bfunction[ \t]++['\"]?sha256sum\b" + r"|(? list[tuple[int, str]]: - """Decode literal payloads whose decoder output reaches an executor. + """Decode literal payloads fed to a shell decoder. Only an inline echo/printf argument or a here-string is decoded, as base64, base32 or (for xxd) hex; an ``openssl enc`` cipher is not applied. + Direct checksum consumers without a local override are data-only; other + output flows remain conservative because a saved payload can be executed later. Variables, files and downloads stay opaque. Invalid UTF-8 becomes U+FFFD, so a stray byte cannot hide the command. Returns ``(offset, text)`` pairs in source order. """ - if not any( - pattern.search(content) for pattern in (_SC3_PIPED_LITERAL, _SC3_HERE_STRING_LITERAL) - ): - return [] - pipeline_executor = re.compile(SC3_SHELL_PATTERNS[0][0], re.IGNORECASE | re.MULTILINE) - substitution_ranges: list[tuple[int, int]] = [] - substitution_ends: dict[int, int | None] = {} - backtick_ends: dict[int, int | None] = {} - for executor in re.finditer(_SC3_SUBSTITUTION_EXECUTOR, content, re.MULTILINE): - backtick = content[executor.end() - 1] == "`" - start = executor.end() - (1 if backtick else 2) - limit = min(len(content), start + _MAX_SHELL_LITERAL_CHARS) - skip = _skip_backtick_substitution if backtick else _skip_command_substitution - end = skip( - content, - start, - limit, - end_cache=backtick_ends if backtick else substitution_ends, - ) - # The parse limit cannot prove later decoder output is outside this wrapper. - body_end = len(content) if end is None else end - 1 - if substitution_ranges and executor.end() <= substitution_ranges[-1][1]: - previous_start, previous_end = substitution_ranges[-1] - substitution_ranges[-1] = (previous_start, max(previous_end, body_end)) - else: - substitution_ranges.append((executor.end(), body_end)) - if end is None: - break - substitution_starts = [start for start, _ in substitution_ranges] + checksum_overridden = _SC3_CHECKSUM_OVERRIDE.search(content) is not None decoded: list[tuple[int, str]] = [] for pattern in (_SC3_PIPED_LITERAL, _SC3_HERE_STRING_LITERAL): for match in pattern.finditer(content): - decoder_start = match.start("decoder") - index = bisect_right(substitution_starts, decoder_start) - 1 - in_execution_substitution = index >= 0 and decoder_start < substitution_ranges[index][1] - if _SC3_DECODED_DATA_SINK.match(content, match.end()) is not None: - continue if ( - not in_execution_substitution - and pipeline_executor.match(content, decoder_start) is None + not checksum_overridden + and _SC3_DECODED_DATA_SINK.match(content, match.end()) is not None ): continue literal = match.group("literal") diff --git a/tests/unit/test_patterns_new.py b/tests/unit/test_patterns_new.py index a97f0b87d..1a187cc7d 100644 --- a/tests/unit/test_patterns_new.py +++ b/tests/unit/test_patterns_new.py @@ -2700,12 +2700,6 @@ def test_sc3_shell_decode_to_interpreter(self, command: str) -> None: [ pytest.param(f"echo {_SHELL_FETCH_B64} | base64 -d > payload.sh", id="to_file"), pytest.param(f"echo {_SHELL_FETCH_B64} | base64 -d | sha256sum", id="sha256sum"), - pytest.param(f"echo {_SHELL_FETCH_B32} | base32 -d | sha256sum", id="base32_data"), - pytest.param(f"echo {_SHELL_FETCH_HEX} | xxd -r -p | sha256sum", id="hex_data"), - pytest.param( - f"echo {_SHELL_FETCH_B64} | openssl base64 -d -A | sha256sum", id="openssl_data" - ), - pytest.param(f"base64 -d <<< '{_SHELL_FETCH_B64}' | sha256sum", id="here_string_data"), pytest.param(f"echo {_SHELL_FETCH_B64} | base64 -d | ssh host 'cat > p'", id="ssh"), pytest.param( 'echo "$TOKEN" | cut -d. -f2 | base64 -d | python3 -m json.tool', id="json_tool" @@ -2742,11 +2736,55 @@ def test_sc3_shell_decode_to_interpreter(self, command: str) -> None: pytest.param("tar -xzf - -C . < <(base64 -d bundle.b64)", id="dot_argument"), ], ) - def test_shell_decode_without_execution_is_not_sc2_or_sc3(self, command: str) -> None: + def test_shell_decode_without_execution_is_not_sc3(self, command: str) -> None: + findings = sc_mod.analyze(command + "\n", "setup.sh", "shell") + + assert not [f for f in findings if f.rule_id == "SC3"] + + @pytest.mark.parametrize( + "command", + [ + pytest.param(f"echo {_SHELL_FETCH_B64} | base64 -d | sha256sum", id="base64"), + pytest.param(f"echo {_SHELL_FETCH_B32} | base32 -d | sha256sum", id="base32"), + pytest.param(f"echo {_SHELL_FETCH_HEX} | xxd -r -p | sha256sum", id="hex"), + pytest.param( + f"echo {_SHELL_FETCH_B64} | openssl base64 -d -A | sha256sum", id="openssl" + ), + pytest.param(f"base64 -d <<< '{_SHELL_FETCH_B64}' | sha256sum", id="here_string"), + ], + ) + def test_checksum_only_literal_is_not_sc2_or_sc3(self, command: str) -> None: findings = sc_mod.analyze(command + "\n", "setup.sh", "shell") assert not [f for f in findings if f.rule_id in {"SC2", "SC3"}] + @pytest.mark.parametrize( + "override", + [ + "sha256sum() { sh; }", + "sha256sum ( ) { sh; }", + "sha256sum" + " " * 17 + "() { sh; }", + "function sha256sum { sh; }", + "function sha256sum() { sh; }", + "function" + " " * 30 + "sha256sum { sh; }", + "'sha256sum'() { sh; }", + "alias sha256sum=sh", + "alias" + " " * 24 + "sha256sum=sh", + "alias 'sha256sum=sh'", + 'alias "sha256sum=sh"', + "alias 'sha256sum'=sh", + 'alias "sha256sum"=sh', + "alias -- sha256sum=sh", + "alias x=y sha256sum=sh", + ], + ) + def test_shadowed_checksum_keeps_decoded_sc2(self, override: str) -> None: + content = override + f"\necho {_SHELL_FETCH_B64} | base64 -d | sha256sum\n" + + findings = sc_mod.analyze(content, "setup.sh", "shell") + + assert any(f.rule_id == "SC2" and f.matched_text == _SHELL_FETCH for f in findings) + @pytest.mark.parametrize("separator", ["\n", "; ", " && "]) @pytest.mark.parametrize("data_first", [True, False]) def test_only_executed_literal_produces_sc2(self, separator: str, data_first: bool) -> None: @@ -2810,6 +2848,22 @@ def test_execution_substitution_keeps_data_only_literal_out_of_sc2(self, wrapper f'eval "$(echo {_SHELL_FETCH_B64} | base64 -d > payload.sh; cat payload.sh)"', id="substitution_with_file_round_trip", ), + pytest.param( + f"echo {_SHELL_FETCH_B64} | base64 -d > payload.sh; sh payload.sh", + id="saved_then_executed", + ), + pytest.param( + f"echo {_SHELL_FETCH_B64} | base64 -d | tee payload.sh | sha256sum; sh payload.sh", + id="saved_before_checksum_then_executed", + ), + pytest.param( + f"echo {_SHELL_FETCH_B64} | base64 -d > payload.sh | sha256sum; sh payload.sh", + id="redirected_before_checksum_then_executed", + ), + pytest.param( + f"echo {_SHELL_FETCH_B64} | base64 -d | sha256sum-wrapper", + id="unknown_checksum_named_wrapper", + ), pytest.param( "eval \"$(echo ZWNobyBoZWxsbwo= | base64 -d; $(printf ''); " f'echo {_SHELL_FETCH_B64} | base64 -d)"', @@ -2936,24 +2990,20 @@ def test_decoded_shell_literal_payloads_reports_literal_offsets(self) -> None: ("content", "expected"), [ pytest.param( - f"echo {_SHELL_FETCH_HEX.upper()}0 | xxd -r -p | sh", - _SHELL_FETCH, - id="hex_odd_nibble", + f"echo {_SHELL_FETCH_HEX.upper()}0 | xxd -r -p", _SHELL_FETCH, id="hex_odd_nibble" ), pytest.param( - f"base32 --decode <<< '{_SHELL_FETCH_B32.rstrip('=')}' | sh", + f"base32 --decode <<< '{_SHELL_FETCH_B32.rstrip('=')}'", _SHELL_FETCH, id="base32_unpadded", ), pytest.param( - f'echo \\"{_SHELL_FETCH_B64}\\" | base64 -d | sh', - _SHELL_FETCH, - id="json_escaped_quotes", + f'echo \\"{_SHELL_FETCH_B64}\\" | base64 -d', _SHELL_FETCH, id="json_escaped_quotes" ), pytest.param( "echo " + base64.b64encode(_SHELL_FETCH.encode() + b" #\xff").decode() - + " | base64 -d | sh", + + " | base64 -d", _SHELL_FETCH + " #�", id="invalid_utf8", ), @@ -2978,7 +3028,7 @@ def test_decoded_shell_literal_payloads_decodes_variants( ], ) def test_decoded_shell_literal_payloads_skips_unusable_literals(self, content: str) -> None: - assert sc_mod._decoded_shell_literal_payloads(content + " | sh") == [] + assert sc_mod._decoded_shell_literal_payloads(content) == [] @pytest.mark.parametrize( "content", From 885e619043f9f833c33702d72f64fdc0400eda43 Mon Sep 17 00:00:00 2001 From: yashrajbasav Date: Thu, 15 Oct 2026 14:52:29 +0530 Subject: [PATCH 4/4] fix(scan): normalize checksum context before suppressing decoded findings Signed-off-by: yashrajbasav --- .../analyzers/static_patterns_supply_chain.py | 29 ++++++++------- tests/unit/test_patterns_new.py | 35 +++++++++++++++++-- 2 files changed, 49 insertions(+), 15 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py index 044611f59..0b5aa3f21 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py +++ b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py @@ -334,11 +334,6 @@ def _decoded_literal_xor_calls(content: str) -> list[tuple[int, str]]: r"(?:[ \t]{1,16}-[A-Za-z0-9-]{1,16}){0,4}[ \t]{0,16}" r"\|(?!\|)&?[ \t]{0,16}sha256sum(?=$|[\s;&|)`])" ) -_SC3_CHECKSUM_OVERRIDE = re.compile( - r"\bfunction[ \t]++['\"]?sha256sum\b" - r"|(? list[tuple[int, str]]: @@ -346,21 +341,29 @@ def _decoded_shell_literal_payloads(content: str) -> list[tuple[int, str]]: Only an inline echo/printf argument or a here-string is decoded, as base64, base32 or (for xxd) hex; an ``openssl enc`` cipher is not applied. - Direct checksum consumers without a local override are data-only; other - output flows remain conservative because a saved payload can be executed later. + Direct checksum consumers are exempt only in a simple shell context without + alias, function or dynamic-command syntax. Other flows remain conservative + because a saved payload can be executed later. Variables, files and downloads stay opaque. Invalid UTF-8 becomes U+FFFD, so a stray byte cannot hide the command. Returns ``(offset, text)`` pairs in source order. """ - checksum_overridden = _SC3_CHECKSUM_OVERRIDE.search(content) is not None + checksum_exemption_allowed: bool | None = None decoded: list[tuple[int, str]] = [] for pattern in (_SC3_PIPED_LITERAL, _SC3_HERE_STRING_LITERAL): for match in pattern.finditer(content): - if ( - not checksum_overridden - and _SC3_DECODED_DATA_SINK.match(content, match.end()) is not None - ): - continue + if _SC3_DECODED_DATA_SINK.match(content, match.end()) is not None: + if checksum_exemption_allowed is None: + # Normalize once: escaped or quoted words can still configure aliases. + tokens = _shell_tokens(content) + checksum_exemption_allowed = tokens is not None and not any( + token in {"alias", "function", "eval", "source", "."} + or token.startswith("sha256sum=") + or any(character.isspace() or character in "()" for character in token) + for token in tokens + ) + if checksum_exemption_allowed: + continue literal = match.group("literal") decoder = match.group("decoder").lower() try: diff --git a/tests/unit/test_patterns_new.py b/tests/unit/test_patterns_new.py index 1a187cc7d..2a486ebba 100644 --- a/tests/unit/test_patterns_new.py +++ b/tests/unit/test_patterns_new.py @@ -2776,6 +2776,10 @@ def test_checksum_only_literal_is_not_sc2_or_sc3(self, command: str) -> None: 'alias "sha256sum"=sh', "alias -- sha256sum=sh", "alias x=y sha256sum=sh", + r"alias sha256su\m=sh", + r"a\lias sha256su\m=sh", + "a'li'as sha256su'm'=sh", + "alias x=y sha'256'sum=sh", ], ) def test_shadowed_checksum_keeps_decoded_sc2(self, override: str) -> None: @@ -2785,6 +2789,23 @@ def test_shadowed_checksum_keeps_decoded_sc2(self, override: str) -> None: assert any(f.rule_id == "SC2" and f.matched_text == _SHELL_FETCH for f in findings) + @pytest.mark.parametrize( + "setup", + [ + "eval setup", + "source aliases.sh", + ". aliases.sh", + "helper() { :; }", + "printf 'unclosed", + ], + ) + def test_uncertain_command_identity_keeps_decoded_sc2(self, setup: str) -> None: + content = setup + f"\necho {_SHELL_FETCH_B64} | base64 -d | sha256sum\n" + + findings = sc_mod.analyze(content, "setup.sh", "shell") + + assert any(f.rule_id == "SC2" and f.matched_text == _SHELL_FETCH for f in findings) + @pytest.mark.parametrize("separator", ["\n", "; ", " && "]) @pytest.mark.parametrize("data_first", [True, False]) def test_only_executed_literal_produces_sc2(self, separator: str, data_first: bool) -> None: @@ -2798,13 +2819,13 @@ def test_only_executed_literal_produces_sc2(self, separator: str, data_first: bo assert not any(f.rule_id == "SC2" for f in findings) @pytest.mark.parametrize("wrapper", ['eval "$(%s)"', "eval `%s`", "bash <(%s)"]) - def test_execution_substitution_keeps_data_only_literal_out_of_sc2(self, wrapper: str) -> None: + def test_execution_substitution_keeps_conservative_decoded_sc2(self, wrapper: str) -> None: body = f"echo {_SHELL_FETCH_B64} | base64 -d | sha256sum; echo ZWNobyBoZWxsbwo= | base64 -d" findings = sc_mod.analyze(wrapper % body, "setup.sh", "shell") assert any(f.rule_id == "SC3" for f in findings) - assert not any(f.rule_id == "SC2" for f in findings) + assert any(f.rule_id == "SC2" and f.matched_text == _SHELL_FETCH for f in findings) @pytest.mark.parametrize( "command", @@ -2964,6 +2985,12 @@ def test_shell_decode_mentions_in_skill_markdown_are_not_sc3(self, body: str) -> {"SC2"}, id="newline_ends_stage", ), + pytest.param( + r"a\lias sha256su\m=sh; " + f"echo {_SHELL_FETCH_B64} | base64 -d | sha256sum", + {"SC2"}, + id="fragmented_alias", + ), ], ) def test_shell_decode_in_hook_config(self, command: str, expected: set[str]) -> None: @@ -3034,6 +3061,10 @@ def test_decoded_shell_literal_payloads_skips_unusable_literals(self, content: s "content", [ pytest.param("echo QUFB | base64 -d " * 12_000, id="decoders_without_sink"), + pytest.param( + f"echo {_SHELL_FETCH_B64} | base64 -d | sha256sum\n" * 2_000, + id="repeated_checksum_pipelines", + ), pytest.param("eval $(echo " * 20_000 + "base64 -d", id="unclosed_substitutions"), pytest.param("eval $(" + "a " * 128_000 + "base64 -d", id="long_substitution"), pytest.param("echo " + "A" * 256_000 + " | base64 -d | sh", id="long_literal"),