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/analyzers/static_patterns_supply_chain.py b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py index 27f8f0864..0b5aa3f21 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py +++ b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py @@ -330,6 +330,10 @@ 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]]: @@ -337,13 +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 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_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 _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/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 diff --git a/tests/unit/test_patterns_new.py b/tests/unit/test_patterns_new.py index ca37fb85b..2a486ebba 100644 --- a/tests/unit/test_patterns_new.py +++ b/tests/unit/test_patterns_new.py @@ -2741,6 +2741,92 @@ def test_shell_decode_without_execution_is_not_sc3(self, command: str) -> None: 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", + 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: + 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( + "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: + 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_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 any(f.rule_id == "SC2" and f.matched_text == _SHELL_FETCH for f in findings) + @pytest.mark.parametrize( "command", [ @@ -2755,6 +2841,55 @@ 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( + 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)"', + id="second_decoder_after_nested_substitution", + ), ], ) def test_sc2_shell_literal_decoded_command(self, command: str) -> None: @@ -2765,6 +2900,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") @@ -2840,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: @@ -2910,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"),