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/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."""