brain-of-reese/tests/unit/test_archive_upload.py

"""Unit: phase 49 upload settings + the safe archive unpack utility.

Covers ``app.rag.archive_upload`` end to end:

* the name-derivation matrix (suffix stripping incl. the compound
  ``.tar.gz``, case handling, and the rejection list — ``..``,
  separators, control chars, empty stems);
* safe extraction — valid zip/tar.gz archives extract byte-identically,
  while zip-slip, absolute members, escaping symlinks/hardlinks, device
  members, and the extracted-byte cap all raise
  :class:`ArchiveUploadError` **and** leave no partial target behind;
* ``swap_in`` — fresh, in-place replace with full content replacement
  (no interleave, no ``.old-`` leftovers), and restore-on-failure;
* the two new settings (``upload_dir`` / ``upload_max_mb``) with
  env overrides and the fail-loud ``<= 0`` validator.
"""
from __future__ import annotations

import io
import os
import stat
import tarfile
import zipfile
from pathlib import Path
from typing import Any

import pytest
from pydantic import ValidationError

from app.config import Settings
from app.rag.archive_upload import (
    ARCHIVE_SUFFIXES,
    ArchiveUploadError,
    _link_target_resolved,  # pyright: ignore[reportPrivateUsage]
    _member_dest,  # pyright: ignore[reportPrivateUsage]
    archive_source_name,
    swap_in,
    unpack_archive,
)


def _settings(**kwargs: Any) -> Settings:
    """Build Settings without reading a .env file (deterministic tests)."""
    kwargs.setdefault("_env_file", None)
    return Settings(**kwargs)  # pyright: ignore[reportCallIssue]  (kwarg exists at runtime)


# ---------------------------------------------------------------------------
# Settings (phase 49, task 01)
# ---------------------------------------------------------------------------


