tests/pytests/functional/modules/test_archive.py - master - githubmirror/salt

tests/pytests/functional/modules/test_archive.py
433 строки13 KB
Daniel A. Wozniak
Fix archive.extracted ownership on tar archives with no directory members
16 июн 2026, 00:50
16 июн 2026, 00:504c6e761
"""
Tests for the archive state
"""

import os
import pathlib
import shutil
import tempfile
import textwrap

import attr
import pytest

import salt.utils.files
import salt.utils.path
import salt.utils.platform
import salt.utils.stringutils

try:
    import zipfile  # pylint: disable=unused-import

    HAS_ZIPFILE = True
except ImportError:
    HAS_ZIPFILE = False

pytestmark = [
    pytest.mark.windows_whitelisted,
]


@attr.s(frozen=True, slots=True)
class Archive:
    fmt = attr.ib()
    unicode_filename = attr.ib(default=False)

    path = attr.ib(init=False)
    src = attr.ib(init=False)
    src_file = attr.ib(init=False)
    archive = attr.ib(init=False)
    dst = attr.ib(init=False)
    filename = attr.ib(init=False)

    @path.default
    def _path(self):
        return pathlib.Path(tempfile.mkdtemp())

    @src.default
    def _src(self):
        return self.path / f"{self.fmt}_src_dir"

    @src_file.default
    def _src_file(self):
        return self.src / "file"

    @archive.default
    def _archive(self):
        return self.path / f"archive.{self.fmt}"

    @dst.default
    def _dst(self):
        return self.path / f"{self.fmt}_dst_dir"

    @filename.default
    def _filename(self):
        if self.unicode_filename:
            return "file®"
        return "file"

    def __attrs_post_init__(self):
        # Create source
        self.src.mkdir()
        self.dst.mkdir()

        if salt.utils.platform.is_windows():
            encoding = "utf-8"
        else:
            encoding = None

        dst_filename = self.src / self.filename
        dst_filename.write_bytes(
            salt.utils.stringutils.to_bytes(
                textwrap.dedent(
                    """\
            Compression theorem of computational complexity theory:

            Given a Gödel numbering $φ$ of the computable functions and a
            Blum complexity measure $Φ$ where a complexity class for a
            boundary function $f$ is defined as

                $\\mathrm C(f) := \\{φ_i ∈ \\mathbb R^{(1)} | (∀^∞ x) Φ_i(x) ≤ f(x)\\}$.

            Then there exists a total computable function $f$ so that for
            all $i$

                $\\mathrm{Dom}(φ_i) = \\mathrm{Dom}(φ_{f(i)})$

            and

                $\\mathrm C(φ_i) ⊊ \\mathrm{C}(φ_{f(i)})$.
        """
                ),
                encoding=encoding,
            )
        )

    def assert_artifacts_in_ret(self, ret, file_only=False, unix_sep=False):
        """
        Assert that the artifact source files are printed in the source command
        output
        """

        def normdir(path):
            normdir = os.path.normcase(os.path.abspath(str(path)))
            if salt.utils.platform.is_windows():
                # Remove the drive portion of path
                if len(normdir) >= 2 and normdir[1] == ":":
                    normdir = normdir.split(":", 1)[1]
            normdir = normdir.lstrip(os.path.sep)
            # Unzipped paths might have unix line endings
            if unix_sep:
                normdir = normdir.replace(os.path.sep, "/")
            return normdir

        # Try to find source directory and file in output lines
        dir_in_ret = None
        file_in_ret = None
        for line in ret:
            if normdir(self.src) in os.path.normcase(line) and not normdir(
                self.src_file
            ) in os.path.normcase(line):
                dir_in_ret = True
            if normdir(self.src_file) in os.path.normcase(line):
                file_in_ret = True

        # Assert number of lines, reporting of source directory and file
        assert len(ret) >= 1 if file_only else 2
        if not file_only:
            assert dir_in_ret is True
        assert file_in_ret is True

    def __enter__(self):
        return self

    def __exit__(self, *_):
        shutil.rmtree(str(self.path), ignore_errors=True)


@pytest.fixture(scope="module")
def archive(modules):
    return modules.archive


def unicode_filename_ids(value):
    return f"unicode_filename={value}"


@pytest.fixture(params=[True, False], ids=unicode_filename_ids)
def unicode_filename(request):
    return request.param


