office_translator/tests/test_security_fixes_c1_c4.py

"""Tests for security fixes C1–C4 (audit 2026-08-26)."""

from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from routes import translate_routes as tr
from middleware.cleanup import FileCleanupManager


class TestSanitizeUrlFilename:
    """AC1 — path traversal in URL-downloaded filenames is neutralized."""

    @pytest.mark.parametrize(
        "raw",
        [
            "../../evil.xlsx",
            "..\\..\\evil.docx",
            "normal_file.pptx",
            "..\t..evil.pdf",
            "...",
            "",
        ],
    )
    def test_traversal_stripped(self, raw):
        result = tr._sanitize_url_filename(raw)
        # Core guarantee: no traversal or path separators survive
        assert ".." not in result
        assert "/" not in result
        assert "\\" not in result
        assert result != ""

    def test_long_filename_truncated(self):
        raw = "a" * 300 + ".xlsx"
        result = tr._sanitize_url_filename(raw)
        assert len(result) <= 255
        assert result.endswith(".xlsx")


class TestRedirectSsrf:
    """AC2 — redirect to internal address is blocked."""

    @pytest.mark.asyncio
    async def test_redirect_to_metadata_blocked(self):
        def fake_response(status, location=None):
            resp = MagicMock()
            resp.status_code = status
            resp.headers = {"location": location} if location else {}
            return resp

        class FakeClient:
            def build_request(self, method, url):
                return (method, url)

            async def send(self, req, stream=False):
                url = req[1]
                if "public.example" in url:
                    return fake_response(302, "http://169.254.169.254/latest/meta-data")
                return fake_response(200)

            async def __aenter__(self):
                return self

            async def __aexit__(self, *a):
                return False

        with patch.object(tr.httpx, "AsyncClient", return_value=FakeClient()):
            with pytest.raises(tr.TranslateEndpointError) as exc:
                await tr.download_from_url("http://public.example/file.xlsx")
            assert exc.value.details.get("reason") == "ssrf_blocked"


class TestCleanupOrphanGrace:
    """AC3 — young orphans are not deleted."""

    def _manager(self, tmp_path):
        m = FileCleanupManager(
            upload_dir=tmp_path / "uploads",
            output_dir=tmp_path / "outputs",
            temp_dir=tmp_path / "temp",
        )
        for d in (m.upload_dir, m.output_dir, m.temp_dir):
            d.mkdir(parents=True, exist_ok=True)
        return m

    @pytest.mark.asyncio
    async def test_young_orphan_kept(self, tmp_path):
        """Orphan younger than the grace period is NOT deleted."""
        m = self._manager(tmp_path)
        f = m.upload_dir / "recent_orphan.xlsx"
        f.write_bytes(b"x")

        fake_redis = MagicMock()
        fake_redis.keys = AsyncMock(return_value=[])
        fake_redis.get = AsyncMock(return_value=None)
        with patch(
            "middleware.cleanup._get_async_redis", return_value=fake_redis
        ):
            stats = await m.cleanup()
        assert f.exists()

    @pytest.mark.asyncio
    async def test_input_path_key_recognized(self, tmp_path):
        """Files tracked under 'input_path' are not orphans (key mismatch fix)."""
        import json as _json

        m = self._manager(tmp_path)
        f = m.upload_dir / "tracked.xlsx"
        f.write_bytes(b"x")

        fake_redis = MagicMock()
        fake_redis.keys = AsyncMock(return_value=["translation:file:tr_1"])
        fake_redis.get = AsyncMock(
            return_value=_json.dumps({"input_path": str(f), "user_id": "u1"})
        )
        with patch(
            "middleware.cleanup._get_async_redis", return_value=fake_redis
        ):
            stats = await m.cleanup()
        assert f.exists()
        assert stats["orphaned_deleted"] == 0

    @pytest.mark.asyncio
    async def test_old_orphan_deleted(self, tmp_path):
        import json as _json
        import os
        import time

        m = self._manager(tmp_path)
        f = m.upload_dir / "old_orphan.xlsx"
        f.write_bytes(b"x")
        old = time.time() - (m.orphan_grace_seconds + 600)
        os.utime(f, (old, old))

        fake_redis = MagicMock()
        fake_redis.keys = AsyncMock(return_value=[])
        fake_redis.get = AsyncMock(return_value=None)
        with patch(
            "middleware.cleanup._get_async_redis", return_value=fake_redis
        ):
            stats = await m.cleanup()
        assert not f.exists()
        assert stats["orphaned_deleted"] == 1


class TestJobAccessControl:
    """H2 — ownership / token checks on status and download."""

    def _job(self, user_id=None, token="tok123"):
        return {"id": "tr_abc", "user_id": user_id, "access_token": token}

    def _user(self, uid):
        u = MagicMock()
        u.id = uid
        return u

    def test_owner_allowed(self):
        job = self._job(user_id=7)
        assert tr._check_job_access(job, self._user(7), None) is None

    def test_other_user_denied(self):
        job = self._job(user_id=7)
        resp = tr._check_job_access(job, self._user(8), None)
        assert resp is not None and resp.status_code == 403

    def test_anonymous_caller_on_owned_job_denied(self):
        job = self._job(user_id=7)
        resp = tr._check_job_access(job, None, "tok123")
        assert resp is not None and resp.status_code == 401

    def test_anonymous_job_requires_token(self):
        job = self._job(user_id=None)
        assert tr._check_job_access(job, None, "wrong") is not None
        assert tr._check_job_access(job, None, None) is not None
        assert tr._check_job_access(job, None, "tok123") is None

    def test_old_anonymous_job_without_token_denied(self):
        job = {"id": "tr_old", "user_id": None}  # job created before the fix
        assert tr._check_job_access(job, None, "anything") is not None


class TestZipBomb:
    """H1 — dangerous archives are rejected."""

    def _make_zip(self, tmp_path, entries):
        import zipfile

        p = tmp_path / "bomb.xlsx"
        with zipfile.ZipFile(p, "w", zipfile.ZIP_DEFLATED) as zf:
            for name, data in entries:
                zf.writestr(name, data)
        return p

    def test_normal_file_accepted(self, tmp_path):
        from utils.file_handler import validate_zip_safety

        p = self._make_zip(tmp_path, [("sheet1.xml", b"<xml/>ok" * 100)])
        validate_zip_safety(p)  # no exception

    def test_not_a_zip_rejected(self, tmp_path):
        from utils.file_handler import validate_zip_safety

        p = tmp_path / "fake.xlsx"
        p.write_bytes(b"this is not a zip file")
        with pytest.raises(ValueError):
            validate_zip_safety(p)

    def test_high_ratio_rejected(self, tmp_path):
        from utils.file_handler import validate_zip_safety

        # 50 MB of zeros compresses far beyond the 100:1 ratio cap
        p = self._make_zip(tmp_path, [("huge.xml", b"\0" * (50 * 1024 * 1024))])
        with pytest.raises(ValueError):
            validate_zip_safety(p)

    def test_declared_total_too_big_rejected(self, tmp_path):
        import zipfile
        from unittest.mock import patch as _patch
        from utils.file_handler import validate_zip_safety

        p = self._make_zip(tmp_path, [("a.xml", b"<xml/>")])
        fake_info = MagicMock()
        fake_info.is_dir = lambda: False
        fake_info.file_size = 5 * 1024 * 1024 * 1024  # 5 GB declared
        fake_info.compress_size = 50 * 1024 * 1024
        with _patch.object(zipfile.ZipFile, "infolist", return_value=[fake_info]):
            with pytest.raises(ValueError):
                validate_zip_safety(p)