def test_upload_settings_defaults(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("BOR_UPLOAD_DIR", raising=False)
    monkeypatch.delenv("BOR_UPLOAD_MAX_MB", raising=False)
    s = _settings()
    # Deliberately separate from the git checkouts (``sources_dir``).
    assert s.upload_dir == "~/bor-sources/uploads"
    assert s.sources_dir == "~/bor-sources"
    assert s.upload_max_mb == 512


def test_upload_settings_env_overrides(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("BOR_UPLOAD_DIR", "/data/bor/uploads")
    monkeypatch.setenv("BOR_UPLOAD_MAX_MB", "128")
    s = _settings()
    assert s.upload_dir == "/data/bor/uploads"
    assert s.upload_max_mb == 128


@pytest.mark.parametrize("value", ["0", "-1", "-512"])
def test_upload_max_mb_rejects_zero_and_negative(
    monkeypatch: pytest.MonkeyPatch, value: str
) -> None:
    """``<= 0`` would reject every upload — the validator fails loudly at
    startup (the ``agent_max_rounds`` pattern)."""
    monkeypatch.setenv("BOR_UPLOAD_MAX_MB", value)
    with pytest.raises(ValidationError, match="upload_max_mb"):
        _settings()


# ---------------------------------------------------------------------------
# archive_source_name — the derivation matrix
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    ("filename", "expected"),
    [
        ("homelab.tar.gz", "homelab"),
        ("notes.tgz", "notes"),
        ("a.zip", "a"),
        ("x.tar", "x"),
        # Case-insensitive suffix match; the stem keeps its case (the name
        # becomes a folder name on a case-sensitive Linux FS).
        ("upper.TAR.GZ", "upper"),
        ("Homelab.Zip", "Homelab"),
        # ONE compound strip — never a double strip (``.tar.gz`` matches
        # before ``.tar`` would; there is no ``.gz`` suffix at all).
        ("a.tar.gz", "a"),
        # Only the LAST suffix is stripped.
        ("a.zip.zip", "a.zip"),
        ("y.tar.tgz", "y.tar"),
        ("café.tar", "café"),
    ],
)
def test_archive_source_name_strips_one_suffix(filename: str, expected: str) -> None:
    assert archive_source_name(filename) == expected


def test_archive_suffixes_are_longest_first() -> None:
    # ``.tar.gz`` must precede ``.tar`` or ``a.tar.gz`` would yield
    # ``a.tar``.
    assert ARCHIVE_SUFFIXES == (".tar.gz", ".tgz", ".zip", ".tar")


@pytest.mark.parametrize(
    "filename",
    [
        "",  # empty
        "tar.gz",  # empty stem (bare suffix, no leading dot)
        "tgz",  # empty stem (bare suffix)
        "zip",  # empty stem (bare suffix)
        ".zip",  # empty stem (hidden-file suffix)
        "..tar.gz",  # ``..`` after the strip
        "..",  # no suffix, ``..`` stem
        ".",  # no suffix, ``.`` stem
        "a/b.tar",  # forward-separator path
        "a\\b.tar",  # backslash path
        "a\tb.zip",  # tab control character
        "a\x00b.tar",  # NUL control character
        "a\x1bb.tar",  # escape-sequence control character
    ],
)
def test_archive_source_name_rejects_unsafe_names(filename: str) -> None:
    with pytest.raises(ArchiveUploadError):
        archive_source_name(filename)


# ---------------------------------------------------------------------------
# Archive builders (deterministic, in-memory)
# ---------------------------------------------------------------------------


def _make_zip(
    path: Path,
    files: dict[str, bytes] | None = None,
    dirs: tuple[str, ...] = (),
    extra_attrs: dict[str, int] | None = None,
) -> None:
    with zipfile.ZipFile(path, "w") as zf:
        for name in dirs:
            info = zipfile.ZipInfo(name if name.endswith("/") else name + "/")
            info.external_attr = (0o40755 << 16)
            zf.writestr(info, b"")
        for name, data in (files or {}).items():
            info = zipfile.ZipInfo(name)
            info.external_attr = (0o100644 << 16)
            zf.writestr(info, data)
        for name, attr in (extra_attrs or {}).items():
            info = zipfile.ZipInfo(name)
            info.external_attr = attr
            zf.writestr(info, b"")


def _make_tar(
    path: Path,
    spec: list[tuple[str, bytes, str]],
    gz: bool = False,
) -> None:
    """``spec`` entries: ``(name, payload, kind)`` with kind one of
    ``f`` (file), ``d`` (dir), ``sym`` (symlink, payload = target),
    ``lnk`` (hardlink, payload = target), ``chr`` (char device),
    ``fifo`` (FIFO)."""
    mode = "w:gz" if gz else "w"
    with tarfile.open(path, mode) as tf:
        for name, payload, kind in spec:
            ti = tarfile.TarInfo(name)
            if kind == "f":
                ti.size = len(payload)
                ti.mode = 0o644
                tf.addfile(ti, io.BytesIO(payload))
            elif kind == "d":
                ti.type = tarfile.DIRTYPE
                ti.mode = 0o755
                tf.addfile(ti)
            elif kind in ("sym", "lnk"):
                ti.type = tarfile.SYMTYPE if kind == "sym" else tarfile.LNKTYPE
                ti.linkname = payload.decode()
                tf.addfile(ti)
            elif kind == "chr":
                ti.type = tarfile.CHRTYPE
                ti.devmajor, ti.devminor = 1, 3
                tf.addfile(ti)
            elif kind == "fifo":
                ti.type = tarfile.FIFOTYPE
                tf.addfile(ti)


# ---------------------------------------------------------------------------
# unpack_archive — valid archives extract byte-identically
# ---------------------------------------------------------------------------


def test_unpack_zip_valid_nested(tmp_path: Path) -> None:
    archive = tmp_path / "notes.zip"
    _make_zip(
        archive,
        files={"a/b.txt": b"hello\n", "c.txt": b"x" * 50, "a/deep/n.md": b"# deep"},
        dirs=("a", "a/deep"),
    )
    target = tmp_path / "out"
    unpack_archive(archive, target, 10_000)
    assert (target / "a/b.txt").read_bytes() == b"hello\n"
    assert (target / "c.txt").read_bytes() == b"x" * 50
    assert (target / "a/deep/n.md").read_bytes() == b"# deep"
    assert (target / "a").is_dir()
    assert (target / "a/deep").is_dir()


def test_unpack_zip_unknown_mode_entries_treated_as_files(tmp_path: Path) -> None:
    """Windows-made / ``writestr``-style zips carry no Unix mode bits
    (external_attr 0) — decided by the member name, not rejected."""
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w") as zf:
        zf.writestr("plain.txt", b"data")  # default external_attr = 0
    archive = tmp_path / "plain.zip"
    archive.write_bytes(buf.getvalue())
    target = tmp_path / "out"
    unpack_archive(archive, target, 10_000)
    assert (target / "plain.txt").read_bytes() == b"data"


def test_unpack_tar_gz_valid(tmp_path: Path) -> None:
    archive = tmp_path / "notes.tar.gz"
    _make_tar(
        archive,
        [
            ("dir/", b"", "d"),
            ("dir/a.md", b"# hi", "f"),
            ("top.txt", b"t", "f"),
        ],
        gz=True,
    )
    target = tmp_path / "out"
    unpack_archive(archive, target, 10_000)
    assert (target / "dir/a.md").read_bytes() == b"# hi"
    assert (target / "top.txt").read_bytes() == b"t"
    assert (target / "dir").is_dir()


def test_unpack_plain_tar_valid(tmp_path: Path) -> None:
    """``r:*`` handles uncompressed ``.tar`` too."""
    archive = tmp_path / "notes.tar"
    _make_tar(archive, [("only.txt", b"plain tar", "f")], gz=False)
    target = tmp_path / "out"
    unpack_archive(archive, target, 10_000)
    assert (target / "only.txt").read_bytes() == b"plain tar"


def test_unpack_internal_symlink_and_hardlink_allowed(tmp_path: Path) -> None:
    """Links that stay INSIDE the unpack directory are fine (the spec
    resolves the target and rejects only escapes)."""
    archive = tmp_path / "links.tar"
    _make_tar(
        archive,
        [
            ("sub/", b"", "d"),
            ("sub/data.txt", b"inner", "f"),
            ("alias", b"sub/data.txt", "sym"),
            ("dup", b"sub/data.txt", "lnk"),
        ],
        gz=False,
    )
    target = tmp_path / "out"
    unpack_archive(archive, target, 10_000)
    assert target.joinpath("alias").is_symlink()
    assert target.joinpath("alias").read_bytes() == b"inner"
    assert target.joinpath("dup").read_bytes() == b"inner"


def test_extracted_bytes_exactly_at_cap_passes(tmp_path: Path) -> None:
    """The cap bounds EXCEEDING bytes — landing exactly on it is OK."""
    archive = tmp_path / "exact.zip"
    _make_zip(archive, files={"f.txt": b"a" * 10})
    target = tmp_path / "out"
    unpack_archive(archive, target, 10)
    assert (target / "f.txt").read_bytes() == b"a" * 10


# ---------------------------------------------------------------------------
# unpack_archive — every guard raises AND leaves no partial target
# ---------------------------------------------------------------------------


def _assert_no_partial(target: Path) -> None:
    assert not target.exists() and not target.is_symlink()


def test_member_dest_rejects_empty_name(tmp_path: Path) -> None:
    target = tmp_path / "t"
    target.mkdir()
    with pytest.raises(ArchiveUploadError, match="empty name"):
        _member_dest("", target)


def test_member_dest_dot_name_is_the_target_itself(tmp_path: Path) -> None:
    """``.`` resolves to the target itself — inside, so allowed (the
    containment check's equality arm)."""
    target = tmp_path / "t"
    target.mkdir()
    assert _member_dest(".", target) == target


def test_member_dest_rejects_resolution_escape(tmp_path: Path) -> None:
    """Defense in depth (the resolve-based containment check): a symlink
    already inside the target that points OUT makes any member routed
    through it escape once resolved."""
    target = tmp_path / "t"
    target.mkdir()
    (target / "sneaky").symlink_to(tmp_path / "outside")
    with pytest.raises(ArchiveUploadError, match="escapes"):
        _member_dest("sneaky/evil.txt", target)


def test_link_target_rejects_empty_target(tmp_path: Path) -> None:
    with pytest.raises(ArchiveUploadError, match="empty target"):
        _link_target_resolved("", tmp_path / "d", tmp_path / "t")


def test_unpack_tar_extractfile_none_is_corrupt(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """A REG member whose data tarfile cannot hand back (``extractfile``
    → None) is a corrupt member — rejected, partial state cleaned."""
    archive = tmp_path / "corrupt.tar"
    _make_tar(archive, [("f.txt", b"x", "f")], gz=False)
    monkeypatch.setattr(tarfile.TarFile, "extractfile", lambda self, member: None)
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="corrupt archive member"):
        unpack_archive(archive, target, 10_000)
    _assert_no_partial(target)


def test_unpack_zip_slip_rejected_and_cleaned(tmp_path: Path) -> None:
    archive = tmp_path / "evil.zip"
    _make_zip(archive, files={"../evil.txt": b"evil", "ok.txt": b"ok"})
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="traversal"):
        unpack_archive(archive, target, 10_000)
    _assert_no_partial(target)


