lfnovo--open-notebook/tests/test_source_path_containment.py

"""
Tests for the sources.py path-containment fix in _resolve_source_file() and
_is_source_file_available() (api/routers/sources.py).

Both compared `resolved_path.startswith(safe_root)` without a trailing
separator - a sibling directory that merely starts with the same string
(e.g. "uploads_evil/") would incorrectly be treated as contained, unlike
the file's other two path checks (generate_unique_filename(), the inline
check in create_source()) which already guard with `+ os.sep`. Not
reachable today (source.asset.file_path is only ever set server-side), but
these lock in the fix and guard against regressions.
"""

from unittest.mock import AsyncMock, patch

import pytest
from fastapi.testclient import TestClient

from api.routers.sources import _is_source_file_available, _resolve_source_file
from open_notebook.config import UPLOADS_FOLDER
from open_notebook.domain.notebook import Asset, Source


def make_source(file_path=None, **overrides):
    defaults = dict(
        id="source:test123",
        title="Test Source",
        asset=Asset(file_path=file_path) if file_path else None,
    )
    defaults.update(overrides)
    return Source(**defaults)


@pytest.fixture
def client():
    from api.main import app

    return TestClient(app)


class TestIsSourceFileAvailableRejectsSiblingDirectoryBypass:
    def test_sibling_directory_with_matching_prefix_is_not_available(
        self, tmp_path, monkeypatch
    ):
        real_root = tmp_path / "uploads"
        real_root.mkdir()
        monkeypatch.setattr("api.routers.sources.UPLOADS_FOLDER", str(real_root))

        sibling = tmp_path / "uploads_evil"
        sibling.mkdir()
        evil_file = sibling / "secret.txt"
        evil_file.write_bytes(b"not yours")

        source = make_source(file_path=str(evil_file))
        assert _is_source_file_available(source) is False

    def test_file_genuinely_inside_uploads_folder_is_available(
        self, tmp_path, monkeypatch
    ):
        real_root = tmp_path / "uploads"
        real_root.mkdir()
        monkeypatch.setattr("api.routers.sources.UPLOADS_FOLDER", str(real_root))

        legit_file = real_root / "document.pdf"
        legit_file.write_bytes(b"pdf bytes")

        source = make_source(file_path=str(legit_file))
        assert _is_source_file_available(source) is True

    def test_missing_file_inside_uploads_folder_is_unavailable_not_error(
        self, tmp_path, monkeypatch
    ):
        real_root = tmp_path / "uploads"
        real_root.mkdir()
        monkeypatch.setattr("api.routers.sources.UPLOADS_FOLDER", str(real_root))

        missing_file = real_root / "does-not-exist.pdf"
        source = make_source(file_path=str(missing_file))
        assert _is_source_file_available(source) is False

    def test_no_asset_returns_none(self):
        source = make_source(file_path=None)
        assert _is_source_file_available(source) is None

    def test_path_traversal_outside_root_is_not_available(self, tmp_path, monkeypatch):
        root = tmp_path / "uploads"
        root.mkdir()
        monkeypatch.setattr("api.routers.sources.UPLOADS_FOLDER", str(root))

        outside = tmp_path / "outside.pdf"
        outside.write_bytes(b"not yours")
        traversal_path = str(root / ".." / "outside.pdf")

        source = make_source(file_path=traversal_path)
        assert _is_source_file_available(source) is False


class TestResolveSourceFileRejectsSiblingDirectoryBypass:
    @pytest.mark.asyncio
    async def test_sibling_directory_raises_403(self, tmp_path, monkeypatch):
        real_root = tmp_path / "uploads"
        real_root.mkdir()
        monkeypatch.setattr("api.routers.sources.UPLOADS_FOLDER", str(real_root))

        sibling = tmp_path / "uploads_evil"
        sibling.mkdir()
        evil_file = sibling / "secret.txt"
        evil_file.write_bytes(b"not yours")

        source = make_source(file_path=str(evil_file))
        with patch(
            "api.routers.sources.Source.get", new=AsyncMock(return_value=source)
        ):
            from fastapi import HTTPException

            with pytest.raises(HTTPException) as exc_info:
                await _resolve_source_file("source:test123")
            assert exc_info.value.status_code == 403

    @pytest.mark.asyncio
    async def test_file_genuinely_inside_uploads_folder_resolves(
        self, tmp_path, monkeypatch
    ):
        real_root = tmp_path / "uploads"
        real_root.mkdir()
        monkeypatch.setattr("api.routers.sources.UPLOADS_FOLDER", str(real_root))

        legit_file = real_root / "document.pdf"
        legit_file.write_bytes(b"pdf bytes")

        source = make_source(file_path=str(legit_file))
        with patch(
            "api.routers.sources.Source.get", new=AsyncMock(return_value=source)
        ):
            resolved_path, filename = await _resolve_source_file("source:test123")

        assert filename == "document.pdf"
        assert resolved_path == str(legit_file.resolve())


class TestDownloadEndpointRejectsSiblingDirectoryBypass:
    """End-to-end through the actual HTTP endpoint."""

    def test_download_returns_403_for_sibling_directory_file(
        self, client, tmp_path, monkeypatch
    ):
        real_root = tmp_path / "uploads"
        real_root.mkdir()
        monkeypatch.setattr("api.routers.sources.UPLOADS_FOLDER", str(real_root))

        sibling = tmp_path / "uploads_evil"
        sibling.mkdir()
        evil_file = sibling / "secret.txt"
        evil_file.write_bytes(b"not yours")

        source = make_source(file_path=str(evil_file))
        with patch(
            "api.routers.sources.Source.get", new=AsyncMock(return_value=source)
        ):
            response = client.get("/api/sources/source:test123/download")

        assert response.status_code == 403


class TestRealUploadsFolderStillWorks:
    """Sanity check against the real (non-monkeypatched) UPLOADS_FOLDER."""

    def test_real_uploads_folder_file_is_available(self):
        from pathlib import Path

        test_file = Path(UPLOADS_FOLDER) / "containment_test_file.txt"
        test_file.write_bytes(b"test")
        try:
            source = make_source(file_path=str(test_file))
            assert _is_source_file_available(source) is True
        finally:
            test_file.unlink(missing_ok=True)