"""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)