""" Purchase Validation and Anti-Cheat Handles rate limiting, idempotency, and IAP validation. """ import hashlib import time import json from functools import wraps from typing import Dict, Any, Optional import logging from database import get_database_connection def rate_limit(max_calls: int, time_window: int): """ Rate limiting decorator. Args: max_calls: Maximum calls allowed time_window: Time window in seconds Usage: @rate_limit(max_calls=10, time_window=60) def some_function(player_id: int, ...): ... """ calls = {} def decorator(func): @wraps(func) def wrapper(player_id: int, *args, **kwargs): now = time.time() # Clean old entries if player_id in calls: calls[player_id] = [t for t in calls[player_id] if now - t < time_window] else: calls[player_id] = [] # Check rate limit if len(calls[player_id]) >= max_calls: logging.warning(f"Rate limit exceeded for player {player_id} on {func.__name__}") return { 'success': False, 'message': 'Too many requests. Please wait and try again.', 'error_code': 'RATE_LIMIT_EXCEEDED' } # Add this call calls[player_id].append(now) return func(player_id, *args, **kwargs) return wrapper return decorator class IdempotencyManager: """ Prevent double-processing of transactions (especially IAP receipts). """ @staticmethod def generate_key(player_id: int, transaction_id: str) -> str: """ Generate idempotency key from transaction. Args: player_id: The player's ID transaction_id: Unique transaction ID Returns: SHA256 hash as idempotency key """ return hashlib.sha256(f"{player_id}:{transaction_id}".encode()).hexdigest() @staticmethod def check_processed(idempotency_key: str) -> bool: """ Check if transaction already processed. Args: idempotency_key: The idempotency key Returns: True if already processed, False otherwise """ conn = None cursor = None try: conn = get_database_connection() cursor = conn.cursor(dictionary=True) cursor.execute( "SELECT COUNT(*) as count FROM processed_transactions WHERE idempotency_key = %s", (idempotency_key,) ) result = cursor.fetchone() return result['count'] > 0 except Exception as e: logging.error(f"Error checking idempotency key: {e}") return False finally: if cursor: cursor.close() if conn: conn.close() @staticmethod def mark_processed(idempotency_key: str, player_id: int, transaction_data: Dict[str, Any]): """ Mark transaction as processed. Args: idempotency_key: The idempotency key player_id: The player's ID transaction_data: Transaction data to store """ conn = None cursor = None try: conn = get_database_connection() cursor = conn.cursor() cursor.execute( """INSERT INTO processed_transactions (idempotency_key, player_id, transaction_data) VALUES (%s, %s, %s)""", (idempotency_key, player_id, json.dumps(transaction_data)) ) conn.commit() except Exception as e: if conn: conn.rollback() logging.error(f"Error marking transaction as processed: {e}") finally: if cursor: cursor.close() if conn: conn.close() def validate_iap_receipt(player_id: int, receipt_data: str, transaction_id: str, product_id: str) -> Dict[str, Any]: """ Validate IAP purchase with Apple. Args: player_id: The player's ID receipt_data: Base64 encoded receipt from Apple transaction_id: Unique transaction ID product_id: Product identifier Returns: dict with 'success' and 'diamonds_awarded' keys """ # Generate idempotency key idem_key = IdempotencyManager.generate_key(player_id, transaction_id) # Check if already processed if IdempotencyManager.check_processed(idem_key): logging.warning(f"Duplicate IAP transaction attempted: {transaction_id}") return { 'success': False, 'message': 'Transaction already processed', 'error_code': 'DUPLICATE_TRANSACTION' } # Product ID to diamond mapping diamond_packages = { 'com.baolife.diamonds.small': 100, 'com.baolife.diamonds.medium': 300, 'com.baolife.diamonds.large': 750, 'com.baolife.diamonds.mega': 2000, } try: # TODO: Actual validation with Apple's receipt validation API # For now, we'll assume validation succeeds # In production, you would: # 1. Send receipt_data to Apple's validation endpoint # 2. Parse and verify the response # 3. Check product_id matches # 4. Verify transaction hasn't been used before validation_result = {'status': 0} # 0 = valid (Apple's format) if validation_result['status'] == 0: # Mark as processed IdempotencyManager.mark_processed(idem_key, player_id, { 'transaction_id': transaction_id, 'product_id': product_id, 'validation_status': 'success' }) # Award diamonds based on product ID diamond_amount = diamond_packages.get(product_id, 0) if diamond_amount > 0: from .diamond_economy import award_diamonds award_diamonds(player_id, f'iap_purchase_{product_id}', diamond_amount) return { 'success': True, 'diamonds_awarded': diamond_amount, 'message': f'Successfully purchased {diamond_amount} diamonds' } else: return { 'success': False, 'message': 'Invalid product ID', 'error_code': 'INVALID_PRODUCT' } else: logging.error(f"IAP validation failed for player {player_id}: status {validation_result['status']}") return { 'success': False, 'message': 'Purchase validation failed', 'error_code': 'VALIDATION_FAILED' } except Exception as e: logging.error(f"Error validating IAP for player {player_id}: {e}", exc_info=True) return { 'success': False, 'message': 'Server error during validation', 'error_code': 'SERVER_ERROR' } def handle_validate_purchase(player_id: int, message_data: Dict[str, Any], send_to_client): """ WebSocket handler for IAP validation. Args: player_id: The player's ID message_data: Message containing receipt_data, transaction_id, product_id send_to_client: Function to send response to client """ receipt_data = message_data.get('receipt_data') transaction_id = message_data.get('transaction_id') product_id = message_data.get('product_id') if not all([receipt_data, transaction_id, product_id]): send_to_client(player_id, { 'type': 'error', 'error_code': 'INVALID_REQUEST', 'message': 'Missing required fields for IAP validation' }) return result = validate_iap_receipt(player_id, receipt_data, transaction_id, product_id) if result['success']: send_to_client(player_id, { 'type': 'purchaseValidated', 'success': True, 'diamonds_awarded': result['diamonds_awarded'], 'message': result['message'] }) # Send diamond balance update from .diamond_economy import get_diamond_balance new_balance = get_diamond_balance(player_id) send_to_client(player_id, { 'type': 'diamondUpdate', 'balance': new_balance }) else: send_to_client(player_id, { 'type': 'error', 'error_code': result.get('error_code', 'PURCHASE_FAILED'), 'message': result['message'] })