@pytest.mark.parametrize("name", ["/etc/x", "C:\\evil", "\\\\server\\share"])
def test_unpack_zip_absolute_member_rejected_and_cleaned(tmp_path: Path, name: str) -> None:
    archive = tmp_path / "abs.zip"
    _make_zip(archive, files={name: b"no"})
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="absolute"):
        unpack_archive(archive, target, 10_000)
    _assert_no_partial(target)


def test_unpack_zip_symlink_entry_rejected_and_cleaned(tmp_path: Path) -> None:
    archive = tmp_path / "slink.zip"
    _make_zip(
        archive,
        extra_attrs={"link": (stat.S_IFLNK | 0o777) << 16},
    )
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="symlink"):
        unpack_archive(archive, target, 10_000)
    _assert_no_partial(target)


def test_unpack_zip_non_regular_entry_rejected_and_cleaned(tmp_path: Path) -> None:
    archive = tmp_path / "dev.zip"
    _make_zip(archive, extra_attrs={"dev": (stat.S_IFCHR | 0o644) << 16})
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="non-regular"):
        unpack_archive(archive, target, 10_000)
    _assert_no_partial(target)


def test_unpack_tar_absolute_member_rejected_and_cleaned(tmp_path: Path) -> None:
    archive = tmp_path / "abs.tar"
    _make_tar(archive, [("/etc/x", b"no", "f")], gz=False)
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="absolute"):
        unpack_archive(archive, target, 10_000)
    _assert_no_partial(target)


