Repository: kedro-org/kedro
File: kedro/framework/cli/micropkg.py (Python)

Code before:

    try:
        return fsspec.filesystem(protocol, **fs_args_config)
    except Exception as exc:  # noqa: broad-except
        # Specified protocol is not supported by `fsspec`
        # or requires extra dependencies
        click.secho(str(exc), fg="red")
        click.secho("Trying to use 'pip download'...", fg="red")
        return None


def _is_within_directory(directory: Path, target: Path) -> bool:
    abs_directory = directory.resolve()
    abs_target = target.resolve()
    return abs_directory in abs_target.parents


def safe_extract(tar: tarfile.TarFile, path: Path) -> None:
    safe_members = []
    for member in tar.getmembers():
        member_path = path / member.name
        if not _is_within_directory(path, member_path):
            # noqa: broad-exception-raised
            raise Exception("Failed to safely extract tar file.")
        safe_members.append(member)
    tar.extractall(path, members=safe_members)  # nosec B202


def _unpack_sdist(location: str, destination: Path, fs_args: str | None) -> None:
    filesystem = _get_fsspec_filesystem(location, fs_args)

    if location.endswith(".tar.gz") and filesystem and filesystem.exists(location):
        with filesystem.open(location) as fs_file:
            with tarfile.open(fileobj=fs_file, mode="r:gz") as tar_file:
                safe_extract(tar_file, destination)
    else:
        python_call(
            "pip",
            [
                "download",
                "--no-deps",
                "--no-binary",  # the micropackaging expects an sdist,
                ":all:",  # wheels are not supported
                "--dest",
                str(destination),
                location,
            ],
        )
        sdist_file = list(destination.glob("*.tar.gz"))
        # `--no-deps --no-binary :all:` should fetch only one source distribution file,
        # and CLI should fail if that's not the case.

Review comment (suggestion):
> The [B202:tarfile_unsafe_members](https://bandit.readthedocs.io/en/latest/plugins/b202_tarfile_unsafe_members.html) documentation says to pass a callable as the members argument but that’s not supported in [the official type signature](https://github.com/python/typeshed/blob/1d7f0d087b36d07ce34edca5ab119fdaabf7aed7/stdlib/tarfile.pyi#L295) and not implemented in [CPython stdlib](https://github.com/python/cpython/blob/025995feadaeebeef5d808f2564f0fd65b704ea5/Lib/tarfile.py#L2254C5-L2254C5).

https://github.com/PyCQA/bandit/issues/1038

Please add a link to bandit#1038 for future clarity

Code after:

    try:
        return fsspec.filesystem(protocol, **fs_args_config)
    except Exception as exc:  # noqa: broad-except
        # Specified protocol is not supported by `fsspec`
        # or requires extra dependencies
        click.secho(str(exc), fg="red")
        click.secho("Trying to use 'pip download'...", fg="red")
        return None


def _is_within_directory(directory: Path, target: Path) -> bool:
    abs_directory = directory.resolve()
    abs_target = target.resolve()
    return abs_directory in abs_target.parents


def safe_extract(tar: tarfile.TarFile, path: Path) -> None:
    safe_members = []
    for member in tar.getmembers():
        member_path = path / member.name
        if not _is_within_directory(path, member_path):
            # noqa: broad-exception-raised
            raise Exception("Failed to safely extract tar file.")
        safe_members.append(member)
    tar.extractall(path, members=safe_members)  # nosec B202
    # The nosec is still required because bandit still flags this.
    # Related issue: https://github.com/PyCQA/bandit/issues/1038


def _unpack_sdist(location: str, destination: Path, fs_args: str | None) -> None:
    filesystem = _get_fsspec_filesystem(location, fs_args)

    if location.endswith(".tar.gz") and filesystem and filesystem.exists(location):
        with filesystem.open(location) as fs_file:
            with tarfile.open(fileobj=fs_file, mode="r:gz") as tar_file:
                safe_extract(tar_file, destination)
    else:
        python_call(
            "pip",
            [
                "download",
                "--no-deps",
                "--no-binary",  # the micropackaging expects an sdist,
                ":all:",  # wheels are not supported
                "--dest",
                str(destination),
                location,
            ],
        )
        sdist_file = list(destination.glob("*.tar.gz"))