# import json # import base64 # from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes # from cryptography.hazmat.backends import default_backend # import os # import time # def encrypt_billdesk_payment(merc_id, amount, return_url, encryption_key, key_id, client_id, # currency="356", itemcode="DIRECT", user_ip="123.0.0.1"): # """ # Create and encrypt a BillDesk payment payload # Args: # merc_id (str): Merchant ID provided by BillDesk # amount (str): Payment amount # return_url (str): Return URL after payment # encryption_key (str): BillDesk provided encryption key # key_id (str): BillDesk provided encryption key ID # client_id (str): BillDesk provided client ID # currency (str): Currency code (default: "356" for INR) # itemcode (str): Item code (default: "DIRECT") # user_ip (str): User's IP address # Returns: # str: JWE encrypted payment data # """ # # Create the payment payload # payload = { # "orderid": f"TEST{int(time.time())}", # unique order id # "mercid": merc_id, # "order_date": time.strftime("%Y-%m-%dT%H:%M:%S+05:30"), # "amount": amount, # "currency": currency, # INR # "ru": return_url, # "itemcode": itemcode, # "device": { # "init_channel": "internet", # "ip": user_ip, # "user_agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:51.0) Gecko/20100101 Firefox/51.0", # "accept_header": "application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9" # } # } # # Convert to JSON string # payload_json = json.dumps(payload, separators=(',', ':')) # # Encrypt and return # return encrypt_jwe_billdesk(payload_json, encryption_key, key_id, client_id) # def encrypt_jwe_billdesk(response_string, encryption_key, key_id, client_id): # """ # Encrypt data using JWE with AES-256-GCM for BillDesk integration # Args: # response_string (str): The payload to encrypt # encryption_key (str): BillDesk provided encryption key # key_id (str): BillDesk provided encryption key ID # client_id (str): BillDesk provided client ID # Returns: # str: JWE encrypted string # """ # # Convert encryption key to bytes (equivalent to encryptionKey.getBytes()) # key = encryption_key.encode('utf-8') # # Ensure key is 32 bytes for AES-256 # if len(key) < 32: # key = key.ljust(32, b'\0') # elif len(key) > 32: # key = key[:32] # # Create JWE Header (equivalent to JWEHeader.Builder) # jwe_header = { # "alg": "dir", # Direct encryption (JWEAlgorithm.DIR) # "enc": "A256GCM", # AES-256-GCM (EncryptionMethod.A256GCM) # "kid": key_id, # keyID # "clientid": client_id # custom parameter # } # # Base64URL encode the header # header_json = json.dumps(jwe_header, separators=(',', ':')) # header_b64 = base64.urlsafe_b64encode(header_json.encode('utf-8')).decode('utf-8').rstrip('=') # # Generate a random 96-bit (12-byte) IV for GCM # iv = os.urandom(12) # # Encrypt the payload using AES-256-GCM # cipher = Cipher(algorithms.AES(key), modes.GCM(iv), backend=default_backend()) # encryptor = cipher.encryptor() # # Additional Authenticated Data (AAD) is the encoded header # aad = header_b64.encode('ascii') # encryptor.authenticate_additional_data(aad) # # Encrypt the payload # payload_bytes = response_string.encode('utf-8') # ciphertext = encryptor.update(payload_bytes) + encryptor.finalize() # # Get the authentication tag # auth_tag = encryptor.tag # # Base64URL encode components # encrypted_key = "" # Empty for direct encryption (dir algorithm) # iv_b64 = base64.urlsafe_b64encode(iv).decode('utf-8').rstrip('=') # ciphertext_b64 = base64.urlsafe_b64encode(ciphertext).decode('utf-8').rstrip('=') # tag_b64 = base64.urlsafe_b64encode(auth_tag).decode('utf-8').rstrip('=') # # Construct JWE: header.encrypted_key.iv.ciphertext.tag # jwe_encrypted_data = f"{header_b64}.{encrypted_key}.{iv_b64}.{ciphertext_b64}.{tag_b64}" # return jwe_encrypted_data # # Example usage with BillDesk payload # if __name__ == "__main__": # # BillDesk credentials (replace with actual values) # MERC_ID = "BDUAT2K676" # UAT Merchant ID # CLIENT_ID = client_id = "bduat2k676sj" # Provided by BillDesk # KEY_ID = key_id = "vstkEu52BWR9" # Provided by BillDesk # # ENCRYPTION_KEY = b"12345678901234567890123456789012" # Must be exactly 32 bytes # encryption_key = ENCRYPTION_KEY = "e9Y5khgyMyluQRrH4XjgQLgZ9oZdwwk2" # # Method 1: Manual payload creation # print("Method 1: Manual payload creation") # print("="*50) # payload = { # "orderid": f"TEST{int(time.time())}", # "mercid": MERC_ID, # "order_date": time.strftime("%Y-%m-%dT%H:%M:%S+05:30"), # "amount": "29999.28", # "currency": "356", # "ru": "https://www.merchant.com", # "itemcode": "DIRECT", # "device": { # "init_channel": "internet", # "ip": "123.0.0.1", # "user_agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:51.0) Gecko/20100101 Firefox/51.0", # "accept_header": "application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9" # } # } # payload_json = json.dumps(payload, separators=(',', ':')) # print("Original payload:") # print(payload_json) # print() # encrypted_data = encrypt_jwe_billdesk(payload_json, ENCRYPTION_KEY, KEY_ID, CLIENT_ID) # print("JWE Encrypted Data:") # print(encrypted_data) # print() # # Method 2: Using convenience function # print("Method 2: Using convenience function") # print("="*50) # encrypted_payment = encrypt_billdesk_payment( # merc_id=MERC_ID, # amount="29999.28", # return_url="https://www.merchant.com", # encryption_key=ENCRYPTION_KEY, # key_id=KEY_ID, # client_id=CLIENT_ID # ) # print("Encrypted Payment Data:") # print(encrypted_payment) # # # Example usage # # if __name__ == "__main__": # # # Example parameters (replace with actual values) # # response_string = "Your response data here" # MERC_ID = "BDUAT2K676" # UAT Merchant ID # CLIENT_ID = client_id = "bduat2k676sj" # Provided by BillDesk # KEY_ID = key_id = "vstkEu52BWR9" # Provided by BillDesk # ENCRYPTION_KEY = "e9Y5khgyMyluQRrH4XjgQLgZ9oZdwwk2" # # payload = { # # "orderid": f"TEST{int(time.time())}", # unique order id # # "mercid": MERC_ID, # # "order_date": time.strftime("%Y-%m-%dT%H:%M:%S+05:30"), # # "amount": "29999.28", # # "currency": "356", # INR # # "ru": "https://www.merchant.com", # # "itemcode": "DIRECT", # # "device": { # # "init_channel": "internet", # # "ip": "123.0.0.1", # # "user_agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:51.0) Gecko/20100101 Firefox/51.0", # # "accept_header": "application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9" # # } # # } # # encrypted_data = encrypt_jwe_billdesk(response_string, encryption_key, key_id, client_id) # # print("JWE Encrypted Data:") # # print(encrypted_data) import json import base64 import hmac import hashlib from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.backends import default_backend import os import time, requests, uuid def sign_jws_billdesk(payload_string, signing_key, key_id, client_id, algorithm="HS256"): """ Sign data using JWS with HMAC for BillDesk integration Args: payload_string (str): The data to sign (usually encrypted JWE) signing_key (str): BillDesk provided signing key key_id (str): BillDesk provided signing key ID client_id (str): BillDesk provided client ID algorithm (str): JWS algorithm (default: "HS256") Returns: str: JWS signed string """ # Create JWS Header jws_header = { "alg": algorithm, # HMAC SHA-256 "typ": "JWT", # Standard JWT type "kid": key_id, # Key ID "clientid": client_id # Custom parameter } # Base64URL encode header header_json = json.dumps(jws_header, separators=(',', ':')) header_b64 = base64.urlsafe_b64encode(header_json.encode('utf-8')).decode('utf-8').rstrip('=') # Base64URL encode payload payload_b64 = base64.urlsafe_b64encode(payload_string.encode('utf-8')).decode('utf-8').rstrip('=') # Create signing input: header.payload signing_input = f"{header_b64}.{payload_b64}" # Sign with HMAC-SHA256 signing_key_bytes = signing_key.encode('utf-8') if algorithm == "HS256": signature = hmac.new(signing_key_bytes, signing_input.encode('utf-8'), hashlib.sha256).digest() elif algorithm == "HS384": signature = hmac.new(signing_key_bytes, signing_input.encode('utf-8'), hashlib.sha384).digest() elif algorithm == "HS512": signature = hmac.new(signing_key_bytes, signing_input.encode('utf-8'), hashlib.sha512).digest() else: raise ValueError(f"Unsupported algorithm: {algorithm}") # Base64URL encode signature signature_b64 = base64.urlsafe_b64encode(signature).decode('utf-8').rstrip('=') # Construct JWS: header.payload.signature jws_signed_data = f"{header_b64}.{payload_b64}.{signature_b64}" return jws_signed_data def encrypt_and_sign_billdesk_payment(merc_id, amount, return_url, encryption_key, enc_key_id, enc_client_id, signing_key, sign_key_id, sign_client_id, currency="356", itemcode="DIRECT", user_ip="123.0.0.1"): """ Complete BillDesk payment flow: Create payload -> Encrypt with JWE -> Sign with JWS Args: merc_id (str): Merchant ID amount (str): Payment amount return_url (str): Return URL encryption_key (str): Encryption key enc_key_id (str): Encryption key ID enc_client_id (str): Encryption client ID signing_key (str): Signing key sign_key_id (str): Signing key ID sign_client_id (str): Signing client ID currency (str): Currency code (default: "356" for INR) itemcode (str): Item code (default: "DIRECT") user_ip (str): User's IP address Returns: tuple: (original_payload, encrypted_data, signed_data) """ # Step 1: Create payload payload = { "orderid": f"TEST{int(time.time())}", "mercid": merc_id, "order_date": time.strftime("%Y-%m-%dT%H:%M:%S+05:30"), "amount": amount, "currency": currency, "ru": return_url, "itemcode": itemcode, "device": { "init_channel": "internet", "ip": user_ip, "user_agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:51.0) Gecko/20100101 Firefox/51.0", "accept_header": "application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9" } } payload_json = json.dumps(payload, separators=(',', ':')) # Step 2: Encrypt with JWE encrypted_data = encrypt_jwe_billdesk(payload_json, encryption_key, enc_key_id, enc_client_id) # Step 3: Sign the encrypted data with JWS signed_data = sign_jws_billdesk(encrypted_data, signing_key, sign_key_id, sign_client_id) return payload_json, encrypted_data, signed_data def encrypt_billdesk_payment(merc_id, amount, return_url, encryption_key, key_id, client_id, currency="356", itemcode="DIRECT", user_ip="123.0.0.1"): """ Create and encrypt a BillDesk payment payload Args: merc_id (str): Merchant ID provided by BillDesk amount (str): Payment amount return_url (str): Return URL after payment encryption_key (str): BillDesk provided encryption key key_id (str): BillDesk provided encryption key ID client_id (str): BillDesk provided client ID currency (str): Currency code (default: "356" for INR) itemcode (str): Item code (default: "DIRECT") user_ip (str): User's IP address Returns: str: JWE encrypted payment data """ # Create the payment payload payload = { "orderid": f"TEST{int(time.time())}", # unique order id "mercid": merc_id, "order_date": time.strftime("%Y-%m-%dT%H:%M:%S+05:30"), "amount": amount, "currency": currency, # INR "ru": return_url, "itemcode": itemcode, "device": { "init_channel": "internet", "ip": user_ip, "user_agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:51.0) Gecko/20100101 Firefox/51.0", "accept_header": "application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9" } } # Convert to JSON string payload_json = json.dumps(payload, separators=(',', ':')) # Encrypt and return return encrypt_jwe_billdesk(payload_json, encryption_key, key_id, client_id) def encrypt_jwe_billdesk(response_string, encryption_key, key_id, client_id): """ Encrypt data using JWE with AES-256-GCM for BillDesk integration Args: response_string (str): The payload to encrypt encryption_key (str): BillDesk provided encryption key key_id (str): BillDesk provided encryption key ID client_id (str): BillDesk provided client ID Returns: str: JWE encrypted string """ # Convert encryption key to bytes (equivalent to encryptionKey.getBytes()) key = encryption_key.encode('utf-8') # Ensure key is 32 bytes for AES-256 if len(key) < 32: key = key.ljust(32, b'\0') elif len(key) > 32: key = key[:32] # Create JWE Header (equivalent to JWEHeader.Builder) jwe_header = { "alg": "dir", # Direct encryption (JWEAlgorithm.DIR) "enc": "A256GCM", # AES-256-GCM (EncryptionMethod.A256GCM) "kid": key_id, # keyID "clientid": client_id # custom parameter } # Base64URL encode the header header_json = json.dumps(jwe_header, separators=(',', ':')) header_b64 = base64.urlsafe_b64encode(header_json.encode('utf-8')).decode('utf-8').rstrip('=') # Generate a random 96-bit (12-byte) IV for GCM iv = os.urandom(12) # Encrypt the payload using AES-256-GCM cipher = Cipher(algorithms.AES(key), modes.GCM(iv), backend=default_backend()) encryptor = cipher.encryptor() # Additional Authenticated Data (AAD) is the encoded header aad = header_b64.encode('ascii') encryptor.authenticate_additional_data(aad) # Encrypt the payload payload_bytes = response_string.encode('utf-8') ciphertext = encryptor.update(payload_bytes) + encryptor.finalize() # Get the authentication tag auth_tag = encryptor.tag # Base64URL encode components encrypted_key = "" # Empty for direct encryption (dir algorithm) iv_b64 = base64.urlsafe_b64encode(iv).decode('utf-8').rstrip('=') ciphertext_b64 = base64.urlsafe_b64encode(ciphertext).decode('utf-8').rstrip('=') tag_b64 = base64.urlsafe_b64encode(auth_tag).decode('utf-8').rstrip('=') # Construct JWE: header.encrypted_key.iv.ciphertext.tag jwe_encrypted_data = f"{header_b64}.{encrypted_key}.{iv_b64}.{ciphertext_b64}.{tag_b64}" return jwe_encrypted_data def decrypt_jwe_billdesk(encrypted_data, encryption_key): """ Decrypt JWE data using AES-256-GCM for BillDesk integration Handles both direct JWE and nested JWS->JWE formats Args: encrypted_data (str): The JWE encrypted string or JWS containing JWE encryption_key (str): BillDesk provided encryption key Returns: str: Decrypted payload string """ try: # Split the input to check format parts = encrypted_data.split('.') # Check if this is a JWS (3 parts) containing a JWE payload if len(parts) == 3: # This is JWS format: header.payload.signature # The payload should be the actual JWE jws_header_b64, jws_payload_b64, jws_signature_b64 = parts # Decode the JWS payload (which should be the JWE) jws_payload_padded = jws_payload_b64 + '=' * ((4 - len(jws_payload_b64) % 4) % 4) jwe_encrypted_data = base64.urlsafe_b64decode(jws_payload_padded).decode('utf-8') #print(f"Extracted JWE from JWS payload: {jwe_encrypted_data}") elif len(parts) == 5: # This is direct JWE format jwe_encrypted_data = encrypted_data else: raise ValueError(f"Invalid format - got {len(parts)} parts, expected 3 (JWS) or 5 (JWE)") # Now process the JWE jwe_parts = jwe_encrypted_data.split('.') if len(jwe_parts) != 5: raise ValueError("Invalid JWE format - must have 5 parts separated by dots") header_b64, encrypted_key_b64, iv_b64, ciphertext_b64, tag_b64 = jwe_parts # Decode the header to verify algorithm and get metadata # Add padding if needed for base64 decoding header_b64_padded = header_b64 + '=' * ((4 - len(header_b64) % 4) % 4) header_json = base64.urlsafe_b64decode(header_b64_padded).decode('utf-8') header = json.loads(header_json) # Verify this is the expected encryption method if header.get('alg') != 'dir': raise ValueError(f"Unsupported algorithm: {header.get('alg')}") if header.get('enc') != 'A256GCM': raise ValueError(f"Unsupported encryption method: {header.get('enc')}") # Prepare the encryption key (same as in encryption) key = encryption_key.encode('utf-8') if len(key) < 32: key = key.ljust(32, b'\0') elif len(key) > 32: key = key[:32] # Decode the components (add padding if needed) def base64url_decode(data): padded = data + '=' * ((4 - len(data) % 4) % 4) return base64.urlsafe_b64decode(padded) # For 'dir' algorithm, encrypted_key should be empty if encrypted_key_b64: raise ValueError("Expected empty encrypted key for 'dir' algorithm") iv = base64url_decode(iv_b64) ciphertext = base64url_decode(ciphertext_b64) auth_tag = base64url_decode(tag_b64) # Verify IV length for GCM (should be 12 bytes) if len(iv) != 12: raise ValueError(f"Invalid IV length: {len(iv)} (expected 12)") # Decrypt using AES-256-GCM cipher = Cipher(algorithms.AES(key), modes.GCM(iv, auth_tag), backend=default_backend()) decryptor = cipher.decryptor() # Additional Authenticated Data (AAD) is the encoded header aad = header_b64.encode('ascii') decryptor.authenticate_additional_data(aad) # Decrypt the payload decrypted_bytes = decryptor.update(ciphertext) + decryptor.finalize() # Convert back to string decrypted_string = decrypted_bytes.decode('utf-8') return decrypted_string except Exception as e: raise ValueError(f"Decryption failed: {str(e)}") # Helper function to analyze the encrypted data structure def analyze_encrypted_data(encrypted_data): """ Analyze the structure of encrypted data from BillDesk """ parts = encrypted_data.split('.') # print(f"Number of parts: {len(parts)}") if len(parts) == 3: # print("Format: JWS (JSON Web Signature)") jws_header_b64, jws_payload_b64, jws_signature_b64 = parts # Decode JWS header try: jws_header_padded = jws_header_b64 + '=' * (4 - len(jws_header_b64) % 4) % 4 jws_header = json.loads(base64.urlsafe_b64decode(jws_header_padded).decode('utf-8')) print(f"JWS Header: {json.dumps(jws_header, indent=2)}") except: print("Could not decode JWS header") # Try to decode JWS payload try: jws_payload_padded = jws_payload_b64 + '=' * (4 - len(jws_payload_b64) % 4) % 4 jws_payload = base64.urlsafe_b64decode(jws_payload_padded).decode('utf-8') print(f"JWS Payload (JWE): {jws_payload}") # Check if payload is JWE jwe_parts = jws_payload.split('.') print(f"JWE parts in payload: {len(jwe_parts)}") if len(jwe_parts) == 5: # Decode JWE header try: jwe_header_padded = jwe_parts[0] + '=' * (4 - len(jwe_parts[0]) % 4) % 4 jwe_header = json.loads(base64.urlsafe_b64decode(jwe_header_padded).decode('utf-8')) print(f"JWE Header: {json.dumps(jwe_header, indent=2)}") except: print("Could not decode JWE header") except Exception as e: print(f"Could not decode JWS payload: {e}") elif len(parts) == 5: print("Format: JWE (JSON Web Encryption)") # Decode JWE header try: header_padded = parts[0] + '=' * (4 - len(parts[0]) % 4) % 4 header = json.loads(base64.urlsafe_b64decode(header_padded).decode('utf-8')) print(f"JWE Header: {json.dumps(header, indent=2)}") except: print("Could not decode JWE header") print("-" * 50) # Example usage with BillDesk payload if __name__ == "__main__": MERC_ID = "BDUAT2K676" # UAT Merchant ID CLIENT_ID = client_id = "bduat2k676sj" # Provided by BillDesk KEY_ID = key_id = "vstkEu52BWR9" # Provided by BillDesk ENCRYPTION_KEY = "e9Y5khgyMyluQRrH4XjgQLgZ9oZdwwk2" SIGNING_KEY = "5ZvlXPsYeuk6VTJRlB5HyBOMpb4wPYQJ" payload = { "orderid": f"TEST{int(time.time())}", "mercid": MERC_ID, "order_date": time.strftime("%Y-%m-%dT%H:%M:%S+05:30"), "amount": "29999.28", "currency": "356", "ru": "https://www.merchant.com", "itemcode": "DIRECT", "device": { "init_channel": "internet", "ip": "123.0.0.1", "user_agent": "Mozilla/5.0 (Windows NT 10.0; WOW64; rv:51.0) Gecko/20100101 Firefox/51.0", "accept_header": "application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9" } } payload_json = json.dumps(payload, separators=(',', ':')) encrypted_data = encrypt_jwe_billdesk(payload_json, ENCRYPTION_KEY, KEY_ID, CLIENT_ID) encrypted_payment = encrypt_billdesk_payment( merc_id=MERC_ID, amount="29999.28", return_url="https://www.merchant.com", encryption_key=ENCRYPTION_KEY, key_id=KEY_ID, client_id=CLIENT_ID ) signed_encrypted_data = sign_jws_billdesk(encrypted_payment, SIGNING_KEY, KEY_ID, CLIENT_ID) headers = { "Content-Type": "application/jose", "Accept": "application/jose", "BD-Traceid": str(uuid.uuid4()).replace("-", "")[:20], # Unique ≤35 chars "BD-Timestamp": str(int(time.time())) # Epoch in seconds } # ------------------------------- # Step 4: Call BillDesk API # ------------------------------- url = "https://uat1.billdesk.com/u2/payments/ve1_2/orders/create" response = requests.post(url, headers=headers, data=signed_encrypted_data) print("Response text:", response.text[:1000]) decrypted_data = decrypt_jwe_billdesk(response.text, ENCRYPTION_KEY) print(decrypted_data)