def test_unpack_tar_symlink_escape_rejected_and_cleaned(tmp_path: Path) -> None:
    archive = tmp_path / "slink.tar"
    _make_tar(archive, [("link", b"/etc/passwd", "sym")], gz=False)
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="escapes"):
        unpack_archive(archive, target, 10_000)
    _assert_no_partial(target)


def test_unpack_tar_relative_symlink_escape_rejected_and_cleaned(tmp_path: Path) -> None:
    archive = tmp_path / "slink2.tar"
    _make_tar(
        archive,
        [("sub/", b"", "d"), ("sub/escape", b"../../outside", "sym")],
        gz=False,
    )
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="escapes"):
        unpack_archive(archive, target, 10_000)
    _assert_no_partial(target)


def test_unpack_tar_hardlink_escape_rejected_and_cleaned(tmp_path: Path) -> None:
    archive = tmp_path / "hlink.tar"
    _make_tar(archive, [("hard", b"/etc/passwd", "lnk")], gz=False)
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="escapes"):
        unpack_archive(archive, target, 10_000)
    _assert_no_partial(target)


@pytest.mark.parametrize("kind", ["chr", "fifo"])
def test_unpack_tar_device_and_fifo_rejected_and_cleaned(tmp_path: Path, kind: str) -> None:
    archive = tmp_path / "dev.tar"
    _make_tar(archive, [("dev", b"", kind)], gz=False)
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="device or FIFO"):
        unpack_archive(archive, target, 10_000)
    _assert_no_partial(target)


@pytest.mark.parametrize("kind", ["zip", "tar"])
def test_unpack_extracted_cap_exceeded_rejected_and_cleaned(tmp_path: Path, kind: str) -> None:
    """cap=10 with a 20-byte file → the cap (not the content) is named
    and the partial tree is gone."""
    archive = tmp_path / ("cap." + kind)
    payload = b"b" * 20
    if kind == "zip":
        _make_zip(archive, files={"big.txt": payload})
    else:
        _make_tar(archive, [("big.txt", payload, "f")], gz=False)
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="10-byte extraction cap"):
        unpack_archive(archive, target, 10)
    _assert_no_partial(target)


def test_unpack_cap_accumulates_across_members(tmp_path: Path) -> None:
    """6 + 6 bytes under a 10-byte cap: each file alone is under, the
    total is not."""
    archive = tmp_path / "acc.zip"
    _make_zip(archive, files={"a.txt": b"a" * 6, "b.txt": b"b" * 6})
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="extraction cap"):
        unpack_archive(archive, target, 10)
    _assert_no_partial(target)


