""" Sahih Islam - Subscription Security & Anti-Modding Test Suite ============================================================ 14 Comprehensive Security Scenarios verifying: 1. Server-authoritative Pro entitlement gating 2. Tampered/spoofed client payload resistance 3. Time-based subscription expiry enforcement 4. Instant revocation on Google Play refunds 5. Active subscriber access authorization 6. Google Play In-App Purchase token verification 7. Rejection of malformed/invalid Google Play purchase tokens 8. Anti-Replay protection across multiple user accounts 9. Idempotent purchase verification for the same user 10. RTDN Webhook renewal lifecycle updates in MongoDB 11. RTDN Webhook revocation/refund updates in MongoDB 12. IDOR access control protection (enforce_self_or_admin) 13. Prevention of unverified direct subscription activation 14. Production prohibition of mock/sandbox tokens """ import os import sys import asyncio from datetime import datetime, timedelta from unittest.mock import AsyncMock, patch, MagicMock import pytest from fastapi import HTTPException, status, Request # Ensure backend root on sys.path sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) from app.auth_deps import is_user_pro, require_pro, require_entitlement, enforce_self_or_admin from app.rag_config import resolve_user_tier from app.services.google_play_service import GooglePlayService, google_play_service from app.core.config import settings from app.models import SubscriptionCreate def run_async(coro): """Executes asynchronous coroutine synchronously for pytest.""" return asyncio.run(coro) # ============================================================================ # 1. Server Authority & Tier Resolution Tests (Scenarios 1 - 5) # ============================================================================ def test_free_user_pro_endpoint_denial(): """Scenario 1: Free user attempting to access a Pro-protected endpoint receives 403 Forbidden.""" async def _test(): free_user = {"uid": "user_free_1", "role": "USER", "is_pro": False} assert await is_user_pro(free_user) is False with pytest.raises(HTTPException) as exc_info: await require_pro(current_user=free_user) assert exc_info.value.status_code == status.HTTP_403_FORBIDDEN assert "Pro subscription required" in exc_info.value.detail checker = require_entitlement("ai_deepseek_v3") with pytest.raises(HTTPException) as exc_info2: await checker(current_user=free_user) assert exc_info2.value.status_code == status.HTTP_403_FORBIDDEN assert "ai_deepseek_v3" in exc_info2.value.detail run_async(_test()) def test_client_spoofed_is_pro_ignored(): """Scenario 2: Client passes spoofed 'is_pro': True in request, but DB record has is_pro: False.""" async def _test(): db_user_record = { "uid": "user_spoof_attacker", "role": "USER", "is_pro": False, "subscription": {"is_pro": False, "tier": "FREE", "status": "FREE"} } # Attacker's request body may claim whatever, but the authoritative user dict is evaluated assert await is_user_pro(db_user_record) is False assert resolve_user_tier(db_user_record) == "free" run_async(_test()) def test_expired_subscription_downgrade(): """Scenario 3: User with is_pro=True whose expires_at is in the past is downgraded to free.""" async def _test(): past_expiry = datetime.utcnow() - timedelta(days=3) expired_user = { "uid": "user_expired_sub", "role": "USER", "is_pro": True, # Stale flag left in DB "subscription": { "is_pro": True, "tier": "PRO", "status": "ACTIVE", "expires_at": past_expiry } } assert await is_user_pro(expired_user) is False assert resolve_user_tier(expired_user) == "free" with pytest.raises(HTTPException) as exc_info: await require_pro(current_user=expired_user) assert exc_info.value.status_code == status.HTTP_403_FORBIDDEN run_async(_test()) def test_revoked_or_refunded_subscription(): """Scenario 4: User with status=REFUNDED or REVOKED has Pro access revoked immediately.""" async def _test(): future_date = datetime.utcnow() + timedelta(days=20) refunded_user = { "uid": "user_refunded_1", "role": "USER", "is_pro": True, "subscription": { "is_pro": True, "tier": "PRO", "status": "REFUNDED", "expires_at": future_date } } assert await is_user_pro(refunded_user) is False assert resolve_user_tier(refunded_user) == "free" revoked_user = { "uid": "user_revoked_1", "role": "USER", "subscription": { "is_pro": True, "tier": "PRO", "status": "REVOKED", "expires_at": future_date } } assert await is_user_pro(revoked_user) is False assert resolve_user_tier(revoked_user) == "free" run_async(_test()) def test_valid_subscription_allows_access(): """Scenario 5: Active subscriber with future expires_at passes require_pro and receives 'pro' tier.""" async def _test(): future_date = datetime.utcnow() + timedelta(days=25) valid_pro_user = { "uid": "user_valid_pro", "role": "USER", "is_pro": True, "subscription": { "is_pro": True, "tier": "PRO", "status": "ACTIVE", "expires_at": future_date } } assert await is_user_pro(valid_pro_user) is True assert resolve_user_tier(valid_pro_user) == "pro" # require_pro returns the user without raising result = await require_pro(current_user=valid_pro_user) assert result["uid"] == "user_valid_pro" run_async(_test()) # ============================================================================ # 2. Google Play Developer API Verification & Anti-Replay (Scenarios 6 - 9) # ============================================================================ def test_google_play_purchase_verification_success(): """Scenario 6: Valid Google Play purchase token verifies and activates subscription.""" async def _test(): from app.payment_routes import verify_google_play_purchase mock_user = {"uid": "user_gp_buyer_1"} token = "test_token_valid_purchase_1001" payload = { "purchase_token": token, "product_id": "sahih_pro_monthly", "package_name": "com.sahihislam.app", "purpose": "subscription", "amount": 199.0, } with patch("app.payment_routes.subscriptions_collection.find_one", new_callable=AsyncMock) as mock_find_sub, \ patch("app.payment_routes.execute_payment_fulfillment", new_callable=AsyncMock) as mock_fulfill, \ patch("app.payment_routes.invalidate_user_cache", new_callable=AsyncMock): mock_find_sub.return_value = None # Token not yet claimed mock_fulfill.return_value = {"is_pro": True, "tier": "PRO"} response = await verify_google_play_purchase(payload=payload, current_user=mock_user) assert response["success"] is True assert response["product_id"] == "sahih_pro_monthly" assert mock_fulfill.called call_kwargs = mock_fulfill.call_args[1] assert call_kwargs["metadata"]["purchase_token"] == token run_async(_test()) def test_google_play_purchase_verification_invalid_token(): """Scenario 7: Malformed or fraudulent Google Play purchase token is rejected with 400.""" async def _test(): from app.payment_routes import verify_google_play_purchase mock_user = {"uid": "user_gp_fraud_1"} payload = { "purchase_token": "mock_invalid_fraud_token_xyz", "product_id": "sahih_pro_monthly", "package_name": "com.sahihislam.app", "purpose": "subscription", } with patch("app.payment_routes.subscriptions_collection.find_one", new_callable=AsyncMock) as mock_find_sub: mock_find_sub.return_value = None with pytest.raises(HTTPException) as exc_info: await verify_google_play_purchase(payload=payload, current_user=mock_user) assert exc_info.value.status_code == status.HTTP_400_BAD_REQUEST assert "Invalid" in exc_info.value.detail or "inactive" in exc_info.value.detail run_async(_test()) def test_anti_replay_token_reuse_different_user(): """Scenario 8: Token already claimed by User A is submitted by User B -> 409 Conflict.""" async def _test(): from app.payment_routes import verify_google_play_purchase token = "test_token_already_claimed_by_alice" existing_claim = { "purchase_token": token, "user_id": "user_alice", "plan_id": "sahih_pro_monthly", "renews_at": datetime.utcnow() + timedelta(days=20), } attacker_user = {"uid": "user_bob_attacker"} payload = { "purchase_token": token, "product_id": "sahih_pro_monthly", "package_name": "com.sahihislam.app", "purpose": "subscription", } with patch("app.payment_routes.subscriptions_collection.find_one", new_callable=AsyncMock) as mock_find_sub: mock_find_sub.return_value = existing_claim with pytest.raises(HTTPException) as exc_info: await verify_google_play_purchase(payload=payload, current_user=attacker_user) assert exc_info.value.status_code == status.HTTP_409_CONFLICT assert "already been claimed by another account" in exc_info.value.detail run_async(_test()) def test_idempotent_token_submission_same_user(): """Scenario 9: Same user re-submitting an already verified token receives idempotent 200 OK without duplicate billing.""" async def _test(): from app.payment_routes import verify_google_play_purchase token = "test_token_same_user_repeat" user_alice = {"uid": "user_alice"} expiry = datetime.utcnow() + timedelta(days=20) existing_claim = { "purchase_token": token, "user_id": "user_alice", "transaction_id": "GPA.PREVIOUS_TXN_001", "plan_id": "sahih_pro_monthly", "renews_at": expiry, } payload = { "purchase_token": token, "product_id": "sahih_pro_monthly", "package_name": "com.sahihislam.app", "purpose": "subscription", } with patch("app.payment_routes.subscriptions_collection.find_one", new_callable=AsyncMock) as mock_find_sub, \ patch("app.payment_routes.execute_payment_fulfillment", new_callable=AsyncMock) as mock_fulfill: mock_find_sub.return_value = existing_claim res = await verify_google_play_purchase(payload=payload, current_user=user_alice) assert res["success"] is True assert res["status"] == "already_verified" assert res["order_id"] == "GPA.PREVIOUS_TXN_001" assert not mock_fulfill.called run_async(_test()) # ============================================================================ # 3. Google RTDN Webhook Tests (Scenarios 10 - 11) # ============================================================================ def test_rtdn_renewal_webhook_updates_user(): """Scenario 10: RTDN notificationType=2 (SUBSCRIPTION_RENEWED) updates subscription and user expiry in MongoDB.""" async def _test(): from app.payment_routes import google_play_rtdn_webhook token = "test_rtdn_renew_token_55" sub_record = { "purchase_token": token, "user_id": "user_subscriber_99", "plan_id": "sahih_pro_monthly", "status": "active" } mock_request = AsyncMock(spec=Request) mock_request.json = AsyncMock(return_value={ "subscriptionNotification": { "notificationType": 2, # RENEWED "purchaseToken": token, "subscriptionId": "sahih_pro_monthly", } }) with patch("app.payment_routes.subscriptions_collection.find_one", new_callable=AsyncMock) as mock_find_sub, \ patch("app.payment_routes.subscriptions_collection.update_one", new_callable=AsyncMock) as mock_sub_update, \ patch("app.payment_routes.users_collection.update_one", new_callable=AsyncMock) as mock_user_update, \ patch("app.payment_routes.transactions_collection.insert_one", new_callable=AsyncMock), \ patch("app.payment_routes.invalidate_user_cache", new_callable=AsyncMock) as mock_invalidate: mock_find_sub.return_value = sub_record res = await google_play_rtdn_webhook(request=mock_request) assert res["status"] == "acknowledged" assert mock_sub_update.called sub_update_call = mock_sub_update.call_args[0] assert sub_update_call[0] == {"purchase_token": token} assert sub_update_call[1]["$set"]["status"] == "active" assert mock_user_update.called user_update_call = mock_user_update.call_args[0] assert user_update_call[0] == {"uid": "user_subscriber_99"} assert user_update_call[1]["$set"]["is_pro"] is True assert user_update_call[1]["$set"]["subscription.status"] == "ACTIVE" mock_invalidate.assert_called_with("user_subscriber_99") run_async(_test()) def test_rtdn_refund_webhook_revokes_pro(): """Scenario 11: RTDN notificationType=12 (SUBSCRIPTION_REVOKED) immediately strips Pro in user record.""" async def _test(): from app.payment_routes import google_play_rtdn_webhook token = "test_rtdn_refund_token_77" sub_record = { "purchase_token": token, "user_id": "user_refund_target", "plan_id": "sahih_pro_monthly", "status": "active" } mock_request = AsyncMock(spec=Request) mock_request.json = AsyncMock(return_value={ "subscriptionNotification": { "notificationType": 12, # REVOKED / REFUND "purchaseToken": token, "subscriptionId": "sahih_pro_monthly", } }) with patch("app.payment_routes.subscriptions_collection.find_one", new_callable=AsyncMock) as mock_find_sub, \ patch("app.payment_routes.subscriptions_collection.update_one", new_callable=AsyncMock) as mock_sub_update, \ patch("app.payment_routes.users_collection.update_one", new_callable=AsyncMock) as mock_user_update, \ patch("app.payment_routes.transactions_collection.insert_one", new_callable=AsyncMock), \ patch("app.payment_routes.invalidate_user_cache", new_callable=AsyncMock) as mock_invalidate: mock_find_sub.return_value = sub_record res = await google_play_rtdn_webhook(request=mock_request) assert res["status"] == "acknowledged" assert mock_sub_update.called assert mock_sub_update.call_args[0][1]["$set"]["status"] == "refunded" assert mock_user_update.called user_update_doc = mock_user_update.call_args[0][1]["$set"] assert user_update_doc["is_pro"] is False assert user_update_doc["subscription.is_pro"] is False assert user_update_doc["subscription.status"] == "REVOKED" mock_invalidate.assert_called_with("user_refund_target") run_async(_test()) # ============================================================================ # 4. Access Control, IDOR & Production Isolation (Scenarios 12 - 14) # ============================================================================ def test_idor_protection_enforce_self_or_admin(): """Scenario 12: enforce_self_or_admin prevents User A from acting on User B's resource.""" user_a = {"uid": "user_alpha", "role": "USER"} user_b_uid = "user_beta" admin_user = {"uid": "superadmin_1", "role": "ADMIN"} # User A acting on own resource -> OK enforce_self_or_admin(user_a, target_uid="user_alpha") # User A acting on User B -> 403 Forbidden with pytest.raises(HTTPException) as exc_info: enforce_self_or_admin(user_a, target_uid=user_b_uid) assert exc_info.value.status_code == status.HTTP_403_FORBIDDEN assert "Cannot act on another user's resource" in exc_info.value.detail # Admin acting on User B -> OK enforce_self_or_admin(admin_user, target_uid=user_b_uid) def test_direct_subscribe_endpoint_locked(): """Scenario 13: Direct unverified subscription payload is rejected with 400 Bad Request.""" async def _test(): from app.payment_routes import subscribe_user mock_user = {"uid": "user_direct_cheat"} payload = SubscriptionCreate( plan_id="sahih_pro_annual", billing_cycle="annual", payment_method="direct_cheat_bypass", amount_paid=0.0, ) with pytest.raises(HTTPException) as exc_info: await subscribe_user(payload=payload, current_user=mock_user) assert exc_info.value.status_code == status.HTTP_400_BAD_REQUEST assert "Direct unverified subscription activation is disabled" in exc_info.value.detail run_async(_test()) def test_production_mock_token_forbidden(): """Scenario 14: In production mode (settings.IS_PROD=True), mock purchase tokens are rejected with 403.""" async def _test(): gps = GooglePlayService() with patch.object(settings, "IS_PROD", True), \ patch.dict(os.environ, {"TEST_MODE": "0", "ALLOW_MOCK_GOOGLE_PLAY": "0"}): with pytest.raises(HTTPException) as exc_info: await gps.verify_purchase( package_name="com.sahihislam.app", product_id="sahih_pro_monthly", purchase_token="mock_cheat_token_123", is_subscription=True ) assert exc_info.value.status_code == status.HTTP_403_FORBIDDEN assert "Mock purchase tokens are strictly forbidden in production" in exc_info.value.detail run_async(_test())