From 805fe44bfba97702a81a475dc836a45d04592018 Mon Sep 17 00:00:00 2001 From: curo1305 Date: Mon, 22 Jun 2026 08:12:39 +0200 Subject: [PATCH] feat(12.1-01): add cross-origin nextLink guard and Drive/OneDrive contract tests - Validate @odata.nextLink hostname against graph.microsoft.com before following (T-12.1-03); cross-origin nextLink returns complete=False with prior items retained - Add TestOneDriveCrossOriginNextLink: cross-origin rejection, same-origin follows, page failure retains prior items - Add TestGoogleDriveAuthFailureControl: 401 returns complete=False, page1+page2 error retains page1 items - All 163 four-provider contract tests pass --- backend/storage/onedrive_backend.py | 16 ++- backend/tests/test_cloud_backends.py | 158 +++++++++++++++++++++++++++ 2 files changed, 172 insertions(+), 2 deletions(-) diff --git a/backend/storage/onedrive_backend.py b/backend/storage/onedrive_backend.py index ad28dc6..ce42f33 100644 --- a/backend/storage/onedrive_backend.py +++ b/backend/storage/onedrive_backend.py @@ -28,6 +28,8 @@ import io import uuid from typing import Optional +import urllib.parse + import httpx import msal @@ -50,6 +52,9 @@ from storage.google_drive_backend import CloudConnectionError # reuse shared ex GRAPH_BASE = "https://graph.microsoft.com/v1.0" CHUNK_SIZE = 10 * 1024 * 1024 # 10 MB — above Graph's 4 MB simple upload limit (Pitfall 6) +# Trusted Graph API origin for @odata.nextLink validation (T-12.1-03) +_GRAPH_HOST = "graph.microsoft.com" + class OneDriveBackend(StorageBackend, CloudResourceAdapter): """Microsoft Graph / OneDrive implementation of StorageBackend. @@ -351,7 +356,6 @@ class OneDriveBackend(StorageBackend, CloudResourceAdapter): content_type: Optional[str] = None if "file" in item: content_type = item["file"].get("mimeType") - parent_ref_val = item.get("parentReference", {}).get("id") resources.append( CloudResource( id=uuid.uuid4(), @@ -367,7 +371,15 @@ class OneDriveBackend(StorageBackend, CloudResourceAdapter): etag=item.get("eTag") or item.get("cTag"), ) ) - url = data.get("@odata.nextLink") + # Validate @odata.nextLink origin before following (T-12.1-03) + next_link = data.get("@odata.nextLink") + if next_link: + parsed_next = urllib.parse.urlparse(next_link) + if parsed_next.hostname != _GRAPH_HOST: + # Cross-origin continuation URL — untrusted; stop pagination + complete = False + break + url = next_link except Exception: complete = False diff --git a/backend/tests/test_cloud_backends.py b/backend/tests/test_cloud_backends.py index 438f981..129f2de 100644 --- a/backend/tests/test_cloud_backends.py +++ b/backend/tests/test_cloud_backends.py @@ -535,6 +535,164 @@ class TestOneDriveListFolder: assert len(result.items) == 2 +class TestOneDriveCrossOriginNextLink: + """OneDrive @odata.nextLink must stay on graph.microsoft.com — T-12.1-03.""" + + def _make_backend(self): + from storage.onedrive_backend import OneDriveBackend + return OneDriveBackend({ + "access_token": "tok", + "refresh_token": "ref", + "expires_at": "2099-01-01T00:00:00", + }) + + @pytest.mark.asyncio + async def test_cross_origin_next_link_returns_incomplete(self): + """A cross-origin @odata.nextLink must stop pagination and return complete=False.""" + from unittest.mock import patch, MagicMock, AsyncMock + import uuid + + backend = self._make_backend() + page1 = MagicMock() + page1.is_success = True + page1.json.return_value = { + "value": [{"id": "i1", "name": "A.txt", "file": {}, "size": 10}], + # Cross-origin nextLink — must be rejected + "@odata.nextLink": "https://evil.example.com/v1.0/me/drive/root/children?$skip=1", + } + + with patch("httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_client.get = AsyncMock(return_value=page1) + mock_client_cls.return_value = mock_client + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + + # Items from page1 should be retained + assert len(result.items) == 1 + assert result.items[0].name == "A.txt" + # But listing is not complete (cross-origin continuation not trusted) + assert result.complete is False + + @pytest.mark.asyncio + async def test_same_origin_graph_next_link_followed(self): + """A same-origin graph.microsoft.com @odata.nextLink is trusted and followed.""" + from unittest.mock import patch, MagicMock, AsyncMock + import uuid + + backend = self._make_backend() + page1 = MagicMock() + page1.is_success = True + page1.json.return_value = { + "value": [{"id": "i1", "name": "A.txt", "file": {}, "size": 10}], + "@odata.nextLink": "https://graph.microsoft.com/v1.0/me/drive/root/children?$skip=1", + } + page2 = MagicMock() + page2.is_success = True + page2.json.return_value = { + "value": [{"id": "i2", "name": "B.txt", "file": {}, "size": 20}], + } + + with patch("httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_client.get = AsyncMock(side_effect=[page1, page2]) + mock_client_cls.return_value = mock_client + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + + assert result.complete is True + assert len(result.items) == 2 + + @pytest.mark.asyncio + async def test_page_failure_retains_prior_items(self): + """A page HTTP failure returns complete=False but retains already-normalized items.""" + from unittest.mock import patch, MagicMock, AsyncMock + import uuid + + backend = self._make_backend() + page1 = MagicMock() + page1.is_success = True + page1.json.return_value = { + "value": [{"id": "i1", "name": "A.txt", "file": {}, "size": 10}], + "@odata.nextLink": "https://graph.microsoft.com/v1.0/me/drive/root/children?$skip=1", + } + page2_fail = MagicMock() + page2_fail.is_success = False + page2_fail.status_code = 503 + + with patch("httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_client.get = AsyncMock(side_effect=[page1, page2_fail]) + mock_client_cls.return_value = mock_client + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + + # Page1 items retained + assert len(result.items) == 1 + assert result.items[0].name == "A.txt" + # Not complete because page2 failed + assert result.complete is False + + +class TestGoogleDriveAuthFailureControl: + """Drive auth failures map to controlled complete=False — T-12.1-03.""" + + def _make_backend(self): + from storage.google_drive_backend import GoogleDriveBackend + return GoogleDriveBackend({ + "access_token": "tok", + "refresh_token": "ref", + "token_uri": "https://oauth2.googleapis.com/token", + "client_id": "cid", + "client_secret": "csec", + }) + + @pytest.mark.asyncio + async def test_401_returns_incomplete_not_raises(self): + """401 from Drive returns complete=False — not an uncaught exception.""" + from unittest.mock import MagicMock, patch + from googleapiclient.errors import HttpError + import uuid + + backend = self._make_backend() + fake_service = MagicMock() + resp = MagicMock() + resp.status = 401 + fake_service.files().list().execute.side_effect = HttpError(resp, b"unauthorized") + with patch.object(backend, "_get_service", return_value=fake_service): + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + assert result.complete is False + + @pytest.mark.asyncio + async def test_page1_ok_page2_error_retains_items(self): + """Error on page2 still returns items from page1 with complete=False.""" + from unittest.mock import MagicMock, patch + from googleapiclient.errors import HttpError + import uuid + + backend = self._make_backend() + fake_service = MagicMock() + resp403 = MagicMock() + resp403.status = 403 + fake_service.files().list().execute.side_effect = [ + { + "files": [{"id": "f1", "name": "a.pdf", "mimeType": "application/pdf", "size": "100"}], + "nextPageToken": "synth_tok", + }, + HttpError(resp403, b"scope"), + ] + with patch.object(backend, "_get_service", return_value=fake_service): + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + + # Items from page1 must be retained + assert len(result.items) == 1 + assert result.items[0].name == "a.pdf" + assert result.complete is False + + class TestWebDAVListFolder: """WebDAV list_folder tests — SSRF guard and no byte-download."""