"""remote/transfer.py の仕様ベーステスト（tar ストリームの書き出しと安全な展開）． 仕様出典: - 実装サマリー 仕様 1（push: PAX 形式・UTF-8 ファイル名） - 実装サマリー 仕様 4（pull: tarfile の filter="data"・要求パス以外のメンバーを拒否 ＝パストラバーサル対策） テスト方針: - ssh は使わず，BytesIO をストリームの代わりにする．展開先は tmp_path の実ディレクトリ """ from __future__ import annotations import gzip import io import tarfile from pathlib import Path import pytest from sm_tias_lightsim.remote.files import PushEntry from sm_tias_lightsim.remote.transfer import extract_tar_stream, write_tar_stream # --------------------------------------------------------------------------- # ヘルパ # --------------------------------------------------------------------------- def _entry(root: Path, rel: str, data: bytes) -> PushEntry: path = root / rel path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) return PushEntry(path=path, arcname=rel, size=len(data)) def _tar_bytes(members: list[tuple[str, bytes | None, dict]], *, gz: bool = False) -> bytes: """(name, data, attrs) の並びから tar を作る．data=None ならディレクトリ等（attrs で指定）．""" buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w:gz" if gz else "w", format=tarfile.PAX_FORMAT) as tar: for name, data, attrs in members: info = tarfile.TarInfo(name) for key, value in attrs.items(): setattr(info, key, value) if data is not None: info.size = len(data) tar.addfile(info, io.BytesIO(data)) else: tar.addfile(info) return buf.getvalue() def _file(name: str, data: bytes = b"x") -> tuple[str, bytes, dict]: return (name, data, {"mode": 0o644}) def _all_files(root: Path) -> set[str]: return {p.relative_to(root).as_posix() for p in root.rglob("*") if p.is_file()} # --------------------------------------------------------------------------- # write_tar_stream（push） # --------------------------------------------------------------------------- class TestWriteTarStream: def test_writes_pax_with_utf8_names_and_contents(self, tmp_path): """PAX 形式で書き，日本語ファイル名と内容がそのまま読めること．""" entries = [ _entry(tmp_path, "configs/meshes/前側.stl", b"solid front\n"), _entry(tmp_path, "src/sm_tias_lightsim/a.py", b"x = 1\n"), ] buf = io.BytesIO() write_tar_stream(entries, buf, compress=False) buf.seek(0) with tarfile.open(fileobj=buf, mode="r:", encoding="utf-8") as tar: assert tar.format == tarfile.PAX_FORMAT names = tar.getnames() assert names == ["configs/meshes/前側.stl", "src/sm_tias_lightsim/a.py"] assert tar.extractfile("configs/meshes/前側.stl").read() == b"solid front\n" def test_pax_header_carries_utf8_path(self, tmp_path): """日本語名は PAX 拡張ヘッダ（path=UTF-8）で運ばれること． 受信側（サーバの GNU tar）の文字コード設定に依存しないため． """ entries = [_entry(tmp_path, "configs/meshes/照明ムラ補正ボード.stl", b"s")] buf = io.BytesIO() write_tar_stream(entries, buf, compress=False) assert "configs/meshes/照明ムラ補正ボード.stl".encode() in buf.getvalue() def test_returns_total_bytes_and_reports_progress(self, tmp_path): entries = [_entry(tmp_path, "a.txt", b"12345"), _entry(tmp_path, "b.txt", b"123")] calls: list[tuple[int, int]] = [] total = write_tar_stream( entries, io.BytesIO(), compress=False, progress=lambda d, t: calls.append((d, t)) ) assert total == 8 assert calls == [(5, 8), (8, 8)] def test_owner_and_mode_are_normalized(self, tmp_path): """所有者は 0，通常ファイルは 644 にそろえる（送信元 OS の stat に引きずられない）．""" entries = [_entry(tmp_path, "a.py", b"x")] buf = io.BytesIO() write_tar_stream(entries, buf, compress=False) buf.seek(0) with tarfile.open(fileobj=buf, mode="r:") as tar: info = tar.getmember("a.py") assert (info.uid, info.gid, info.uname, info.gname) == (0, 0, "", "") assert info.mode in (0o644, 0o755) def test_does_not_close_stream(self, tmp_path): """ssh の標準入力は呼び出し側が閉じる（EOF を伝える）ため，関数内では閉じないこと．""" buf = io.BytesIO() write_tar_stream([_entry(tmp_path, "a.txt", b"a")], buf, compress=False) assert not buf.closed def test_compress_writes_gzip(self, tmp_path): entries = [_entry(tmp_path, "a.txt", b"a" * 1000)] buf = io.BytesIO() write_tar_stream(entries, buf, compress=True) data = buf.getvalue() assert data[:2] == b"\x1f\x8b" with tarfile.open(fileobj=io.BytesIO(gzip.decompress(data)), mode="r:") as tar: assert tar.getnames() == ["a.txt"] def test_roundtrip_with_extract(self, tmp_path): """書き出したストリームをそのまま展開すると同じファイルが得られること．""" src = tmp_path / "src" entries = [ _entry(src, "output/sweep/前側.npy", b"\x00\x01\x02"), _entry(src, "output/sweep/sub/metrics.csv", b"a,b\n1,2\n"), ] buf = io.BytesIO() write_tar_stream(entries, buf, compress=True) buf.seek(0) dest = tmp_path / "dest" count, size = extract_tar_stream(buf, dest, ["output/sweep"]) assert (count, size) == (2, 3 + 8) assert (dest / "output/sweep/前側.npy").read_bytes() == b"\x00\x01\x02" assert (dest / "output/sweep/sub/metrics.csv").read_bytes() == b"a,b\n1,2\n" # --------------------------------------------------------------------------- # extract_tar_stream（pull） # --------------------------------------------------------------------------- class TestExtractTarStream: def test_extracts_requested_paths(self, tmp_path): data = _tar_bytes( [ ("output/foo", None, {"type": tarfile.DIRTYPE, "mode": 0o755}), _file("output/foo/a.txt", b"hello"), _file("output/foo/sub/b.txt", b"wor"), ] ) dest = tmp_path / "dest" assert extract_tar_stream(io.BytesIO(data), dest, ["output/foo"]) == (2, 8) assert (dest / "output/foo/a.txt").read_bytes() == b"hello" assert (dest / "output/foo/sub/b.txt").read_bytes() == b"wor" def test_creates_dest_dir(self, tmp_path): dest = tmp_path / "a" / "b" extract_tar_stream(io.BytesIO(_tar_bytes([_file("output/x/a")])), dest, ["output/x"]) assert (dest / "output/x/a").is_file() def test_gzip_is_detected_automatically(self, tmp_path): data = _tar_bytes([_file("output/foo/a.txt", b"z")], gz=True) dest = tmp_path / "dest" extract_tar_stream(io.BytesIO(data), dest, ["output/foo"]) assert (dest / "output/foo/a.txt").read_bytes() == b"z" def test_multiple_prefixes(self, tmp_path): data = _tar_bytes([_file("output/a/1"), _file("output/b/2")]) dest = tmp_path / "dest" assert extract_tar_stream(io.BytesIO(data), dest, ["output/a", "output/b"])[0] == 2 def test_overwrites_existing_file(self, tmp_path): """pull は手元の同名ファイルを上書きする（ENV_04 注意）．""" dest = tmp_path / "dest" (dest / "output/foo").mkdir(parents=True) (dest / "output/foo/a.txt").write_bytes(b"old") extract_tar_stream( io.BytesIO(_tar_bytes([_file("output/foo/a.txt", b"new")])), dest, ["output/foo"] ) assert (dest / "output/foo/a.txt").read_bytes() == b"new" def test_progress_is_reported(self, tmp_path): calls: list[tuple[int, int]] = [] data = _tar_bytes([_file("output/foo/a", b"12"), _file("output/foo/b", b"345")]) extract_tar_stream( io.BytesIO(data), tmp_path / "d", ["output/foo"], progress=lambda d, t: calls.append((d, t)), ) assert calls == [(2, 0), (5, 0)] def test_rejects_member_outside_requested_path(self, tmp_path): """要求していないパスのメンバーが来たら中断し，そのファイルは書かないこと．""" data = _tar_bytes([_file("output/foo/a.txt"), _file("src/sm_tias_lightsim/__init__.py")]) dest = tmp_path / "dest" with pytest.raises(tarfile.TarError): extract_tar_stream(io.BytesIO(data), dest, ["output/foo"]) assert not (dest / "src").exists() def test_prefix_match_respects_path_boundary(self, tmp_path): """`output/foo` を要求したとき `output/foobar/...` は受け入れないこと．""" data = _tar_bytes([_file("output/foobar/x")]) dest = tmp_path / "dest" with pytest.raises(tarfile.TarError): extract_tar_stream(io.BytesIO(data), dest, ["output/foo"]) assert not (dest / "output/foobar/x").exists() @pytest.mark.parametrize( "name", [ "/etc/evil.txt", # 絶対パス "../evil.txt", # 展開先の外 "output/foo/../../../evil.txt", # 要求パスで始まるが展開先の外へ出る ], ) def test_rejects_traversal_outside_dest(self, tmp_path, name): dest = tmp_path / "dest" data = _tar_bytes([_file(name, b"evil")]) with pytest.raises(tarfile.TarError): extract_tar_stream(io.BytesIO(data), dest, ["output/foo"]) assert not (tmp_path / "evil.txt").exists() assert not (dest / "evil.txt").exists() @pytest.mark.parametrize( ("name", "landing"), [ ("output/foo/../../evil.txt", "evil.txt"), ("output/foo/../bar/evil.txt", "output/bar/evil.txt"), ( "output/foo/../../src/sm_tias_lightsim/__init__.py", "src/sm_tias_lightsim/__init__.py", ), ], ) def test_rejects_dotdot_escape_from_requested_path(self, tmp_path, name, landing): """要求パスで始まっても `..` で要求パスの外（展開先の内側）へ出るメンバーは拒否すること． 仕様 4: 要求したパス以外のメンバーは拒否する（パストラバーサル対策）． 展開先はプロジェクトルートなので，これを許すと src/ 等を上書きできてしまう． """ dest = tmp_path / "dest" data = _tar_bytes([_file(name, b"evil")]) with pytest.raises(tarfile.TarError): extract_tar_stream(io.BytesIO(data), dest, ["output/foo"]) assert not (dest / landing).exists() def test_rejects_symlink_pointing_outside_dest(self, tmp_path): data = _tar_bytes( [("output/foo/link", None, {"type": tarfile.SYMTYPE, "linkname": "../../../outside"})] ) with pytest.raises(tarfile.TarError): extract_tar_stream(io.BytesIO(data), tmp_path / "dest", ["output/foo"]) def test_rejects_absolute_symlink(self, tmp_path): data = _tar_bytes( [("output/foo/link", None, {"type": tarfile.SYMTYPE, "linkname": "/etc/passwd"})] ) with pytest.raises(tarfile.TarError): extract_tar_stream(io.BytesIO(data), tmp_path / "dest", ["output/foo"]) def test_rejects_device_file(self, tmp_path): data = _tar_bytes([("output/foo/dev", None, {"type": tarfile.CHRTYPE})]) with pytest.raises(tarfile.TarError): extract_tar_stream(io.BytesIO(data), tmp_path / "dest", ["output/foo"]) def test_empty_stream_raises(self, tmp_path): with pytest.raises(tarfile.TarError): extract_tar_stream(io.BytesIO(b""), tmp_path / "dest", ["output/foo"]) def test_broken_stream_raises(self, tmp_path): with pytest.raises(tarfile.TarError): extract_tar_stream(io.BytesIO(b"not a tar" * 100), tmp_path / "dest", ["output/foo"]) # --------------------------------------------------------------------------- # 仕様 4 の強化: 要求パスの外への書き込みにつながるメンバーの拒否 # --------------------------------------------------------------------------- # 展開先（プロジェクトルート相当）に最初からある，上書きされてはならないファイル _SENTINEL_REL = "src/sm_tias_lightsim/__init__.py" _SENTINEL_DATA = b"original\n" @pytest.fixture def dest_with_sentinel(tmp_path) -> Path: dest = tmp_path / "dest" sentinel = dest / _SENTINEL_REL sentinel.parent.mkdir(parents=True) sentinel.write_bytes(_SENTINEL_DATA) return dest def _assert_nothing_outside(dest: Path, prefix: str = "output/foo") -> None: """要求パスの外に新しいファイル・リンクが無く，既存ファイルも変わっていないこと．""" sentinel = dest / _SENTINEL_REL assert sentinel.read_bytes() == _SENTINEL_DATA assert sentinel.stat().st_nlink == 1, "既存ファイルへのハードリンクが作られている" outside = [] for path in dest.rglob("*"): rel = path.relative_to(dest).as_posix() if rel == prefix or rel.startswith(prefix + "/") or rel == _SENTINEL_REL: continue if path.is_dir() and not path.is_symlink(): continue # output/・src/ 等の親ディレクトリ outside.append(rel) assert outside == [], f"要求パスの外に作られた: {outside}" class TestExtractRejectsEscapes: """仕様 4: 要求したパス以外のメンバーは拒否する（パストラバーサル対策）． リンクや名前の書き方で要求パス（output/foo）の外へ書き込めてしまうメンバーを TarError 系で拒否し，展開先の要求パスの外にファイルが作られない・変わらないこと． """ @staticmethod def _extract_rejected(dest: Path, members) -> None: with pytest.raises(tarfile.TarError): extract_tar_stream(io.BytesIO(_tar_bytes(members)), dest, ["output/foo"]) def test_hardlink_then_overwrite_is_rejected(self, dest_with_sentinel): """要求パス内のハードリンクで要求パス外の既存ファイルを指し，続く同名の通常ファイルで 上書きするストリームを拒否すること（ハードリンクは Windows でも特権なしで作れる）．""" dest = dest_with_sentinel self._extract_rejected( dest, [ ("output/foo/h", None, {"type": tarfile.LNKTYPE, "linkname": _SENTINEL_REL}), _file("output/foo/h", b"evil\n"), ], ) _assert_nothing_outside(dest) def test_hardlink_alone_to_outside_is_rejected(self, dest_with_sentinel): dest = dest_with_sentinel self._extract_rejected( dest, [("output/foo/h", None, {"type": tarfile.LNKTYPE, "linkname": _SENTINEL_REL})] ) _assert_nothing_outside(dest) assert not (dest / "output/foo/h").exists() @pytest.mark.parametrize( "linkname", ["../bar", "../../src", "../../src/sm_tias_lightsim/__init__.py", "../../../outside"], ) def test_symlink_to_outside_requested_path_is_rejected(self, dest_with_sentinel, linkname): """要求パスの外（展開先の内側・外側とも）を指す symlink は拒否すること．""" dest = dest_with_sentinel link = ("output/foo/link", None, {"type": tarfile.SYMTYPE, "linkname": linkname}) self._extract_rejected(dest, [link]) link_path = dest / "output/foo/link" assert not link_path.is_symlink() and not link_path.exists() _assert_nothing_outside(dest) @pytest.mark.parametrize("linkname", ["a.txt", "sub/b.txt", "./a.txt"]) def test_symlink_inside_requested_path_is_rejected(self, dest_with_sentinel, linkname): """展開先の内側（要求パス内）を指す symlink も拒否すること． フィルタの段階で弾くので，symlink を作る特権が無い環境でも同じ結果になる． """ dest = dest_with_sentinel members = [ _file("output/foo/a.txt", b"a"), _file("output/foo/sub/b.txt", b"b"), ("output/foo/link", None, {"type": tarfile.SYMTYPE, "linkname": linkname}), ] self._extract_rejected(dest, members) link_path = dest / "output/foo/link" assert not link_path.is_symlink() and not link_path.exists() _assert_nothing_outside(dest) def test_symlink_then_write_through_is_rejected(self, dest_with_sentinel): """symlink を置いてからそれ越しに書き込むストリームを拒否すること．""" dest = dest_with_sentinel self._extract_rejected( dest, [ ( "output/foo/lnk", None, {"type": tarfile.SYMTYPE, "linkname": "../../src/sm_tias_lightsim"}, ), _file("output/foo/lnk/__init__.py", b"evil\n"), ], ) _assert_nothing_outside(dest) @pytest.mark.parametrize( "name", [ "output/foo/..\\..\\src\\evil.py", "output/foo/..\\..\\src\\sm_tias_lightsim\\__init__.py", "output/foo/sub\\a.txt", "output\\foo\\a.txt", ], ) def test_backslash_names_are_rejected(self, dest_with_sentinel, name): """バックスラッシュ区切りの名前は拒否すること（Windows では区切りとして解釈される）．""" dest = dest_with_sentinel self._extract_rejected(dest, [_file(name, b"evil\n")]) assert not (dest / "src" / "evil.py").exists() _assert_nothing_outside(dest) @pytest.mark.parametrize( "name", [ "output/foo/./a.txt", "./output/foo/a.txt", "output/foo/.", "output/foo//a.txt", "output//foo/a.txt", "output/foo/C:/evil.txt", "output/foo/C:evil.txt", "C:/output/foo/a.txt", "C:output/foo/a.txt", ], ) def test_names_with_dot_empty_or_drive_elements_are_rejected(self, dest_with_sentinel, name): """要素に `.`・空文字（`//`）・ドライブ文字（`C:`）を含む名前は拒否すること．""" dest = dest_with_sentinel self._extract_rejected(dest, [_file(name, b"evil\n")]) _assert_nothing_outside(dest) @pytest.mark.parametrize( "name", [ "output/foo/a.txt", "output/foo/.hidden", "output/foo/a..b.txt", "output/foo/..hidden", "output/foo/sub/前側.npy", "output/foo/with space.txt", "output/foo/-dash.txt", ], ) def test_legitimate_names_are_still_accepted(self, dest_with_sentinel, name): """正当な名前（. で始まる名前・.. を含むファイル名・日本語・空白）は受け入れること． 拒否の強化で通常の pull が壊れていないことの確認（GNU tar のディレクトリ行を含む）． """ dest = dest_with_sentinel members = [ ("output/foo", None, {"type": tarfile.DIRTYPE, "mode": 0o755}), ("output/foo/sub", None, {"type": tarfile.DIRTYPE, "mode": 0o755}), _file(name, b"ok"), ] assert extract_tar_stream(io.BytesIO(_tar_bytes(members)), dest, ["output/foo"]) == (1, 2) assert (dest / name).read_bytes() == b"ok" _assert_nothing_outside(dest)