@pytest.mark.skip_on_windows
@pytest.mark.skip_if_binaries_missing("tar")
def test_tar_pack(archive, unicode_filename):
    """
    Validate using the tar function to create archives
    """
    with Archive("tar", unicode_filename=unicode_filename) as arch:
        ret = archive.tar("-cvf", str(arch.archive), sources=str(arch.src))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret)


@pytest.mark.skip_on_windows
@pytest.mark.skip_if_binaries_missing("tar")
def test_tar_unpack(archive, unicode_filename):
    """
    Validate using the tar function to extract archives
    """
    with Archive("tar", unicode_filename=unicode_filename) as arch:
        ret = archive.tar("-cvf", str(arch.archive), sources=str(arch.src))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret)

        ret = archive.tar("-xvf", str(arch.archive), dest=str(arch.dst))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret)


@pytest.mark.skip_on_windows
@pytest.mark.skip_if_binaries_missing("tar")
def test_tar_list(archive, unicode_filename):
    """
    Validate using the tar function to list archives
    """
    with Archive("tar", unicode_filename=unicode_filename) as arch:
        ret = archive.tar("-cvf", str(arch.archive), sources=str(arch.src))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret)

        ret = archive.list(str(arch.archive))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret)


@pytest.mark.skip_if_binaries_missing("gzip")
def test_gzip(archive, unicode_filename):
    """
    Validate using the gzip function
    """
    with Archive("gz", unicode_filename=unicode_filename) as arch:
        ret = archive.gzip(str(arch.src_file), options="-v")
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret, file_only=True)


@pytest.mark.skip_if_binaries_missing("gzip", "gunzip")
def test_gunzip(archive, unicode_filename):
    """
    Validate using the gunzip function
    """
    with Archive("gz", unicode_filename=unicode_filename) as arch:
        ret = archive.gzip(str(arch.src_file), options="-v")
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret, file_only=True)

        ret = archive.gunzip(str(arch.src_file) + ".gz", options="-v")
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret, file_only=True)


@pytest.mark.skip_if_binaries_missing("zip")
def test_cmd_zip(archive, unicode_filename, grains):
    """
    Validate using the cmd_zip function
    """
    with Archive("zip", unicode_filename=unicode_filename) as arch:
        ret = archive.cmd_zip(str(arch.archive), str(arch.src))
        assert isinstance(ret, list)
        if (
            "*** buffer overflow detected ***: terminated" in ret
            and grains["os"] == "Fedora"
            and int(grains["osmajorrelease"]) >= 37
        ):
            pytest.skip(
                f"Calling the system zip on {grains['os']} {grains['osmajorrelease']} "
                "is currently failing with a buffer overflow error"
            )
        arch.assert_artifacts_in_ret(ret)


@pytest.mark.skip_if_binaries_missing("zip", "unzip")
def test_cmd_unzip(archive, unicode_filename, grains):
    """
    Validate using the cmd_unzip function
    """
    with Archive("zip", unicode_filename=unicode_filename) as arch:
        ret = archive.cmd_zip(str(arch.archive), str(arch.src))
        assert isinstance(ret, list)
        if (
            "*** buffer overflow detected ***: terminated" in ret
            and grains["os"] == "Fedora"
            and int(grains["osmajorrelease"]) >= 37
        ):
            pytest.skip(
                f"Calling the system unzip on {grains['os']} {grains['osmajorrelease']} "
                "is currently failing with a buffer overflow error"
            )
        arch.assert_artifacts_in_ret(ret)

        ret = archive.cmd_unzip(str(arch.archive), str(arch.dst))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret)


@pytest.mark.skipif(not HAS_ZIPFILE, reason="Cannot find zipfile python module")
def test_zip(archive, unicode_filename):
    """
    Validate using the zip function
    """
    with Archive("zip", unicode_filename=unicode_filename) as arch:
        ret = archive.zip(str(arch.archive), str(arch.src))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret)


@pytest.mark.skipif(not HAS_ZIPFILE, reason="Cannot find zipfile python module")
def test_unzip(archive, unicode_filename):
    """
    Validate using the unzip function
    """
    with Archive("zip", unicode_filename=unicode_filename) as arch:
        ret = archive.zip(str(arch.archive), str(arch.src))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret)

        ret = archive.unzip(str(arch.archive), str(arch.dst))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret, unix_sep=False)


@pytest.mark.skip_if_binaries_missing("rar")
def test_rar(archive, unicode_filename):
    """
    Validate using the rar function
    """
    with Archive("rar", unicode_filename=unicode_filename) as arch:
        ret = archive.rar(str(arch.archive), str(arch.src))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret)


