61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
"""Archive adapter safety tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import tarfile
|
|
import zipfile
|
|
|
|
import pytest
|
|
|
|
from flow.adapters.archive import ArchiveClient
|
|
from flow.adapters.filesystem import FileSystem
|
|
from flow.core.errors import FlowError
|
|
|
|
|
|
def test_extract_tar_uses_safe_member_paths(tmp_path):
|
|
archive = tmp_path / "ok.tar.gz"
|
|
with tarfile.open(archive, "w:gz") as tar:
|
|
content = b"hello"
|
|
info = tarfile.TarInfo("pkg/bin/tool")
|
|
info.size = len(content)
|
|
tar.addfile(info, io.BytesIO(content))
|
|
|
|
target = tmp_path / "extract"
|
|
ArchiveClient(FileSystem()).extract(archive, target)
|
|
|
|
assert (target / "pkg" / "bin" / "tool").read_text() == "hello"
|
|
|
|
|
|
def test_rejects_tar_member_parent_traversal(tmp_path):
|
|
archive = tmp_path / "bad.tar.gz"
|
|
with tarfile.open(archive, "w:gz") as tar:
|
|
content = b"bad"
|
|
info = tarfile.TarInfo("../escape")
|
|
info.size = len(content)
|
|
tar.addfile(info, io.BytesIO(content))
|
|
|
|
with pytest.raises(FlowError, match="escapes"):
|
|
ArchiveClient(FileSystem()).extract(archive, tmp_path / "extract")
|
|
|
|
|
|
def test_rejects_zip_member_parent_traversal(tmp_path):
|
|
archive = tmp_path / "bad.zip"
|
|
with zipfile.ZipFile(archive, "w") as zf:
|
|
zf.writestr("../escape", "bad")
|
|
|
|
with pytest.raises(FlowError, match="escapes"):
|
|
ArchiveClient(FileSystem()).extract(archive, tmp_path / "extract")
|
|
|
|
|
|
def test_rejects_tar_symlink_members(tmp_path):
|
|
archive = tmp_path / "bad-link.tar"
|
|
with tarfile.open(archive, "w") as tar:
|
|
info = tarfile.TarInfo("pkg/link")
|
|
info.type = tarfile.SYMTYPE
|
|
info.linkname = "/etc/passwd"
|
|
tar.addfile(info)
|
|
|
|
with pytest.raises(FlowError, match="member type"):
|
|
ArchiveClient(FileSystem()).extract(archive, tmp_path / "extract")
|