import os from datetime import datetime, timezone import requests from django.conf import settings from core.logger import logger """Apple IAP verification. Implements both legacy verifyReceipt flow and JWS (Server Notifications v2 / StoreKit 2). Env vars expected: - APPLE_SHARE_SECRET (for legacy auto-renew subscriptions) - APPLE_APP_BUNDLE_ID (bundle identifier) - APPLE_IAP_PRIVATE_KEY_P8, APPLE_KEY_ID, APPLE_ISSUER_ID (for App Store Server API if needed) """ PROD_URL = "https://buy.itunes.apple.com/verifyReceipt" SANDBOX_URL = "https://sandbox.itunes.apple.com/verifyReceipt" APPLE_STATUS = { 0: "Success", 21000: "The App Store could not read the JSON object you provided", 21002: "The data in the receipt-data property was malformed or missing", 21003: "The receipt could not be authenticated", 21004: "The shared secret you provided does not match", 21005: "The receipt server is not currently available", 21006: "This receipt is valid but subscription has expired", 21007: "Sandbox receipt sent to production", 21008: "Production receipt sent to sandbox", 21009: "Internal data access error", 21010: "User not found / no receipt" } import os from appstoreserverlibrary.models.Environment import Environment from appstoreserverlibrary.signed_data_verifier import VerificationException, SignedDataVerifier def load_apple_root_certs(): dir_path = 'certs' # List of certificate filenames in the same order as the JavaScript version files = ['AppleRootCA-G3.cer', 'AppleRootCA-G2.cer', 'AppleRootCA.cer'] bufs: list[bytes] = [] for f in files: p = os.path.join(dir_path, f) with open(p, "rb") as fh: bufs.append(fh.read()) return bufs # def verify_apple_receipt(signed_notification): # try: # root_certificates = load_apple_root_certs() # signed_data_verifier = SignedDataVerifier( # root_certificates, # enable_online_checks=True, # environment=Environment.SANDBOX, # bundle_id="com.premiumcrap.app", # app_apple_id=6752025955 # ) # # payload = signed_data_verifier.verify_and_decode_notification(signed_notification) # print("Decoded Payload:", payload) # return payload # # except VerificationException as e: # print(e, '>>>>>>>>>>>>>>>>>>>>>>>>>>>>Verification failed ') # except Exception as e: # print(f"General Exception: {e}") def verify_apple_receipt(base64_receipt): try: payload = { "receipt-data": base64_receipt, "password": settings.APPLE_SHARE_SECRET, "exclude-old-transactions": False, } def post(url): resp = requests.post(url, json=payload, timeout=15) return resp.json() # 1) Try Production environment = "Production" data = post(PROD_URL) # 2) Retry in Sandbox if status == 21007 if data.get("status") == 21007: environment = "Sandbox" data = post(SANDBOX_URL) if data.get("status") != 0: raise Exception( f"verifyReceipt failed: {APPLE_STATUS.get(data.get('status'), 'Unknown')}", data, environment ) data = {"data": data, "environment": environment, "status": data.get("status")} logger.info(f"verify_apple_receipt: {data}") return data except Exception as e: print(e, '>>>>>>>>>>>>>>>>>>>>>>ERROR') logger.info(f"verify_apple_receipt: {e}") logger.error(f"verify_apple_receipt", exc_info=True) def verify_apple_purchase( base64_receipt=None, signed_payload=None, shared_secret=None, exclude_old_transactions=True, timeout_ms=10000, server_api=None, ): logger.info(f">>>>>>>>>>>>>>>>>>>>>>in verify_apple_purchase") try: if signed_payload: roots = load_apple_root_certs() bundle_id = server_api.get("bundleId") app_apple_id = server_api.get("keyId") verifier = SignedDataVerifier( root_certificates=roots, enable_online_checks=True, environment=Environment.SANDBOX, bundle_id=bundle_id, app_apple_id=app_apple_id, ) notif = verifier.verify_and_decode_notification(signed_payload) env_str = "Production" if notif.data.environment == Environment.PRODUCTION else "Sandbox" if env_str == "Production": logger.info(f">>>>>>>>>>>>>>>>>>>>>>in production") verifier = SignedDataVerifier( root_certificates=roots, enable_online_checks=True, environment=Environment.PRODUCTION, bundle_id=bundle_id, app_apple_id=app_apple_id, ) txn = None if notif.data.signedTransactionInfo: verifier = SignedDataVerifier( root_certificates=roots, enable_online_checks=True, environment=Environment.SANDBOX, bundle_id=bundle_id, app_apple_id=None, ) txn = verifier.verify_and_decode_signed_transaction(notif.data.signedTransactionInfo) expires_ms = int(txn.expiresDate) if txn and txn.expiresDate else None current_ms = int(datetime.now().timestamp() * 1000) is_active = expires_ms > current_ms transaction = { "originalTransactionId": txn.originalTransactionId if txn.originalTransactionId else None, "appleTransactionId": txn.transactionId if txn.transactionId else None, "productId": txn.productId if txn.productId else None, "startAt": datetime.fromtimestamp(txn.purchaseDate / 1000, tz=timezone.utc), "expireAt": datetime.fromtimestamp(txn.expiresDate / 1000, tz=timezone.utc), "transaction_reason": txn.transactionReason if txn.transactionReason else None, "web_order_line_item_id": txn.webOrderLineItemId if txn.webOrderLineItemId else None } return { "ok": True, "method": "notification-jws", "environment": env_str, "transaction": transaction, "isActive": is_active, "notificationType": notif.notificationType.value } elif base64_receipt: raise NotImplementedError("Legacy base64 receipt verification not implemented yet.") else: raise ValueError("Provide either base64Receipt or signedPayload") except VerificationException as e: print(e, '>>>>>>>>>>>>>>>>>>>>>>>>>>>>Verification ERROR ') logger.info(f"Verification Exception>>>>>>>>>>>>>>>>>>>>>>>>>{e}") except Exception as e: print(f"General Exception: {e}") logger.info(f"General Exception>>>>>>>>>>>>>>>>>>>>>>>>>{e}") # import json # import requests # from django.conf import settings # from rest_framework import status # # sandbox_url = 'https://sandbox.itunes.apple.com/verifyReceipt' # production_url = 'https://buy.itunes.apple.com/verifyReceipt' # # # def verify_receipt(url, token): # payload = {"receipt-data": token, "password": settings.APPLE_SHARED_SECRET, "exclude-old-transactions": True} # headers = {'content-type': 'application/json'} # response = requests.post(url, data=json.dumps(payload), headers=headers) # if response.status_code == status.HTTP_200_OK: # data = response.json() # if data['status'] == 21007: # verify_receipt(production_url, token) # else: # return response # else: # return response # # # def renew_transaction_by_token(token): # response = verify_receipt(sandbox_url, token) # if response.status_code == status.HTTP_200_OK: # data = response.json() # if data['status'] == 0: # latest_info = data['latest_receipt_info'][0] # expireAt = latest_info["expires_date"] # return expireAt # else: # raise Exception('Token is not valid, check status code:{}'.format(data['status'])) # else: # raise Exception('Token is not valid, please try again later')