def test_unpack_corrupt_archive_rejected_and_cleaned(tmp_path: Path) -> None:
    archive = tmp_path / "junk.bin"
    archive.write_bytes(b"this is not an archive at all")
    target = tmp_path / "out"
    with pytest.raises(ArchiveUploadError, match="could not unpack"):
        unpack_archive(archive, target, 10_000)
    _assert_no_partial(target)


def test_unpack_existing_target_rejected(tmp_path: Path) -> None:
    archive = tmp_path / "ok.zip"
    _make_zip(archive, files={"a.txt": b"1"})
    target = tmp_path / "out"
    target.mkdir()
    with pytest.raises(ArchiveUploadError, match="already exists"):
        unpack_archive(archive, target, 10_000)
    # The pre-existing directory is left exactly as found.
    assert target.is_dir() and not any(target.iterdir())


# ---------------------------------------------------------------------------
# swap_in — atomic in-place replacement
# ---------------------------------------------------------------------------


def test_swap_in_fresh(tmp_path: Path) -> None:
    new = tmp_path / "new"
    new.mkdir()
    (new / "new.txt").write_text("new")
    final = tmp_path / "final"
    swap_in(new, final)
    assert final.is_dir()
    assert (final / "new.txt").read_text() == "new"
    assert not new.exists()


def test_swap_in_replaces_existing_with_full_content(tmp_path: Path) -> None:
    """The previous content is fully gone, the new content complete —
    no interleave — and no ``.old-`` sibling survives."""
    final = tmp_path / "final"
    final.mkdir()
    (final / "old.txt").write_text("old")
    (final / "keepdir/").mkdir()
    (final / "keepdir" / "stale.txt").write_text("stale")
    new = tmp_path / "new"
    new.mkdir()
    (new / "fresh.txt").write_text("fresh")
    (new / "keepdir/").mkdir()
    (new / "keepdir" / "v2.txt").write_text("v2")
    swap_in(new, final)
    assert not (final / "old.txt").exists()
    assert not (final / "keepdir" / "stale.txt").exists()
    assert (final / "fresh.txt").read_text() == "fresh"
    assert (final / "keepdir" / "v2.txt").read_text() == "v2"
    assert not new.exists()
    assert not list(tmp_path.glob("final.old-*"))


def test_swap_in_restores_previous_folder_on_failure(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The second rename (new → final) fails: the previous folder comes
    back intact, the new dir is cleaned, and the error is raised."""
    final = tmp_path / "final"
    final.mkdir()
    (final / "old.txt").write_text("old")
    new = tmp_path / "new"
    new.mkdir()
    (new / "fresh.txt").write_text("fresh")

    real_rename = os.rename

    def fake_rename(
        src: str | os.PathLike[str], dst: str | os.PathLike[str], *args: Any, **kwargs: Any
    ) -> Any:
        if Path(str(src)) == new:  # the new → final rename fails
            raise OSError("simulated swap failure")
        return real_rename(src, dst, *args, **kwargs)

    monkeypatch.setattr(os, "rename", fake_rename)
    with pytest.raises(ArchiveUploadError, match="could not replace"):
        swap_in(new, final)
    # Previous folder intact, byte for byte.
    assert (final / "old.txt").read_text() == "old"
    # New dir cleaned, no orphaned .old sibling.
    assert not new.exists()
    assert not list(tmp_path.glob("final.old-*"))


def test_swap_in_double_failure_leaves_no_orphan(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Both renames fail (new→final AND the ``.old-``→final restore):
    the best-effort path deletes the orphaned ``.old-`` sibling — the
    previous folder is unrecoverable in this scenario, so at least
    nothing is left mixed on disk."""
    final = tmp_path / "final"
    final.mkdir()
    (final / "old.txt").write_text("old")
    new = tmp_path / "new"
    new.mkdir()
    (new / "fresh.txt").write_text("fresh")

    real_rename = os.rename

    def fake_rename(
        src: str | os.PathLike[str], dst: str | os.PathLike[str], *args: Any, **kwargs: Any
    ) -> Any:
        s, d = Path(str(src)), Path(str(dst))
        if s == new or d == final:  # the swap rename and the restore both fail
            raise OSError("simulated failure")
        return real_rename(src, dst, *args, **kwargs)

    monkeypatch.setattr(os, "rename", fake_rename)
    with pytest.raises(ArchiveUploadError, match="could not replace"):
        swap_in(new, final)
    assert not new.exists()
    assert not list(tmp_path.glob("final.old-*"))