From 3c91fbd92abc2dbcb8a9b344eb585c57791f5447 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Sat, 11 Jul 2026 15:17:11 -0600 Subject: [PATCH] fix(spp): harden GPU attestation leg and shared tarball extraction Follow up on the self-audit of the NVIDIA GPU attestation leg. Tar extraction now rejects symlink and hardlink members whose link target escapes the destination and passes filter=data as the stdlib backstop. Apply the guard to all three copies of _safe_extract_tarball in rfdetr, parakeet, and local because it is a security primitive; consolidating those copies remains deferred. nvattest_install no longer trusts a partial or stale cache. _installed now requires a parsed sidecar whose archive_sha256 and version match the current pinned spec, so crash-interrupted or re-pinned installs re-verify instead of running an unrecorded binary. The sidecar temp path is also captured before writing so write failure cannot leak it. classify_nvattest_result now requires a non-bool int returncode equal to 0, closing the same bool/int fail-closed gap already fixed for result_code. The purity scanner now bans dangerous os calls, including system, popen, remove/unlink, rmdir, mkdir/makedirs, rename/replace, and the exec/spawn families, so allowing os.environ in binary.py cannot smuggle process execution or destructive writes into a pure file. Remove dead code: drop the unused build_nvattest_version_command and reduce NvattestAcceptance to the claim field actually consumed. The overall-EAT veto still runs. --- solstone/think/providers/local_install.py | 19 ++- solstone/think/providers/nvattest_install.py | 23 ++- solstone/think/providers/parakeet_install.py | 19 ++- solstone/think/providers/rfdetr_install.py | 19 ++- .../think/services/spp_attest/nvgpu/binary.py | 10 -- .../think/services/spp_attest/nvgpu/claims.py | 23 +-- tests/services/test_spp_attest_nvgpu.py | 15 ++ tests/services/test_spp_attest_purity.py | 60 +++++++- tests/test_nvattest_install.py | 125 ++++++++++++++++ tests/test_provider_tarball_extract.py | 134 ++++++++++++++++++ 10 files changed, 417 insertions(+), 30 deletions(-) create mode 100644 tests/test_nvattest_install.py create mode 100644 tests/test_provider_tarball_extract.py diff --git a/solstone/think/providers/local_install.py b/solstone/think/providers/local_install.py index dd69b642e..993196e0b 100644 --- a/solstone/think/providers/local_install.py +++ b/solstone/think/providers/local_install.py @@ -376,7 +376,24 @@ def _safe_extract_tarball(tarball: Path, dest: Path) -> None: "archive_path_traversal", f"Unsafe tar member path: {member.name}", ) - archive.extractall(dest) + if member.issym() or member.islnk(): + linkname = Path(member.linkname) + if linkname.is_absolute(): + raise LocalProviderError( + "archive_path_traversal", + f"Unsafe tar link target: {member.name} -> {member.linkname}", + ) + link_base = target.parent if member.issym() else dest_resolved + link_target = (link_base / linkname).resolve() + if ( + link_target != dest_resolved + and dest_resolved not in link_target.parents + ): + raise LocalProviderError( + "archive_path_traversal", + f"Unsafe tar link target: {member.name} -> {member.linkname}", + ) + archive.extractall(dest, filter="data") def _find_extracted_binary(dest: Path, binary_name: str) -> Path: diff --git a/solstone/think/providers/nvattest_install.py b/solstone/think/providers/nvattest_install.py index 1701a62d4..d37205392 100644 --- a/solstone/think/providers/nvattest_install.py +++ b/solstone/think/providers/nvattest_install.py @@ -112,7 +112,7 @@ def install_nvattest( """Download, verify, and install nvattest into the journal provider cache.""" root = cache_root(journal_path) - if not force and _installed(root): + if not force and _installed(root, spec): return root archive = _archive_path(spec, journal_path) @@ -137,10 +137,25 @@ def install_nvattest( archive.unlink(missing_ok=True) -def _installed(root: Path) -> bool: +def _has_runtime_layout(root: Path) -> bool: return (root / "bin" / "nvattest").is_file() and (root / "lib").is_dir() +def _installed(root: Path, spec: NvattestArchiveSpec) -> bool: + if not _has_runtime_layout(root): + return False + try: + data = json.loads((root / SIDECAR_NAME).read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return False + if not isinstance(data, dict): + return False + return ( + data.get("archive_sha256") == spec.sha256 + and data.get("version") == spec.version + ) + + def _archive_path( spec: NvattestArchiveSpec, journal_path: str | Path | None = None, @@ -212,7 +227,7 @@ def _safe_extract_nvattest_tarball(tarball: Path, dest: Path) -> None: def _find_extracted_root(extract_dir: Path) -> Path: - if _installed(extract_dir): + if _has_runtime_layout(extract_dir): return extract_dir matches = [ path @@ -264,6 +279,6 @@ def _write_sidecar(path: Path, record: NvattestInstallRecord) -> None: with tempfile.NamedTemporaryFile( "w", dir=path.parent, delete=False, encoding="utf-8" ) as handle: - handle.write(record.to_json()) tmp_path = Path(handle.name) + handle.write(record.to_json()) tmp_path.replace(path) diff --git a/solstone/think/providers/parakeet_install.py b/solstone/think/providers/parakeet_install.py index 398045efe..7c6eb8185 100644 --- a/solstone/think/providers/parakeet_install.py +++ b/solstone/think/providers/parakeet_install.py @@ -317,7 +317,24 @@ def _safe_extract_tarball(tarball: Path, dest: Path) -> None: "archive_path_traversal", f"Unsafe tar member path: {member.name}", ) - archive.extractall(dest) + if member.issym() or member.islnk(): + linkname = Path(member.linkname) + if linkname.is_absolute(): + raise ParakeetProviderError( + "archive_path_traversal", + f"Unsafe tar link target: {member.name} -> {member.linkname}", + ) + link_base = target.parent if member.issym() else dest_resolved + link_target = (link_base / linkname).resolve() + if ( + link_target != dest_resolved + and dest_resolved not in link_target.parents + ): + raise ParakeetProviderError( + "archive_path_traversal", + f"Unsafe tar link target: {member.name} -> {member.linkname}", + ) + archive.extractall(dest, filter="data") def _find_extracted_binary(dest: Path, binary_name: str) -> Path: diff --git a/solstone/think/providers/rfdetr_install.py b/solstone/think/providers/rfdetr_install.py index d393c1e51..8b518cf9c 100644 --- a/solstone/think/providers/rfdetr_install.py +++ b/solstone/think/providers/rfdetr_install.py @@ -323,7 +323,24 @@ def _safe_extract_tarball(tarball: Path, dest: Path) -> None: "archive_path_traversal", f"Unsafe tar member path: {member.name}", ) - archive.extractall(dest) + if member.issym() or member.islnk(): + linkname = Path(member.linkname) + if linkname.is_absolute(): + raise RfdetrInstallError( + "archive_path_traversal", + f"Unsafe tar link target: {member.name} -> {member.linkname}", + ) + link_base = target.parent if member.issym() else dest_resolved + link_target = (link_base / linkname).resolve() + if ( + link_target != dest_resolved + and dest_resolved not in link_target.parents + ): + raise RfdetrInstallError( + "archive_path_traversal", + f"Unsafe tar link target: {member.name} -> {member.linkname}", + ) + archive.extractall(dest, filter="data") def _find_extracted_binary(dest: Path, binary_name: str) -> Path: diff --git a/solstone/think/services/spp_attest/nvgpu/binary.py b/solstone/think/services/spp_attest/nvgpu/binary.py index a26501ce4..3dac097d7 100644 --- a/solstone/think/services/spp_attest/nvgpu/binary.py +++ b/solstone/think/services/spp_attest/nvgpu/binary.py @@ -84,13 +84,3 @@ def build_nvattest_attest_command( return NvattestCommand( argv=argv, env={**os.environ, "LD_LIBRARY_PATH": str(lib_dir)} ) - - -def build_nvattest_version_command(*, nvattest_dir: Path) -> NvattestCommand: - """Build the nvattest version command.""" - - binary, lib_dir = locate_nvattest(nvattest_dir) - return NvattestCommand( - argv=[str(binary), "version"], - env={**os.environ, "LD_LIBRARY_PATH": str(lib_dir)}, - ) diff --git a/solstone/think/services/spp_attest/nvgpu/claims.py b/solstone/think/services/spp_attest/nvgpu/claims.py index 90b5d25a8..169baa4f2 100644 --- a/solstone/think/services/spp_attest/nvgpu/claims.py +++ b/solstone/think/services/spp_attest/nvgpu/claims.py @@ -39,9 +39,7 @@ class GpuAppraisal: @dataclass(frozen=True, slots=True) class NvattestAcceptance: - stdout: dict[str, Any] claim: dict[str, Any] - overall_payload: dict[str, Any] @dataclass(frozen=True, slots=True) @@ -108,13 +106,22 @@ def classify_nvattest_result( result_message = _required_stdout(stdout_obj, "result_message") except _ClaimReject as exc: return NvattestRejection("gpu_appraisal_failed", str(exc)) + returncode_is_green = ( + isinstance(returncode, int) + and not isinstance(returncode, bool) + and returncode == 0 + ) result_code_is_green = ( isinstance(result_code, int) and not isinstance(result_code, bool) and result_code == 0 ) result_message_is_green = isinstance(result_message, str) and result_message == "Ok" - if returncode != 0 or not result_code_is_green or not result_message_is_green: + if ( + not returncode_is_green + or not result_code_is_green + or not result_message_is_green + ): return NvattestRejection( _reason_for_result_code(result_code), ( @@ -140,17 +147,11 @@ def classify_nvattest_result( ) try: - overall_payload = _parse_overall_eat( - _required_stdout(stdout_obj, "detached_eat") - ) + _parse_overall_eat(_required_stdout(stdout_obj, "detached_eat")) _check_claim(claim, owner_nonce.hex()) except _ClaimReject as exc: return NvattestRejection("gpu_appraisal_failed", str(exc)) - return NvattestAcceptance( - stdout=stdout_obj, - claim=claim, - overall_payload=overall_payload, - ) + return NvattestAcceptance(claim=claim) def build_gpu_appraisal( diff --git a/tests/services/test_spp_attest_nvgpu.py b/tests/services/test_spp_attest_nvgpu.py index 6fb0113dc..8601e8a59 100644 --- a/tests/services/test_spp_attest_nvgpu.py +++ b/tests/services/test_spp_attest_nvgpu.py @@ -345,6 +345,21 @@ def test_stderr_does_not_influence_acceptance( assert result.driver_version == "595.71.05" +def test_bool_false_returncode_rejects( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + with pytest.raises(GpuAppraisalError) as exc_info: + _run_appraisal_with_stdout( + monkeypatch, + tmp_path, + _stdout("positive"), + returncode=False, + ) + + assert exc_info.value.reason == "gpu_appraisal_failed" + + def test_nonce_is_written_to_evidence_and_argv( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/services/test_spp_attest_purity.py b/tests/services/test_spp_attest_purity.py index 21ace85d6..b96c17397 100644 --- a/tests/services/test_spp_attest_purity.py +++ b/tests/services/test_spp_attest_purity.py @@ -66,6 +66,18 @@ BANNED_WRITE_ATTRS = { "atomic_replace", } BANNED_WRITE_NAMES = {"atomic_write", "atomic_replace"} +BANNED_OS_ATTRS = { + "makedirs", + "mkdir", + "popen", + "remove", + "rename", + "replace", + "rmdir", + "system", + "unlink", +} +BANNED_OS_ATTR_PREFIXES = ("exec", "spawn") WRITE_MODE_CHARS = frozenset({"w", "a", "x", "+"}) @@ -83,8 +95,9 @@ def test_spp_attest_package_stays_pure_python_read_only_except_nvgpu_appraise() findings: list[str] = [] for path in files: tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + os_names = _imported_module_names(tree, "os") for node in ast.walk(tree): - findings.extend(_scan_node(path, node)) + findings.extend(_scan_node(path, node, os_names=os_names)) assert findings == [] @@ -98,6 +111,7 @@ def test_nvgpu_appraise_impurity_is_narrow() -> None: assert "tempfile" in imports findings: list[str] = [] + os_names = _imported_module_names(tree, "os") for node in ast.walk(tree): findings.extend( _scan_node( @@ -106,6 +120,7 @@ def test_nvgpu_appraise_impurity_is_narrow() -> None: banned_import_roots=APPRAISE_BANNED_IMPORT_ROOTS, banned_write_attrs=BANNED_WRITE_ATTRS - {"unlink"}, ban_solstone_utils=True, + os_names=os_names, ) ) @@ -197,6 +212,18 @@ def test_purity_scanner_bans_aliased_shutil_import() -> None: assert findings == ["snippet.py:1: banned import shutil"] +def test_purity_scanner_bans_aliased_os_system_call() -> None: + tree = ast.parse("import os as operating_system\noperating_system.system('x')\n") + os_names = _imported_module_names(tree, "os") + findings = [ + finding + for node in ast.walk(tree) + for finding in _scan_node(Path("snippet.py"), node, os_names=os_names) + ] + + assert findings == ["snippet.py:2: banned os.system call"] + + def _scan_node( path: Path, node: ast.AST, @@ -204,7 +231,9 @@ def _scan_node( banned_import_roots: set[str] = BANNED_IMPORT_ROOTS, banned_write_attrs: set[str] = BANNED_WRITE_ATTRS, ban_solstone_utils: bool = False, + os_names: set[str] | None = None, ) -> list[str]: + os_names = os_names or set() if isinstance(node, ast.Import): return _scan_import(path, node, banned_import_roots) if isinstance(node, ast.ImportFrom): @@ -215,7 +244,7 @@ def _scan_node( ban_solstone_utils, ) if isinstance(node, ast.Call): - return _scan_call(path, node, banned_write_attrs) + return _scan_call(path, node, banned_write_attrs, os_names) return [] @@ -248,6 +277,8 @@ def _scan_import_from( ): findings.append(f"{path}:{node.lineno}: banned import from {module}") for alias in node.names: + if module == "os" and _is_banned_os_attr(alias.name): + findings.append(f"{path}:{node.lineno}: banned import from os {alias.name}") if alias.name in BANNED_WRITE_NAMES: findings.append(f"{path}:{node.lineno}: banned write helper {alias.name}") return findings @@ -257,9 +288,12 @@ def _scan_call( path: Path, node: ast.Call, banned_write_attrs: set[str], + os_names: set[str], ) -> list[str]: func = node.func if isinstance(func, ast.Attribute): + if _is_banned_os_call(func, os_names): + return [f"{path}:{node.lineno}: banned os.{func.attr} call"] if _is_shutil_which(func): return [f"{path}:{node.lineno}: banned shutil.which call"] if _is_json_dump(func): @@ -274,6 +308,18 @@ def _scan_call( return [] +def _is_banned_os_attr(attr: str) -> bool: + return attr in BANNED_OS_ATTRS or attr.startswith(BANNED_OS_ATTR_PREFIXES) + + +def _is_banned_os_call(func: ast.Attribute, os_names: set[str]) -> bool: + return ( + _is_banned_os_attr(func.attr) + and isinstance(func.value, ast.Name) + and func.value.id in os_names + ) + + def _is_shutil_which(func: ast.Attribute) -> bool: return ( func.attr == "which" @@ -300,6 +346,16 @@ def _import_roots(tree: ast.AST) -> set[str]: return roots +def _imported_module_names(tree: ast.AST, module: str) -> set[str]: + names: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name == module: + names.add(alias.asname or module) + return names + + def _open_uses_write_mode(node: ast.Call) -> bool: mode_node = None if len(node.args) >= 2: diff --git a/tests/test_nvattest_install.py b/tests/test_nvattest_install.py new file mode 100644 index 000000000..107673618 --- /dev/null +++ b/tests/test_nvattest_install.py @@ -0,0 +1,125 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import hashlib +import json +import shutil +import tarfile +from pathlib import Path + +import pytest + +from solstone.think.providers import nvattest_install + + +def test_install_nvattest_reinstalls_partial_cache_without_sidecar( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = nvattest_install.cache_root(tmp_path) + (root / "bin").mkdir(parents=True) + (root / "bin" / "nvattest").write_text("stale\n", encoding="utf-8") + (root / "lib").mkdir() + + spec = _fixture_spec(tmp_path) + calls: list[Path] = [] + + def fake_download(_url: str, dest: Path, _expected_sha256: str) -> None: + calls.append(dest) + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(tmp_path / spec.archive_name, dest) + + monkeypatch.setattr(nvattest_install, "_download_file", fake_download) + + installed = nvattest_install.install_nvattest(spec=spec, journal_path=tmp_path) + + assert installed == root + assert len(calls) == 1 + assert (root / "bin" / "nvattest").read_text(encoding="utf-8") == "new\n" + assert (root / "lib" / "libnvat.so").is_symlink() + assert (root / "lib" / "libnvat.so").readlink() == Path("libnvat.so.1") + sidecar = json.loads( + (root / nvattest_install.SIDECAR_NAME).read_text(encoding="utf-8") + ) + assert sidecar["archive_sha256"] == spec.sha256 + assert sidecar["version"] == spec.version + + +def test_install_nvattest_reinstalls_stale_sidecar( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = nvattest_install.cache_root(tmp_path) + (root / "bin").mkdir(parents=True) + (root / "bin" / "nvattest").write_text("old\n", encoding="utf-8") + (root / "lib").mkdir() + (root / nvattest_install.SIDECAR_NAME).write_text( + json.dumps({"archive_sha256": "old", "version": "0.0.0"}) + "\n", + encoding="utf-8", + ) + + spec = _fixture_spec(tmp_path) + calls: list[Path] = [] + + def fake_download(_url: str, dest: Path, _expected_sha256: str) -> None: + calls.append(dest) + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(tmp_path / spec.archive_name, dest) + + monkeypatch.setattr(nvattest_install, "_download_file", fake_download) + + nvattest_install.install_nvattest(spec=spec, journal_path=tmp_path) + + assert len(calls) == 1 + assert (root / "bin" / "nvattest").read_text(encoding="utf-8") == "new\n" + + +def test_install_nvattest_valid_sidecar_is_noop( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = nvattest_install.cache_root(tmp_path) + (root / "bin").mkdir(parents=True) + (root / "bin" / "nvattest").write_text("installed\n", encoding="utf-8") + (root / "lib").mkdir() + spec = nvattest_install.NvattestArchiveSpec( + version="1.0.0", + url="https://example.invalid/nvattest.tar.gz", + archive_name="nvattest.tar.gz", + sha256="abc123", + ) + (root / nvattest_install.SIDECAR_NAME).write_text( + json.dumps({"archive_sha256": spec.sha256, "version": spec.version}) + "\n", + encoding="utf-8", + ) + + monkeypatch.setattr( + nvattest_install, + "_download_file", + lambda *_args, **_kwargs: pytest.fail("download should not run"), + ) + + assert nvattest_install.install_nvattest(spec=spec, journal_path=tmp_path) == root + + +def _fixture_spec(tmp_path: Path) -> nvattest_install.NvattestArchiveSpec: + archive_name = "nvattest-fixture.tar.gz" + source = tmp_path / "source" / "nvattest-fixture" + (source / "bin").mkdir(parents=True) + (source / "bin" / "nvattest").write_text("new\n", encoding="utf-8") + (source / "lib").mkdir() + (source / "lib" / "libnvat.so.1").write_text("library\n", encoding="utf-8") + (source / "lib" / "libnvat.so").symlink_to("libnvat.so.1") + (source / "LICENSE").write_text("license\n", encoding="utf-8") + + archive_path = tmp_path / archive_name + with tarfile.open(archive_path, "w:gz") as archive: + archive.add(source, arcname=source.name) + return nvattest_install.NvattestArchiveSpec( + version="9.9.9", + url="https://example.invalid/nvattest-fixture.tar.gz", + archive_name=archive_name, + sha256=hashlib.sha256(archive_path.read_bytes()).hexdigest(), + ) diff --git a/tests/test_provider_tarball_extract.py b/tests/test_provider_tarball_extract.py new file mode 100644 index 000000000..d68573367 --- /dev/null +++ b/tests/test_provider_tarball_extract.py @@ -0,0 +1,134 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import io +import tarfile +from collections.abc import Callable +from pathlib import Path + +import pytest + +from solstone.think.providers import local_install, parakeet_install, rfdetr_install +from solstone.think.providers.local import LocalProviderError + +Extractor = Callable[[Path, Path], None] + + +@pytest.mark.parametrize( + ("extract", "error_type"), + [ + (rfdetr_install._safe_extract_tarball, rfdetr_install.RfdetrInstallError), + ( + parakeet_install._safe_extract_tarball, + parakeet_install.ParakeetProviderError, + ), + (local_install._safe_extract_tarball, LocalProviderError), + ], +) +def test_safe_extract_tarball_rejects_symlink_escape( + tmp_path: Path, + extract: Extractor, + error_type: type[Exception], +) -> None: + outside = tmp_path / "outside" + outside.mkdir() + tarball = tmp_path / "symlink-escape.tar" + with tarfile.open(tarball, "w") as archive: + _add_dir(archive, "payload") + _add_symlink(archive, "payload/lib", str(outside)) + _add_file(archive, "payload/lib/evil.txt", b"owned\n") + + with pytest.raises(error_type) as exc_info: + extract(tarball, tmp_path / "dest") + + assert getattr(exc_info.value, "reason_code") == "archive_path_traversal" + assert not (outside / "evil.txt").exists() + + +@pytest.mark.parametrize( + ("extract", "error_type"), + [ + (rfdetr_install._safe_extract_tarball, rfdetr_install.RfdetrInstallError), + ( + parakeet_install._safe_extract_tarball, + parakeet_install.ParakeetProviderError, + ), + (local_install._safe_extract_tarball, LocalProviderError), + ], +) +def test_safe_extract_tarball_rejects_hardlink_escape( + tmp_path: Path, + extract: Extractor, + error_type: type[Exception], +) -> None: + outside = tmp_path / "outside" + outside.mkdir() + tarball = tmp_path / "hardlink-escape.tar" + with tarfile.open(tarball, "w") as archive: + _add_dir(archive, "payload") + _add_hardlink(archive, "payload/libnvat.so", str(outside / "target")) + + with pytest.raises(error_type) as exc_info: + extract(tarball, tmp_path / "dest") + + assert getattr(exc_info.value, "reason_code") == "archive_path_traversal" + + +@pytest.mark.parametrize( + "extract", + [ + rfdetr_install._safe_extract_tarball, + parakeet_install._safe_extract_tarball, + local_install._safe_extract_tarball, + ], +) +def test_safe_extract_tarball_allows_internal_relative_nvattest_symlink( + tmp_path: Path, + extract: Extractor, +) -> None: + tarball = tmp_path / "internal-symlink.tar" + with tarfile.open(tarball, "w") as archive: + _add_dir(archive, "payload") + _add_dir(archive, "payload/lib") + _add_file(archive, "payload/lib/libnvat.so.1", b"library\n") + _add_symlink(archive, "payload/lib/libnvat.so", "libnvat.so.1") + + dest = tmp_path / "dest" + extract(tarball, dest) + + link = dest / "payload" / "lib" / "libnvat.so" + assert link.is_symlink() + assert link.readlink() == Path("libnvat.so.1") + assert link.resolve() == dest / "payload" / "lib" / "libnvat.so.1" + + +def _add_dir(archive: tarfile.TarFile, name: str) -> None: + info = tarfile.TarInfo(name) + info.type = tarfile.DIRTYPE + info.mode = 0o755 + archive.addfile(info) + + +def _add_file(archive: tarfile.TarFile, name: str, data: bytes) -> None: + info = tarfile.TarInfo(name) + info.size = len(data) + info.mode = 0o644 + archive.addfile(info, io.BytesIO(data)) + + +def _add_symlink(archive: tarfile.TarFile, name: str, linkname: str) -> None: + info = tarfile.TarInfo(name) + info.type = tarfile.SYMTYPE + info.mode = 0o777 + info.linkname = linkname + archive.addfile(info) + + +def _add_hardlink(archive: tarfile.TarFile, name: str, linkname: str) -> None: + info = tarfile.TarInfo(name) + info.type = tarfile.LNKTYPE + info.mode = 0o644 + info.linkname = linkname + archive.addfile(info) -- 2.51.2