跳到正文
原文
Google AI:DEV 作者专属(RSS)· Jonathan·· 2 小时前AI 评分61

AI Leak Watch:为 Python AI 包构建导入期发布门

AI Leak Watch: Build an Import-Time Release Gate for Python AI Packages

AI 导读

作者借鉴 Socket 对 MemTensor 被入侵包的分析,指出 import memos 本身即可触发捆绑的原生程序,测试边界应前移到导入之前。

正文

If your test begins after import, you may already be late.

Socket's analysis of the compromised MemTensor packages found that import memos was enough to start a bundled native program. The affected OpenClaw plugin also launched the program when the gateway started and again during memory recall, passing the user's prompt in an environment variable.

In plain English, loading the library was itself an execution trigger. The application did not have to call a risky feature first.

That changes the test boundary. A normal functional test asks whether the package returns the right result. A supply-chain gate must first ask what the package did before the test called any feature.

This article turns that question into a small release gate for Python packages used in AI systems. The gate checks the downloaded artifact, traces a cold import, plants harmless credentials, and fails on behavior the package has no reason to perform.

Keep the incident claims precise

The MemTensor case is serious, but the evidence has limits. Socket reported three malicious npm releases and one malicious PyPI release. Each carried cross-platform Go binaries named sckit. Its researchers found code that started the binaries in the background with the host environment, searched developer locations for credentials, and referenced attacker-controlled infrastructure.

Socket also said its findings came from static analysis. It had not executed the samples, had not confirmed how publishing access was obtained, and had not reported a victim count. The OpenClaw plugin passed prompt text to the hidden process. That establishes an unauthorized data path, but it does not prove that every prompt was successfully exfiltrated.

Those caveats do not weaken the release decision. A package that silently starts an undeclared executable during import should fail before anyone argues about the payload's final success.

Download the artifact without loading it

Do not run this investigation on a developer laptop or a normal CI runner. Use a disposable environment with no real credentials, no repository write token, and no package-publishing token.

Start with the exact registry artifact. Do not inspect only the repository tag.

mkdir -p evidence/wheels evidence/npm

python -m pip download \
  --no-deps \
  --only-binary=:all: \
  --dest evidence/wheels \
  'your-package==2.4.1'

cd evidence/npm
npm pack '@your-scope/your-plugin@1.7.0' --ignore-scripts

Record the registry URL, version, digest, download time, and the source commit that the release claims to represent. Provenance is useful only when the consumer verifies it against an expected source and builder. A provenance badge is not a substitute for that comparison.

Gate 1: Compare the package with the last approved release

Package diffs should be routine. I care most about new native binaries, executable permission changes, new build backends, and abrupt growth in unpacked size.

The following script compares a wheel, npm tarball, or source archive with the last approved artifact. The two-times growth threshold is an example policy, not a universal rule. Set it from the normal history of the package.

#!/usr/bin/env python3
import sys
import tarfile
import zipfile
from dataclasses import dataclass
from hashlib import sha256
from pathlib import Path

MACH_O = {
    b"\xfe\xed\xfa\xce", b"\xce\xfa\xed\xfe",
    b"\xfe\xed\xfa\xcf", b"\xcf\xfa\xed\xfe",
    b"\xca\xfe\xba\xbe", b"\xbe\xba\xfe\xca",
}


@dataclass(frozen=True)
class FileFacts:
    size: int
    digest: str
    native: bool
    executable: bool


def is_native(data: bytes) -> bool:
    return (
        data.startswith(b"\x7fELF")
        or data.startswith(b"MZ")
        or data[:4] in MACH_O
    )


def facts(data: bytes, mode: int) -> FileFacts:
    return FileFacts(
        size=len(data),
        digest=sha256(data).hexdigest(),
        native=is_native(data),
        executable=bool(mode & 0o111),
    )


