diff --git a/backend/tests/test_tenant_isolation_p0.py b/backend/tests/test_tenant_isolation_p0.py index 79018dfe..1c451f4e 100644 --- a/backend/tests/test_tenant_isolation_p0.py +++ b/backend/tests/test_tenant_isolation_p0.py @@ -212,3 +212,197 @@ async def test_ai_session_search_cannot_see_other_users_sessions( assert str(session_b.id) not in ids, ( "User A can see User B's session via search — cross-user leak within account" ) + + +# ── Task 6: Cross-tenant UUID audit ───────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_get_tree_returns_404_not_403_for_other_account( + client: AsyncClient, test_db: AsyncSession +): + """Account A gets 404 (not 403) when accessing Account B's private tree.""" + acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-tree-a") + acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-tree-b") + tree_b = await _create_private_tree(test_db, acct_b, user_b) + await test_db.commit() + + headers_a = await _login(client, user_a.email, pass_a) + resp = await client.get(f"/api/v1/trees/{tree_b.id}", headers=headers_a) + assert resp.status_code == 404, ( + f"Expected 404 for cross-tenant tree access, got {resp.status_code}" + ) + + +@pytest.mark.asyncio +async def test_update_tree_returns_404_not_403_for_other_account( + client: AsyncClient, test_db: AsyncSession +): + """Account A gets 404 (not 403) when trying to update Account B's tree.""" + acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-upd-a") + acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-upd-b") + tree_b = await _create_private_tree(test_db, acct_b, user_b) + await test_db.commit() + + headers_a = await _login(client, user_a.email, pass_a) + resp = await client.put( + f"/api/v1/trees/{tree_b.id}", + json={"name": "Hacked"}, + headers=headers_a, + ) + assert resp.status_code == 404, ( + f"Expected 404 for cross-tenant tree update, got {resp.status_code}" + ) + + +@pytest.mark.asyncio +async def test_get_session_returns_404_not_403_for_other_user( + client: AsyncClient, test_db: AsyncSession +): + """User A gets 404 (not 403) when accessing User B's session.""" + from app.models.session import Session + + acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-sess-a") + acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-sess-b") + tree_b = await _create_private_tree(test_db, acct_b, user_b) + + session_b = Session( + tree_id=tree_b.id, + user_id=user_b.id, + tree_snapshot={"id": "root", "type": "start", "children": []}, + path_taken=[], + decisions=[], + ) + test_db.add(session_b) + await test_db.commit() + + headers_a = await _login(client, user_a.email, pass_a) + resp = await client.get(f"/api/v1/sessions/{session_b.id}", headers=headers_a) + assert resp.status_code == 404, ( + f"Expected 404 for cross-user session access, got {resp.status_code}" + ) + + +@pytest.mark.asyncio +async def test_ai_session_get_returns_404_not_403_for_other_user( + client: AsyncClient, test_db: AsyncSession +): + """User A gets 404 (not 403) when accessing User B's AI session.""" + from app.models.ai_session import AISession + + acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-ais-a") + acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-ais-b") + + ai_session_b = AISession( + user_id=user_b.id, + account_id=acct_b.id, + problem_summary="Test session", + problem_domain="networking", + status="active", + ) + test_db.add(ai_session_b) + await test_db.commit() + + headers_a = await _login(client, user_a.email, pass_a) + resp = await client.get(f"/api/v1/ai-sessions/{ai_session_b.id}", headers=headers_a) + assert resp.status_code == 404, ( + f"Expected 404 for cross-user AI session access, got {resp.status_code}" + ) + + +@pytest.mark.asyncio +async def test_ai_session_retry_psa_push_requires_ownership( + client: AsyncClient, test_db: AsyncSession +): + """User A cannot retry PSA push for User B's AI session.""" + from app.models.ai_session import AISession + + acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-psa-a") + acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-psa-b") + + ai_session_b = AISession( + user_id=user_b.id, + account_id=acct_b.id, + problem_summary="PSA test", + problem_domain="networking", + status="resolved", + ) + test_db.add(ai_session_b) + await test_db.commit() + + headers_a = await _login(client, user_a.email, pass_a) + resp = await client.post( + f"/api/v1/ai-sessions/{ai_session_b.id}/retry-psa-push", + headers=headers_a, + ) + assert resp.status_code == 404, ( + f"Expected 404 for cross-user retry-psa-push, got {resp.status_code}" + ) + + +@pytest.mark.asyncio +async def test_upload_url_returns_404_not_403_for_other_account( + client: AsyncClient, test_db: AsyncSession +): + """User A gets 404 (not 403) when accessing User B's upload URL.""" + from app.models.file_upload import FileUpload + + acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-upl-a") + acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-upl-b") + + upload_b = FileUpload( + account_id=acct_b.id, + uploaded_by=user_b.id, + filename="secret.png", + content_type="image/png", + size_bytes=1024, + storage_key="test/secret.png", + ) + test_db.add(upload_b) + await test_db.commit() + + headers_a = await _login(client, user_a.email, pass_a) + resp = await client.get(f"/api/v1/uploads/{upload_b.id}/url", headers=headers_a) + assert resp.status_code in (404, 503), ( + f"Expected 404 (or 503 if storage not configured) for cross-account upload, got {resp.status_code}" + ) + + +@pytest.mark.asyncio +async def test_share_revoke_returns_404_not_403_for_other_user( + client: AsyncClient, test_db: AsyncSession +): + """User A gets 404 (not 403) when revoking User B's share.""" + from app.models.session import Session + from app.models.session_share import SessionShare + + acct_a, user_a, pass_a = await _create_account_and_user(test_db, "t6-shr-a") + acct_b, user_b, pass_b = await _create_account_and_user(test_db, "t6-shr-b") + tree_b = await _create_private_tree(test_db, acct_b, user_b) + + session_b = Session( + tree_id=tree_b.id, + user_id=user_b.id, + tree_snapshot={"id": "root", "type": "start", "children": []}, + path_taken=[], + decisions=[], + ) + test_db.add(session_b) + await test_db.flush() + + share_b = SessionShare( + session_id=session_b.id, + account_id=acct_b.id, + share_token="test-token-unique-" + uuid.uuid4().hex[:8], + share_name="Test", + visibility="public", + created_by=user_b.id, + ) + test_db.add(share_b) + await test_db.commit() + + headers_a = await _login(client, user_a.email, pass_a) + resp = await client.delete(f"/api/v1/shares/{share_b.id}", headers=headers_a) + assert resp.status_code == 404, ( + f"Expected 404 for cross-user share revoke, got {resp.status_code}" + )