fix: avoid checksum false positives and preserve archive error origins

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)

src/skillspector/nodes/analyzers/static_patterns_supply_chain.py

@@ -330,20 +330,40 @@ 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.
 
     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:

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(

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

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"),