From 779b05086b3e0b7d4a57a1fd25021088e9bccab8 Mon Sep 17 00:00:00 2001 From: curo1305 Date: Mon, 22 Jun 2026 09:02:42 +0200 Subject: [PATCH] test(12.1-04): add opt-in live Nextcloud smoke test suite MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - backend/tests/test_nextcloud_live.py: three read-only live tests behind live_nextcloud marker (adapter root metadata, sanitized diagnostic, connection-ID browse as testuser@docuvault.example with IDOR/admin negatives) - backend/pytest.ini: register live_nextcloud marker; addopts excludes it from ordinary runs so CI never contacts Nextcloud without explicit selection - 12.1-VALIDATION.md: live test contract, commands, skip behaviour, sanitized 7-of-10 probe result, and threat references T-12.1-16 through T-12.1-20 - Manifest status: unconfirmed — Task 2 exact-name gate pending owner decision --- .../12.1-VALIDATION.md | 140 +++++ backend/pytest.ini | 3 + backend/tests/test_nextcloud_live.py | 511 ++++++++++++++++++ 3 files changed, 654 insertions(+) create mode 100644 .planning/phases/12.1-fix-nextcloud-root-listing-and-sync-visibility/12.1-VALIDATION.md create mode 100644 backend/tests/test_nextcloud_live.py diff --git a/.planning/phases/12.1-fix-nextcloud-root-listing-and-sync-visibility/12.1-VALIDATION.md b/.planning/phases/12.1-fix-nextcloud-root-listing-and-sync-visibility/12.1-VALIDATION.md new file mode 100644 index 0000000..40952e2 --- /dev/null +++ b/.planning/phases/12.1-fix-nextcloud-root-listing-and-sync-visibility/12.1-VALIDATION.md @@ -0,0 +1,140 @@ +# Phase 12.1: Fix Nextcloud Root Listing and Sync Visibility — Validation Matrix + +**Status:** Plan 04 Task 1 complete. Awaiting Task 2 (owner fixture reconciliation). +**Updated:** 2026-06-22 + +--- + +## Nyquist Validation Matrix + +| Requirement | Test type | Coverage | Evidence | +|------------|-----------|----------|---------| +| CONN-04: Connection-ID IDOR | Security-negative | All plans | `test_foreign_user_cannot_browse_cloud_item`, `test_admin_cannot_browse_cloud_connection` | +| CLOUD-01: Provider-neutral browse | Contract suite | P01 | `test_cloud_provider_contract.py` — 48 parametrized tests across all 4 providers | +| CLOUD-01: Root listing correct | Live smoke | P04 | `test_nextcloud_root_diagnostic` — sanitized counts and kind breakdown | +| CLOUD-01: Connection-ID API browse | End-to-end live | P04 | `test_nextcloud_connection_id_browse_as_designated_user` | +| CACHE-01: Freshness truthful | TDD GREEN | P02 | `test_incomplete_listing_never_marks_folder_fresh`, `test_browse_complete_false_never_sets_fresh` | +| SYNC-01: Frontend normalized shape | TDD GREEN | P03 | `CloudFolderView.test.js` — kind, provider_item_id, verbatim freshness | + +--- + +## Live Test Contract (Plan 04, Task 1) + +### Marker + +``` +live_nextcloud +``` + +Tests tagged with this marker are excluded from ordinary pytest runs via `addopts = -m "not live_nextcloud"` in `backend/pytest.ini`. Select explicitly with: + +```bash +cd backend && pytest -m live_nextcloud tests/test_nextcloud_live.py +``` + +### Required environment variables (values in ignored `.env` — never committed) + +``` +NEXTCLOUD_URL # Nextcloud server base URL +NEXTCLOUD_USER # Nextcloud username +NEXTCLOUD_APP_PASSWORD # Nextcloud app password +``` + +If any variable is absent, all live tests skip with a message naming only the variable keys. + +### Tests included + +| Test | Purpose | +|------|---------| +| `test_nextcloud_adapter_read_only_root_metadata` | Verifies the production adapter lists root via canonical four-argument signature, returns CloudListing, items carry trusted caller identity, no byte/mutation method is accessible | +| `test_nextcloud_root_diagnostic` | Sanitized count/kind/match diagnostic — nonblocking while manifest is unconfirmed | +| `test_nextcloud_connection_id_browse_as_designated_user` | End-to-end browse via `GET /api/cloud/connections/{id}/items` as `testuser@docuvault.example` in isolated test DB; asserts foreign-user and admin denial | + +### Read-only / no-mutation contract + +The network guard integrated in the test design blocks these HTTP methods before dispatch: +`GET` (byte content), `PUT`, `POST`, `PATCH`, `DELETE`, `MKCOL`, `MOVE`, `COPY`. + +Only `PROPFIND`, `OPTIONS`, and `HEAD` are permitted. Mutation methods are asserted absent +on the adapter object. No MinIO, quota, or object-storage change is made. + +### Threat coverage + +| Threat ID | Mitigation | Test | +|-----------|-----------|------| +| T-12.1-16 | Credentials/full URL excluded from output | `_assert_no_secrets_in_string` on captured output; no values in parametrize IDs, assertions, or exceptions | +| T-12.1-17 | No byte download or mutation | `_NetworkMethodGuard`; mutation method absence assertion | +| T-12.1-18 | Designated regular user only; IDOR/admin negatives | `test_nextcloud_connection_id_browse_as_designated_user` | +| T-12.1-19 | Count-only acceptance blocked | Exact name gate deferred to Task 2 checkpoint | +| T-12.1-20 | Unexpected names never printed | `print(f"Unexpected item count: {unexpected_count} (names withheld)")` | + +--- + +## Sanitized Probe Result (from 12.1-RESEARCH.md, pre-Plan-04) + +| Check | Result | +|-------|--------| +| Endpoint reachable | Yes | +| Authentication accepted | Yes | +| HTTP status | 207 Multi-Status | +| Direct root children returned | 10 | +| Exact supplied-name matches | 7 of 10 | +| Supplied names not exactly matched | `Manual Nextcloud.png`, `Nextcloud Intro.mp4`, `Template credits.md` | +| Additional returned-name count | 3; names withheld | +| Byte download or mutation | None | + +--- + +## Manifest Status: UNCONFIRMED — Task 2 Pending + +The exact-name acceptance gate is blocked. The three supplier-expected names that were not +exactly matched in the prior probe are: + +- `Manual Nextcloud.png` +- `Nextcloud Intro.mp4` +- `Template credits.md` + +The project owner must confirm (via Task 2 checkpoint) which source is authoritative before +the fixture `backend/tests/fixtures/cloud/nextcloud_expected_root.json` is created and +`test_nextcloud_expected_root_manifest` is enabled. + +Unexpected provider names returned for the 3 unmatched slots are withheld here and must not +appear in any committed artifact, log, or response. + +--- + +## Plans 01–03 Evidence Summary + +### Plan 01 — Adapter Contract + +- `test_cloud_provider_contract.py` — 48 tests: all pass +- `test_cloud_backends.py` — 67 tests: all pass +- `test_webdav_backend.py` — 29 tests: all pass +- `test_cloud_security.py` — 19 tests: all pass +- Full backend suite: 585 pass (1 pre-existing `test_extract_docx` failure, unrelated) + +### Plan 02 — Freshness Gate + +- `test_cloud_items.py` — 45 tests: all pass +- `test_cloud.py` — 25 tests: all pass +- `test_cloud_security.py` — 22 tests: all pass +- Full backend suite: 595 pass (1 pre-existing failure, unrelated) +- No unconditional `refresh_state="fresh"` in `browse.py` or `cloud_tasks.py` + +### Plan 03 — Frontend Contract + +- `CloudFolderView.test.js` — 23 tests: all pass +- `CloudProviderTreeItem.test.js` — 8 tests: all pass +- `CloudFolderTreeItem.test.js` — 16 tests: all pass +- `StorageBrowser.skeleton.test.js` — 22 tests: all pass +- `cloudConnections.test.js` — 23 tests: all pass +- `CloudBreadcrumbNavigation.test.js` — 8 tests: all pass +- Full frontend suite: 369 pass +- Production build: clean + +### Plan 04 Task 1 — Live Test Scaffold + +- `test_nextcloud_live.py` created with `live_nextcloud` marker +- Marker excluded from default runs via `pytest.ini` +- Three live tests: adapter metadata, sanitized root diagnostic, connection-ID end-to-end +- Task 2 (exact-name acceptance) deferred to checkpoint — fixture not yet created diff --git a/backend/pytest.ini b/backend/pytest.ini index 78c5011..7e10bfa 100644 --- a/backend/pytest.ini +++ b/backend/pytest.ini @@ -1,3 +1,6 @@ [pytest] asyncio_mode = auto testpaths = tests +markers = + live_nextcloud: opt-in read-only live Nextcloud tests — requires NEXTCLOUD_URL, NEXTCLOUD_USER, NEXTCLOUD_APP_PASSWORD; excluded from ordinary CI runs +addopts = -m "not live_nextcloud" diff --git a/backend/tests/test_nextcloud_live.py b/backend/tests/test_nextcloud_live.py new file mode 100644 index 0000000..140f3e4 --- /dev/null +++ b/backend/tests/test_nextcloud_live.py @@ -0,0 +1,511 @@ +""" +Phase 12.1 Plan 04 — Opt-in read-only Nextcloud live test suite. + +Usage (credentials must be set in the ignored .env file — never committed): + + # Sanitized diagnostic only (nonblocking): + pytest -m live_nextcloud tests/test_nextcloud_live.py + + # Add exact-name acceptance (only after owner confirmation stored in fixture): + pytest -m live_nextcloud tests/test_nextcloud_live.py -k expected_root_manifest + +Security invariants: + T-12.1-16 — credentials/full URL never in output/logs/artifacts + T-12.1-17 — network guard rejects byte GET and all mutation methods + T-12.1-18 — live tests authenticate only as the designated regular user + T-12.1-19 — count-only acceptance is blocked; exact reconciliation required + T-12.1-20 — unexpected provider names never printed or persisted + +Credential skip behaviour: + If NEXTCLOUD_URL, NEXTCLOUD_USER, or NEXTCLOUD_APP_PASSWORD are absent from + the environment the test skips with a message naming the variable keys only. + No values, hostnames, or paths are ever printed. +""" +from __future__ import annotations + +import logging +import os +import re +import uuid as _uuid +from typing import Optional +from unittest.mock import patch, MagicMock + +import pytest +import pytest_asyncio + +# ── Expected candidate manifest (owner-supplied, unconfirmed until Task 2) ──── +# Exact acceptance is blocked while the 7-of-10 discrepancy is unresolved. +# Do NOT add unexpected names here — this is the owner-supplied expectation only. + +_CANDIDATE_NAMES: frozenset[str] = frozenset({ + "Documents", + "docuvault", + "Photos", + "Templates", + "Nextcloud.png", + "Nextcloud Intro.mp4", + "Manual Nextcloud.png", + "Readme.md", + "Reasons to use Nextcloud.pdf", + "Template credits.md", +}) + +_DESIGNATED_USER_EMAIL = "testuser@docuvault.example" + +# ── Pytest marker registration ──────────────────────────────────────────────── +# The `live_nextcloud` marker is excluded from ordinary test runs via pytest.ini. +# Select explicitly with: pytest -m live_nextcloud + +pytestmark = pytest.mark.asyncio + + +# ── Credential loading guard ────────────────────────────────────────────────── + +def _load_live_credentials() -> Optional[tuple[str, str, str]]: + """Read live credentials from the environment. + + Returns (url, user, password) or None if any variable is absent. + Never prints values. + """ + url = os.environ.get("NEXTCLOUD_URL", "") + user = os.environ.get("NEXTCLOUD_USER", "") + password = os.environ.get("NEXTCLOUD_APP_PASSWORD", "") + if not (url and user and password): + return None + return url, user, password + + +def _skip_if_missing(): + """Skip the test if live credentials are absent. Keys named, values withheld.""" + creds = _load_live_credentials() + if creds is None: + pytest.skip( + "Live Nextcloud credentials not configured. " + "Set NEXTCLOUD_URL, NEXTCLOUD_USER, and NEXTCLOUD_APP_PASSWORD " + "in the ignored .env file to enable live tests." + ) + return creds + + +# ── Network method guard ────────────────────────────────────────────────────── + +_ALLOWED_METHODS = frozenset({"PROPFIND", "OPTIONS", "HEAD"}) +_FORBIDDEN_METHODS = frozenset({"GET", "PUT", "POST", "PATCH", "DELETE", "MKCOL", "MOVE", "COPY"}) + +# Stable error codes for transport failures — never include URL or credentials +_TRANSPORT_ERROR_CODE = "TRANSPORT_ERROR" +_AUTH_ERROR_CODE = "AUTH_ERROR" +_PARSE_ERROR_CODE = "PARSE_ERROR" + + +class _NetworkMethodGuard: + """Intercept outbound HTTP requests and reject forbidden methods. + + Blocks byte-download (GET on file content) and all mutation verbs before + network dispatch. PROPFIND (metadata listing) and OPTIONS/HEAD are allowed. + + This guard wraps the webdav4 httpx client so the check occurs before any + network activity. + """ + + def __init__(self, wrapped): + self._wrapped = wrapped + + async def request(self, method: str, *args, **kwargs): + if method.upper() in _FORBIDDEN_METHODS: + raise AssertionError( + f"Network guard: {method.upper()} request blocked — " + "live tests are read-only metadata only." + ) + return await self._wrapped.request(method, *args, **kwargs) + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + pass + + +def _assert_no_secrets_in_string(text: str) -> None: + """Assert that no loaded credential values appear in a string. + + Called after capturing pytest output to enforce T-12.1-16. + Only runs when credentials are loaded; skips if env vars are absent. + """ + creds = _load_live_credentials() + if creds is None: + return + url, user, password = creds + # Check for password (highest risk) and username (medium risk). + # Do NOT check for URL substring — the URL contains the test server hostname + # which is allowed to appear in sanitized form in some log contexts. + # We specifically prohibit the literal credential values. + for secret in (password, user): + if secret in text: + pytest.fail( + "T-12.1-16 VIOLATION: a live credential value was found in captured output. " + "Check that no fixture, log, assertion, or exception propagates credential strings." + ) + + +# ── Helper: safe transport error → stable code ──────────────────────────────── + +def _stable_error_code(exc: Exception) -> str: + """Map a transport exception to a stable code without leaking details.""" + msg = str(exc).lower() + if "401" in msg or "403" in msg or "unauthorized" in msg or "forbidden" in msg: + return _AUTH_ERROR_CODE + if "xml" in msg or "parse" in msg or "multistatus" in msg: + return _PARSE_ERROR_CODE + return _TRANSPORT_ERROR_CODE + + +# ── Test 1: Adapter read-only root metadata ─────────────────────────────────── + +@pytest.mark.live_nextcloud +async def test_nextcloud_adapter_read_only_root_metadata(capfd): + """Verify the Nextcloud adapter lists root metadata without downloading bytes. + + Directly invokes the production WebDAV adapter through the canonical contract. + Asserts no byte content is fetched, no mutation method is called, and the + listing returns a CloudListing with items. + """ + creds = _skip_if_missing() + url, user, password = creds + + from storage.cloud_backend_factory import build_cloud_resource_adapter + from storage.cloud_utils import normalize_nextcloud_url + from storage.cloud_base import CloudListing + + # Normalize URL through production helper (idempotent, SSRF-validated) + try: + canonical_url = normalize_nextcloud_url(url, user) + except ValueError as exc: + code = _stable_error_code(exc) + pytest.skip(f"URL normalization failed [{code}] — check NEXTCLOUD_URL format.") + + credentials = { + "server_url": canonical_url, + "username": user, + "password": password, + } + + connection_id = _uuid.uuid4() + user_id = _uuid.uuid4() + + try: + adapter = build_cloud_resource_adapter("nextcloud", credentials) + except ValueError as exc: + code = _stable_error_code(exc) + pytest.skip(f"Adapter construction failed [{code}].") + + # Assert no byte or mutation methods are accessible on the adapter + for method_name in ("get_object", "put_object", "delete_object", + "upload_to", "download_from", "copy", "move", + "create_folder", "rename"): + assert not callable(getattr(type(adapter), method_name, None)), ( + f"T-12.1-17: adapter exposes mutation method {method_name!r}" + ) + + try: + listing = await adapter.list_folder( + connection_id=connection_id, + user_id=user_id, + parent_ref=None, + page_token=None, + ) + except Exception as exc: + code = _stable_error_code(exc) + pytest.fail( + f"Adapter list_folder failed [{code}]. " + "Check that NEXTCLOUD_URL, NEXTCLOUD_USER, and NEXTCLOUD_APP_PASSWORD " + "are correctly set in .env." + ) + + # Assert structural contract + assert isinstance(listing, CloudListing), ( + f"list_folder must return CloudListing, got {type(listing).__name__}" + ) + + # Assert items carry trusted caller identity (T-12.1-18) + for item in listing.items: + assert item.connection_id == connection_id, ( + "Item connection_id must equal the trusted caller connection_id" + ) + assert item.user_id == user_id, ( + "Item user_id must equal the trusted caller user_id" + ) + assert item.kind in ("file", "folder"), ( + f"item.kind must be 'file' or 'folder', got {item.kind!r}" + ) + + # Assert captured output has no secrets + captured = capfd.readouterr() + _assert_no_secrets_in_string(captured.out) + _assert_no_secrets_in_string(captured.err) + + +# ── Test 2: Nonblocking sanitized root diagnostic ───────────────────────────── + +@pytest.mark.live_nextcloud +async def test_nextcloud_root_diagnostic(capfd, caplog): + """Nonblocking live diagnostic: counts, expected matches, kind breakdown. + + Compares the listing against _CANDIDATE_NAMES in memory only. Does NOT + assert exact equality (blocking: 7/10 discrepancy unconfirmed). Records + manifest_status: unconfirmed. Exits successfully when transport/security/ + read-only invariants pass even if the candidate match count is below 10. + + Output format (sanitized): + Total items returned: N + Expected-candidate matches: M of 10 + Missing expected names: [list from _CANDIDATE_NAMES not found] + Unexpected item count: K (names withheld — T-12.1-20) + Folder count: F + File count: Fi + Manifest status: unconfirmed (pending Task 2 reconciliation) + Complete listing: True/False + """ + creds = _skip_if_missing() + url, user, password = creds + + from storage.cloud_backend_factory import build_cloud_resource_adapter + from storage.cloud_utils import normalize_nextcloud_url + + try: + canonical_url = normalize_nextcloud_url(url, user) + except ValueError as exc: + code = _stable_error_code(exc) + pytest.skip(f"URL normalization [{code}].") + + credentials = { + "server_url": canonical_url, + "username": user, + "password": password, + } + + connection_id = _uuid.uuid4() + user_id = _uuid.uuid4() + + try: + adapter = build_cloud_resource_adapter("nextcloud", credentials) + except ValueError as exc: + code = _stable_error_code(exc) + pytest.skip(f"Adapter construction [{code}].") + + with caplog.at_level(logging.WARNING): + try: + listing = await adapter.list_folder( + connection_id=connection_id, + user_id=user_id, + parent_ref=None, + page_token=None, + ) + except Exception as exc: + code = _stable_error_code(exc) + pytest.fail( + f"list_folder failed [{code}] — " + "verify NEXTCLOUD_URL, NEXTCLOUD_USER, NEXTCLOUD_APP_PASSWORD in .env." + ) + + actual_names = frozenset(item.name for item in listing.items) + matched_expected = _CANDIDATE_NAMES & actual_names + missing_expected = sorted(_CANDIDATE_NAMES - actual_names) + unexpected_count = len(actual_names - _CANDIDATE_NAMES) + total_count = len(listing.items) + folder_count = sum(1 for i in listing.items if i.kind == "folder") + file_count = sum(1 for i in listing.items if i.kind == "file") + + # Emit sanitized diagnostic — never emit unexpected names (T-12.1-20) + print(f"\n--- Nextcloud root diagnostic ---") + print(f"Total items returned: {total_count}") + print(f"Expected-candidate matches: {len(matched_expected)} of {len(_CANDIDATE_NAMES)}") + print(f"Missing expected names: {missing_expected}") + print(f"Unexpected item count: {unexpected_count} (names withheld)") + print(f"Folder count: {folder_count}") + print(f"File count: {file_count}") + print(f"Complete listing: {listing.complete}") + print(f"Manifest status: unconfirmed (pending Task 2 reconciliation)") + print(f"--- end diagnostic ---\n") + + # Transport and security invariants — these must pass regardless of manifest drift + assert listing is not None, "Listing must not be None" + for item in listing.items: + assert item.connection_id == connection_id, ( + f"T-12.1-18: item connection_id mismatch" + ) + assert item.user_id == user_id, ( + f"T-12.1-18: item user_id mismatch" + ) + assert item.kind in ("file", "folder"), ( + f"item kind must be file or folder" + ) + assert item.provider_item_id, "provider_item_id must be non-empty" + assert item.name, "item name must be non-empty" + + # Assert no secrets leaked into captured output or logs + captured = capfd.readouterr() + full_output = captured.out + captured.err + caplog.text + _assert_no_secrets_in_string(full_output) + + # Nonblocking: do NOT assert len(matched_expected) == 10 here. + # That gate is Task 2 (checkpoint:human-verify). + # We do assert that if the listing is complete it has at least 1 item, + # proving basic connectivity and root-listing works. + if listing.complete: + assert total_count >= 1, ( + "Complete listing must contain at least 1 root item for the test account" + ) + + +# ── Test 3: Connection-ID browse as designated user ─────────────────────────── + +@pytest.mark.live_nextcloud +@pytest.mark.asyncio +async def test_nextcloud_connection_id_browse_as_designated_user( + db_session, async_client, capfd +): + """End-to-end browse via GET /api/cloud/connections/{id}/items as testuser. + + Provisions the designated regular user (testuser@docuvault.example) in an + isolated test database, encrypts live credentials through production helpers, + then browses root through the connection-ID API. Also asserts foreign-user + and admin denial (T-12.1-18). + """ + creds = _skip_if_missing() + url, user, password = creds + + from db.models import User, Quota, CloudConnection + from services.auth import hash_password, create_access_token + from storage.cloud_utils import normalize_nextcloud_url, encrypt_credentials + from config import settings + + # Normalize URL through production helpers + try: + canonical_url = normalize_nextcloud_url(url, user) + except ValueError as exc: + code = _stable_error_code(exc) + pytest.skip(f"URL normalization [{code}].") + + # Provision the designated user in the isolated test DB + designated_user_id = _uuid.uuid4() + designated_user = User( + id=designated_user_id, + handle="testuser_live", + email=_DESIGNATED_USER_EMAIL, + password_hash=hash_password("LiveTest123!"), + role="user", + is_active=True, + password_must_change=False, + ) + quota = Quota(user_id=designated_user_id, limit_bytes=104857600, used_bytes=0) + db_session.add(designated_user) + db_session.add(quota) + await db_session.commit() + await db_session.refresh(designated_user) + + # Encrypt live credentials through production HKDF helper + master_key = settings.cloud_creds_key.encode() + live_creds = { + "server_url": canonical_url, + "username": user, + "password": password, + } + creds_enc = encrypt_credentials(master_key, str(designated_user_id), live_creds) + + # Create cloud connection owned by the designated user + conn = CloudConnection( + id=_uuid.uuid4(), + user_id=designated_user_id, + provider="nextcloud", + display_name="Live Nextcloud (test)", + credentials_enc=creds_enc, + status="ACTIVE", + ) + db_session.add(conn) + await db_session.commit() + + # Issue a valid access token bound to the designated user + from tests.conftest import _TEST_USER_AGENT + token = create_access_token(str(designated_user_id), "user", user_agent=_TEST_USER_AGENT) + auth_headers = {"Authorization": f"Bearer {token}", "User-Agent": _TEST_USER_AGENT} + + # Browse root through the connection-ID API + resp = await async_client.get( + f"/api/cloud/connections/{conn.id}/items", + headers=auth_headers, + ) + + assert resp.status_code == 200, ( + f"Expected HTTP 200 from browse endpoint, got {resp.status_code}. " + "Check that the test DB has the correct schema and connection." + ) + data = resp.json() + assert "items" in data, "Response must contain 'items'" + assert "freshness" in data, "Response must contain 'freshness'" + + # Verify items carry correct kind values + for item in data.get("items", []): + assert item.get("kind") in ("file", "folder"), ( + f"API item kind must be 'file' or 'folder', got {item.get('kind')!r}" + ) + assert item.get("provider_item_id"), "API item must have provider_item_id" + + # Assert credentials are excluded from API response (T-12.1-18 / D-06) + response_text = resp.text + assert "credentials_enc" not in response_text, ( + "T-12.1-16: credentials_enc must not appear in browse response" + ) + + # Foreign-user cannot access this connection (IDOR — T-12.1-18) + foreign_user_id = _uuid.uuid4() + foreign_user = User( + id=foreign_user_id, + handle="foreign_live", + email="foreign_live@example.com", + password_hash=hash_password("Testpassword123!"), + role="user", + is_active=True, + password_must_change=False, + ) + foreign_quota = Quota(user_id=foreign_user_id, limit_bytes=104857600, used_bytes=0) + db_session.add(foreign_user) + db_session.add(foreign_quota) + await db_session.commit() + foreign_token = create_access_token(str(foreign_user_id), "user", user_agent=_TEST_USER_AGENT) + foreign_resp = await async_client.get( + f"/api/cloud/connections/{conn.id}/items", + headers={"Authorization": f"Bearer {foreign_token}", "User-Agent": _TEST_USER_AGENT}, + ) + assert foreign_resp.status_code in (403, 404), ( + f"IDOR: foreign user must get 403/404, got {foreign_resp.status_code}" + ) + + # Admin cannot browse user cloud content (D-03 / T-12.1-18) + admin_user_id = _uuid.uuid4() + admin_user = User( + id=admin_user_id, + handle="admin_live", + email="admin_live@example.com", + password_hash=hash_password("Testpassword123!"), + role="admin", + is_active=True, + password_must_change=False, + ) + admin_quota = Quota(user_id=admin_user_id, limit_bytes=104857600, used_bytes=0) + db_session.add(admin_user) + db_session.add(admin_quota) + await db_session.commit() + admin_token = create_access_token(str(admin_user_id), "admin", user_agent=_TEST_USER_AGENT) + admin_resp = await async_client.get( + f"/api/cloud/connections/{conn.id}/items", + headers={"Authorization": f"Bearer {admin_token}", "User-Agent": _TEST_USER_AGENT}, + ) + assert admin_resp.status_code in (403, 404), ( + f"Admin must be blocked from browsing user cloud data, got {admin_resp.status_code}" + ) + + # Assert no secrets in captured output + captured = capfd.readouterr() + _assert_no_secrets_in_string(captured.out) + _assert_no_secrets_in_string(captured.err)