@pytest.mark.skip_if_binaries_missing("rar", "unrar")
def test_unrar(archive, unicode_filename):
    """
    Validate using the unrar function
    """
    with Archive("rar", unicode_filename=unicode_filename) as arch:
        ret = archive.rar(str(arch.archive), str(arch.src))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret)

        ret = archive.unrar(str(arch.archive), str(arch.dst))
        assert isinstance(ret, list)
        arch.assert_artifacts_in_ret(ret)


@pytest.mark.skip_on_windows
@pytest.mark.skip_if_binaries_missing("tar")
def test_tar_list_no_explicit_top_level_directory_member(archive, tmp_path):
    rel_path = pathlib.Path("archive_top/second_level/file.txt")
    srcfile = tmp_path / str(rel_path)
    srcfile.parent.mkdir(parents=True, exist_ok=True)
    srcfile.write_text("hey there")
    archive.tar(
        "-cvzf",
        str(tmp_path / "tld_test.tar.gz"),
        sources=str(rel_path.parent),
        cwd=str(tmp_path),
    )

    expected = {
        "dirs": ["archive_top/second_level/"],
        "files": ["archive_top/second_level/file.txt"],
        "links": [],
        "top_level_dirs": ["archive_top/"],
        "top_level_files": [],
        "top_level_links": [],
    }
    ret = archive.list(
        str(tmp_path / "tld_test.tar.gz"), archive_format="tar", verbose=True
    )
    assert ret == expected


@pytest.mark.skip_on_windows
@pytest.mark.skip_if_binaries_missing("tar")
def test_tar_list_with_similar_top_level_dirs(archive, tmp_path):
    rel_paths = (
        "archive_top/second_level/file.txt",
        "archives/second_level/file.txt",
    )

    for rel_path in rel_paths:
        srcfile = tmp_path / str(pathlib.Path(rel_path))
        srcfile.parent.mkdir(parents=True, exist_ok=True)
        srcfile.write_text("hey there")

    archive.tar(
        "-cvzf",
        str(tmp_path / "tld_test.tar.gz"),
        sources=["archive_top", "archives"],
        cwd=str(tmp_path),
    )

    expected = {
        "dirs": [
            "archive_top/",
            "archive_top/second_level/",
            "archives/",
            "archives/second_level/",
        ],
        "files": [
            "archive_top/second_level/file.txt",
            "archives/second_level/file.txt",
        ],
        "links": [],
        "top_level_dirs": ["archive_top/", "archives/"],
        "top_level_files": [],
        "top_level_links": [],
    }
    ret = archive.list(
        str(tmp_path / "tld_test.tar.gz"), archive_format="tar", verbose=True
    )
    assert ret == expected


@pytest.mark.skip_on_windows
def test_tar_list_no_directory_members_only_files_68227(archive, tmp_path):
    """
    Regression test for issue 68227.

    Some tar archives (e.g. Oracle's GraalVM JDK distribution) contain only
    file members and no explicit directory members, even though all files
    share a common top-level directory prefix. ``archive.list`` should still
    report that common prefix as the top-level directory so that
    ``archive.extracted`` can enforce user/group ownership on it.
    """
    import io
    import tarfile

    archive_path = tmp_path / "no_dir_members.tar.gz"
    # Build the archive directly with tarfile so we can guarantee that no
    # directory members are written. ``tar -cvzf`` always emits an entry for
    # the containing directory.
    with tarfile.open(str(archive_path), "w:gz") as tar:
        for rel in ("LICENSE.txt", "GRAALVM-README.md", "bin/java"):
            data = b"x"
            info = tarfile.TarInfo(name="graalvm-jdk-21.0.8+12.1/" + rel)
            info.size = len(data)
            tar.addfile(info, fileobj=io.BytesIO(data))

    # Sanity-check the archive only has file members
    with tarfile.open(str(archive_path)) as tar:
        assert not any(m.isdir() for m in tar.getmembers())

    ret = archive.list(str(archive_path), archive_format="tar", verbose=True)
    assert ret["dirs"] == []
    assert "graalvm-jdk-21.0.8+12.1/" in ret["top_level_dirs"], (
        "Expected the common file prefix to be reported as a top-level "
        "directory so that archive.extracted can enforce ownership on it; "
        f"got top_level_dirs={ret['top_level_dirs']!r}"
    )