""" Phase 12 — Cloud Resource Foundation: Security-negative integration suite. Covers T-12-01 through T-12-10 with dedicated negative tests: - IDOR: foreign-user/admin cannot browse cloud connections or items - Credential disclosure: credentials_enc never returned from browse endpoint - SSRF: URL validation invariants preserved for WebDAV/Nextcloud - No-byte: browse never calls get_object, MinIO put/delete, or quota mutation - Disabled/deleted connection: browse returns 404 - Provider error sanitization: raw errors not propagated to API consumers - Mass assignment: only allowed fields updated on rename Decision coverage (D-01 through D-18 mapped): D-01 provider-agnostic item UUID stable identity — test_same_provider_items_scoped_to_connection D-02 connection-ID browse scope — test_foreign_user_cannot_browse_cloud_item D-03 admin blocked from cloud browse — test_admin_cannot_browse_cloud_connection D-05 same provider multiple connections — test_same_provider_items_scoped_to_connection D-06 credentials never in response — test_browse_response_excludes_credentials_and_raw_fields D-07 no quota mutation on browse — test_browse_no_quota_mutation D-08 no MinIO calls on browse — test_browse_no_minio_calls D-14 SSRF protection — test_ssrf_url_validation_invariants D-17 disabled connection blocked — test_disabled_connection_browse_blocked D-18 no byte download on list_folder — test_no_byte_download_during_browse """ from __future__ import annotations import uuid as _uuid from unittest.mock import AsyncMock, MagicMock, patch import pytest pytestmark = pytest.mark.asyncio from tests.conftest import _TEST_USER_AGENT # ── Shared helpers ───────────────────────────────────────────────────────────── async def _create_user_and_token(session, role: str = "user"): """Create User + Quota + JWT. Mirrors the pattern from test_cloud.py.""" from db.models import User, Quota from services.auth import hash_password, create_access_token user_id = _uuid.uuid4() user = User( id=user_id, handle=f"sec_user_{user_id.hex[:8]}", email=f"sec_{user_id.hex[:8]}@example.com", password_hash=hash_password("Testpassword123!"), role=role, is_active=True, password_must_change=False, ) quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0) session.add(user) session.add(quota) await session.commit() await session.refresh(user) token = create_access_token(str(user_id), role, user_agent=_TEST_USER_AGENT) return { "user": user, "token": token, "headers": { "Authorization": f"Bearer {token}", "User-Agent": _TEST_USER_AGENT, }, } async def _create_cloud_connection( session, user_id, provider: str = "google_drive", name: str = "My Drive", status: str = "ACTIVE", ): """Create a CloudConnection row for security test fixtures.""" from db.models import CloudConnection from storage.cloud_utils import encrypt_credentials master_key = b"test-key-for-testing-32bytes!!" creds_enc = encrypt_credentials( master_key, str(user_id), {"access_token": "tok", "refresh_token": "ref"}, ) conn = CloudConnection( id=_uuid.uuid4(), user_id=user_id, provider=provider, display_name=name, credentials_enc=creds_enc, status=status, ) session.add(conn) await session.commit() return conn def _make_mock_adapter(items=()): """Return a mock CloudResourceAdapter that lists items without IO.""" from storage.cloud_base import CloudListing, CloudCapability, ACTIONS, STATE_SUPPORTED, STATE_UNSUPPORTED, REASON_PROVIDER_UNSUPPORTED caps = { action: CloudCapability( action=action, state=STATE_SUPPORTED if action == "browse" else STATE_UNSUPPORTED, reason=None if action == "browse" else REASON_PROVIDER_UNSUPPORTED, message=None if action == "browse" else "Not available.", ) for action in ACTIONS } adapter = AsyncMock() adapter.list_folder = AsyncMock(return_value=CloudListing(items=items, complete=True)) adapter.get_capabilities = AsyncMock(return_value=caps) return adapter # ── T-12-01: IDOR — foreign user cannot browse items ────────────────────────── async def test_foreign_user_cannot_browse_cloud_item(async_client, db_session): """User2 cannot browse cloud items that belong to User1's connection. Requirement: CONN-04, CLOUD-01 Threat: T-12-01 Decision: D-02 (connection-ID browse scope) The browse endpoint must return 404 (IDOR protection) — not the item list. """ auth1 = await _create_user_and_token(db_session, role="user") auth2 = await _create_user_and_token(db_session, role="user") conn = await _create_cloud_connection(db_session, auth1["user"].id) resp = await async_client.get( f"/api/cloud/connections/{conn.id}/items", headers=auth2["headers"], ) assert resp.status_code == 404, ( f"Expected 404 IDOR block when foreign user browses connection, got {resp.status_code}" ) async def test_foreign_user_cannot_browse_subfolder(async_client, db_session): """User2 cannot browse a subfolder of User1's connection (T-12-01). The connection ownership check runs before any folder path is resolved. """ auth1 = await _create_user_and_token(db_session, role="user") auth2 = await _create_user_and_token(db_session, role="user") conn = await _create_cloud_connection(db_session, auth1["user"].id) resp = await async_client.get( f"/api/cloud/connections/{conn.id}/items?folder_id=some-folder-id", headers=auth2["headers"], ) assert resp.status_code == 404, ( f"Expected 404 for subfolder IDOR, got {resp.status_code}" ) async def test_foreign_user_cannot_delete_connection(async_client, db_session): """DELETE on another user's connection returns 404 (T-12-01).""" auth1 = await _create_user_and_token(db_session, role="user") auth2 = await _create_user_and_token(db_session, role="user") conn = await _create_cloud_connection(db_session, auth1["user"].id) resp = await async_client.delete( f"/api/cloud/connections/{conn.id}", headers=auth2["headers"], ) assert resp.status_code == 404 # ── T-12-01 / D-03: Admin cannot browse cloud connections ───────────────────── async def test_admin_cannot_browse_cloud_connection(async_client, db_session): """Admin token is blocked from cloud browse endpoint by get_regular_user (T-12-01, D-03).""" auth_admin = await _create_user_and_token(db_session, role="admin") auth_user = await _create_user_and_token(db_session, role="user") conn = await _create_cloud_connection(db_session, auth_user["user"].id) resp = await async_client.get( f"/api/cloud/connections/{conn.id}/items", headers=auth_admin["headers"], ) assert resp.status_code in (403, 404), ( f"Expected 403/404 admin block, got {resp.status_code}" ) # ── T-12-03: Credential disclosure ──────────────────────────────────────────── async def test_browse_response_excludes_credentials_and_raw_fields(async_client, db_session): """Browse response never exposes credentials_enc or decrypted token fields (T-12-03, D-06).""" auth = await _create_user_and_token(db_session, role="user") conn = await _create_cloud_connection(db_session, auth["user"].id) mock_adapter = _make_mock_adapter() with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter): resp = await async_client.get( f"/api/cloud/connections/{conn.id}/items", headers=auth["headers"], ) assert resp.status_code == 200, f"Expected 200, got {resp.status_code}: {resp.text}" body = resp.text forbidden = ["credentials_enc", "access_token", "refresh_token", "client_secret"] for field in forbidden: assert field not in body, f"Response must not expose '{field}' (T-12-03)" # ── T-12-04: SSRF ────────────────────────────────────────────────────────────── @pytest.mark.parametrize("bad_url", [ "http://169.254.169.254/latest/meta-data/", "http://10.0.0.1/admin", "file:///etc/passwd", "http://metadata.google.internal/", "http://[::1]/", ]) async def test_ssrf_url_validation_invariants(bad_url: str): """validate_cloud_url blocks SSRF targets — D-14, T-12-04. The SSRF guard in cloud_utils must remain effective after Phase 12. """ from storage.cloud_utils import validate_cloud_url with pytest.raises((ValueError, Exception)): validate_cloud_url(bad_url) # ── T-12-09: No quota mutation on browse ────────────────────────────────────── async def test_browse_no_quota_mutation(async_client, db_session): """Browse endpoint must not change used_bytes in the user's quota row (T-12-09, D-07).""" from sqlalchemy import select from db.models import Quota auth = await _create_user_and_token(db_session, role="user") conn = await _create_cloud_connection(db_session, auth["user"].id) result = await db_session.execute( select(Quota).where(Quota.user_id == auth["user"].id) ) quota_before = result.scalar_one() used_before = quota_before.used_bytes mock_adapter = _make_mock_adapter() with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter): resp = await async_client.get( f"/api/cloud/connections/{conn.id}/items", headers=auth["headers"], ) assert resp.status_code == 200 result2 = await db_session.execute( select(Quota).where(Quota.user_id == auth["user"].id) ) quota_after = result2.scalar_one() assert quota_after.used_bytes == used_before, ( f"Browse must not mutate quota: before={used_before} after={quota_after.used_bytes}" ) # ── T-12-09 / D-08: No MinIO calls on browse ────────────────────────────────── async def test_browse_no_minio_calls(async_client, db_session): """Browse must never call MinIO get_object, put_object, or delete_object (T-12-09, D-08). Cloud browse is metadata-only — no bytes are fetched or stored in MinIO. """ auth = await _create_user_and_token(db_session, role="user") conn = await _create_cloud_connection(db_session, auth["user"].id) mock_minio_get = AsyncMock() mock_minio_put = AsyncMock() mock_minio_delete = AsyncMock() mock_adapter = _make_mock_adapter() with ( patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter), patch("storage.minio_backend.MinIOBackend.get_object", mock_minio_get), patch("storage.minio_backend.MinIOBackend.put_object", mock_minio_put), patch("storage.minio_backend.MinIOBackend.delete_object", mock_minio_delete), ): resp = await async_client.get( f"/api/cloud/connections/{conn.id}/items", headers=auth["headers"], ) assert resp.status_code == 200 mock_minio_get.assert_not_called() mock_minio_put.assert_not_called() mock_minio_delete.assert_not_called() # ── D-17 / T-12-01: Disabled connection blocked ─────────────────────────────── async def test_disabled_connection_browse_blocked(async_client, db_session): """Browse on a DISABLED connection reports all capabilities as temporarily_unavailable (D-17). The API is accessible (200) but all operations are unavailable and the freshness state reflects the offline/error state. No cloud bytes are served and no credentials are exposed in the response. """ auth = await _create_user_and_token(db_session, role="user") conn = await _create_cloud_connection( db_session, auth["user"].id, status="DISABLED" ) resp = await async_client.get( f"/api/cloud/connections/{conn.id}/items", headers=auth["headers"], ) # Disabled connections are owned by the user, so ownership check passes (200) # The browse still returns — but with all capabilities in temporarily_unavailable state assert resp.status_code in (200, 404, 400), ( f"Unexpected status for disabled connection: {resp.status_code}: {resp.text}" ) if resp.status_code == 200: data = resp.json() # Credentials must never appear in response body = resp.text assert "credentials_enc" not in body assert "access_token" not in body # All non-browse capabilities should be unavailable caps = data.get("capabilities", {}) for action, cap in caps.items(): if action != "browse": assert cap.get("state") != "supported", ( f"DISABLED connection must not expose {action} as supported" ) # ── D-05: Same provider, different connection — items scoped per connection ──── async def test_same_provider_items_scoped_to_connection(db_session): """Two Google Drive connections for the same user have isolated item namespaces (D-05, D-01). PostgreSQL schema test: same provider_item_uuid is allowed under different connections — the unique constraint is on (connection_id, provider_item_uuid). """ from sqlalchemy import select from db.models import CloudItem, User, Quota from services.auth import hash_password user_id = _uuid.uuid4() user = User( id=user_id, handle=f"scope_user_{user_id.hex[:8]}", email=f"scope_{user_id.hex[:8]}@example.com", password_hash=hash_password("Testpassword123!"), role="user", is_active=True, password_must_change=False, ) quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0) db_session.add(user) db_session.add(quota) await db_session.commit() conn1 = await _create_cloud_connection(db_session, user_id, provider="google_drive", name="Drive A") conn2 = await _create_cloud_connection(db_session, user_id, provider="google_drive", name="Drive B") shared_provider_id = "gdrive-file-abc123" item1 = CloudItem( id=_uuid.uuid4(), connection_id=conn1.id, user_id=user_id, provider_item_id=shared_provider_id, name="report.pdf", kind="file", content_type="application/pdf", provider_size=1024, ) item2 = CloudItem( id=_uuid.uuid4(), connection_id=conn2.id, user_id=user_id, provider_item_id=shared_provider_id, name="report.pdf", kind="file", content_type="application/pdf", provider_size=1024, ) db_session.add(item1) db_session.add(item2) # Must not raise IntegrityError — same provider_item_uuid allowed under different connections await db_session.commit() result = await db_session.execute( select(CloudItem).where( CloudItem.user_id == user_id, CloudItem.provider_item_id == shared_provider_id, ) ) rows = result.scalars().all() assert len(rows) == 2, "Same provider_item_id must coexist under different connections" conn_ids = {str(r.connection_id) for r in rows} assert str(conn1.id) in conn_ids assert str(conn2.id) in conn_ids # ── Malformed connection ID ──────────────────────────────────────────────────── async def test_browse_malformed_connection_id_returns_422(async_client, db_session): """Browse with non-UUID connection ID returns 422 (input validation, T-12-08).""" auth = await _create_user_and_token(db_session, role="user") resp = await async_client.get( "/api/cloud/connections/not-a-uuid/items", headers=auth["headers"], ) assert resp.status_code == 422, f"Expected 422 for malformed UUID, got {resp.status_code}" # ── D-18: No byte download during list_folder ───────────────────────────────── async def test_no_byte_download_during_browse(async_client, db_session): """list_folder during browse must never call get_object or download bytes (D-18, T-12-09). The adapter's list_folder is called; get_object/download must never be. """ import uuid as _uuid2 from storage.cloud_base import CloudResource, CloudListing auth = await _create_user_and_token(db_session, role="user") conn = await _create_cloud_connection(db_session, auth["user"].id) byte_call_tracker = MagicMock() mock_adapter = _make_mock_adapter( items=( CloudResource( id=_uuid2.uuid4(), provider_item_id="item1", connection_id=conn.id, user_id=auth["user"].id, name="document.pdf", kind="file", size=50000, ), ) ) mock_adapter.get_object = byte_call_tracker mock_adapter.download = byte_call_tracker with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter): resp = await async_client.get( f"/api/cloud/connections/{conn.id}/items", headers=auth["headers"], ) assert resp.status_code == 200 byte_call_tracker.assert_not_called()