def inspect_archive(path: Path) -> dict[str, FileFacts]:
    files: dict[str, FileFacts] = {}

    if zipfile.is_zipfile(path):
        with zipfile.ZipFile(path) as archive:
            for item in archive.infolist():
                if item.is_dir():
                    continue
                with archive.open(item) as stream:
                    data = stream.read()
                mode = item.external_attr >> 16
                files[item.filename] = facts(data, mode)
        return files

    if tarfile.is_tarfile(path):
        with tarfile.open(path) as archive:
            for item in archive.getmembers():
                if not item.isfile():
                    continue
                stream = archive.extractfile(item)
                data = stream.read() if stream else b""
                files[item.name] = facts(data, item.mode)
        return files

    raise ValueError(f"Unsupported archive: {path}")


baseline = inspect_archive(Path(sys.argv[1]))
candidate = inspect_archive(Path(sys.argv[2]))

new_files = sorted(candidate.keys() - baseline.keys())
changed_native = sorted(
    name
    for name, current in candidate.items()
    if current.native
    and (name not in baseline or baseline[name].digest != current.digest)
)
new_executable = sorted(
    name
    for name, current in candidate.items()
    if current.executable
    and (name not in baseline or not baseline[name].executable)
)
old_size = sum(item.size for item in baseline.values())
new_size = sum(item.size for item in candidate.values())
growth = new_size / old_size if old_size else float("inf")

print(f"unpacked growth: {growth:.2f}x")
print("new files:", *new_files, sep="\n  ")

failures = []
if changed_native:
    failures.append(f"new or changed native binaries: {changed_native}")
if new_executable:
    failures.append(f"new executable permissions: {new_executable}")
if growth > 2.0:
    failures.append(f"unpacked size grew {growth:.2f}x")

if failures:
    raise SystemExit("RELEASE BLOCKED\n" + "\n".join(failures))

Run it against adjacent approved and candidate versions:

python tools/gate_artifact.py \
  evidence/wheels/your_package-2.4.0-py3-none-any.whl \
  evidence/wheels/your_package-2.4.1-py3-none-any.whl

This gate is deliberately narrow. It does not decide that every binary is malicious. It decides that a new or replaced binary requires an owner, source, build record, expected hash, and review before release. The digest comparison matters because an attacker can replace a binary without changing its filename.

Gate 2: Trace the quiet path

Static inspection tells you what arrived. A cold-import trace tells you what ran.

On a disposable Linux runner, install the candidate into an isolated virtual environment. Give the test a temporary home directory containing fake secrets. Then trace process execution, outbound connection attempts, and file opens while importing the module.

#!/usr/bin/env bash
set -euo pipefail

module=${1:?usage: trace_import.sh MODULE}
trace_dir=$(mktemp -d)
trap 'rm -rf "$trace_dir"' EXIT

test_home="$trace_dir/home"
mkdir -p "$test_home/.aws" "$test_home/.ssh"
printf '%s\n' '//registry.npmjs.org/:_authToken=CANARY_NPM_73f8' \
  > "$test_home/.npmrc"
printf '%s\n' '[default]' 'aws_access_key_id=CANARY_AWS_4ca1' \
  > "$test_home/.aws/credentials"
printf '%s\n' 'CANARY_SSH_PRIVATE_KEY_9b62' \
  > "$test_home/.ssh/id_ed25519"

env -i \
  PATH="$PATH" \
  HOME="$test_home" \
  MODULE="$module" \
  API_KEY='CANARY_API_KEY_c012' \
  timeout 15s strace -ff -qq -s 4096 \
    -e trace=execve,connect,openat \
    -o "$trace_dir/import" \
    .venv/bin/python -I -c \
      'import importlib, os; importlib.import_module(os.environ["MODULE"])'

python tools/assert_import_trace.py "$trace_dir" "$test_home"

The assertion script treats any non-Python child executable, internet socket, or read of the planted files as a release failure:

#!/usr/bin/env python3
import re
import sys
from pathlib import Path

trace_dir = Path(sys.argv[1])
test_home = str(Path(sys.argv[2]))
lines = []
for path in trace_dir.glob("import*"):
    lines.extend(path.read_text(errors="replace").splitlines())

exec_paths = []
for line in lines:
    match = re.search(r'execve\("([^"]+)"', line)
    if match:
        exec_paths.append(Path(match.group(1)).name)

allowed = {"python", "python3", "python3.13"}
unexpected_exec = [name for name in exec_paths if name not in allowed]
internet_connect = [
    line for line in lines
    if "connect(" in line and ("AF_INET" in line or "AF_INET6" in line)
]
secret_reads = [
    line for line in lines
    if "openat(" in line
    and test_home in line
    and any(name in line for name in (".npmrc", "credentials", "id_ed25519"))
]

failures = {
    "unexpected executables": unexpected_exec,
    "internet connection attempts": internet_connect,
    "canary credential reads": secret_reads,
}
failures = {name: values for name, values in failures.items() if values}

if failures:
    for name, values in failures.items():
        print(f"{name}:")
        for value in values:
            print(f"  {value}")
    raise SystemExit("RELEASE BLOCKED")

The test does not need a successful connection. An undeclared attempt is enough to stop the release. Run the same probe with the network disabled as a containment control, but keep the trace. A blocked call is still evidence about intent and behavior.

strace is Linux-specific. On macOS, use an isolated VM with Endpoint Security telemetry or an equivalent process and network monitor. On Windows, use a disposable VM with Process Monitor and network capture. Python introspection alone is not a universal integrity control because native code and detached processes can run outside the interpreter's view.

Gate 3: Exercise the AI lifecycle hooks

Import isn't the only window. An AI package also receives real data during memory recall, tool calls, telemetry flushes, and error handling. Any of those paths can carry a canary value out.

Run one black-box scenario for each lifecycle hook. Use a different canary string for every path. Capture subprocess environments, files, logs, DNS, and HTTP. The assertion is simple: the canary may reach only the destinations named in the design.

For a memory plugin, I would test at least these events:

  1. Start the agent gateway without sending a prompt.
  2. Load the plugin but do not call it.
  3. Recall memory with a unique canary prompt.
  4. Store a separate canary value.
  5. Restart the gateway and repeat the recall.

Do not use one canary everywhere. Distinct values tell you which hook created the unexpected path.

Put the checks in one release decision

Define the decision before the pipeline fails. I would start with this matrix and tune only the package-size threshold from the project's release history.

Check Expected result Release blocker
Artifact provenance Registry artifact ties to the approved source and builder Missing or mismatched source, builder, or digest
Artifact diff Binary hashes, executable bits, and build files match the approved change New or changed native binary, executable bit, or build hook without approval
Unpacked size Change stays within the package's normal range Unexplained growth beyond the project threshold
Cold import Import starts only the expected interpreter Any undeclared child process
Credential canaries Package ignores credentials outside its documented scope Read of a planted token, key, or credential file
Lifecycle canaries Prompt and memory values reach only approved components Canary appears in an undeclared process, file, log, or endpoint
Network trace Only documented destinations are contacted Undeclared DNS lookup or connection attempt

Do not average these signals into a score. A package does not become acceptable because six checks passed after it launched an unexplained binary.

OWASP mapping

Under the OWASP GenAI LLM Top 10 for 2026, the primary mapping is LLM04:2026 Supply Chain. The likely impact also maps to LLM02:2026 Sensitive Information Disclosure because the code targeted credentials and exposed prompt text to an untrusted process.

I would not map this primarily to prompt injection or excessive agency. The model did not need to make a bad decision. The package executed before the model mattered.

The practical standard

A dependency scanner can tell you that a version is known to be bad. It cannot prove that the artifact matches the source you reviewed, explain why a wheel grew twenty times larger, or show that an import started a detached process.

The approach is the same across all three gates: treat each phase as a trust boundary, define the expected behavior before the test runs, and block on any deviation. Installation, import, startup, and the AI lifecycle hooks are all surfaces where a bad package can act. A gate that covers one but not the others just makes the attacker pick a different surface.

If the package runs code before your first test step, that behavior belongs inside the test plan.

Learn to test AI supply chains

My course, AI Security Testing: LLM-03 Supply Chain Testing, covers practical ways to verify third-party components, release artifacts, and dependency behavior. The course title follows the 2025 OWASP numbering. Supply Chain moved to LLM04 in the 2026 list.

View all of my AI security testing courses.

References

来源:Google AI:DEV 作者专属(RSS) · dev.to