""" TSYS TransIT + TSEP payment provider implementation. TransIT is TSYS's hosted payment gateway. TSEP (Tokenisation via Secure Encrypt-and-Pass) is the client-side hosted-fields solution that collects card data inside TSYS-controlled iframes, then fires a JavaScript callback with a one-time token. Raw card data never reaches HubWallet servers — PCI compliance is delegated to TSYS. """ import hashlib import hmac as hmac_module import logging from datetime import datetime, timezone from typing import Any, Dict, Optional import httpx from src.core.providers.base import ( BasePaymentProvider, ChargeResult, IframeConfig, ProviderConfig, TokenDetails, ) from src.core.providers.registry import ProviderRegistry logger = logging.getLogger(__name__) # Response approval codes indicating an authorised / successful transaction. APPROVAL_CODES = {"A0000", "00", "08", "11", "T0", "85"} @ProviderRegistry.register class TSYSProvider(BasePaymentProvider): """ TSYS TransIT + TSEP payment provider. Credentials required (stored encrypted in MerchantProviderCredential): merchant_id — TSYS merchant identifier (12-20 chars) device_id — TSEP / TransIT device identifier (up to 24 chars) transaction_key — generated once via GenerateKey; never accept from merchant input """ slug = "tsys" # class attribute — avoids the @property overhead used by Payrix # ------------------------------------------------------------------ # BasePaymentProvider interface # ------------------------------------------------------------------ def get_iframe_config( self, config: ProviderConfig, amount: float = 0.0, currency: str = "USD", ) -> IframeConfig: """ Generate a TSEP manifest and return an IframeConfig the frontend can use to bootstrap the TSEP hosted-fields SDK. The manifest is an AES-128-CBC ciphertext of: merchantID (20 chars, space-padded right) + deviceID (24 chars, space-padded right) + "000000000000" (12 zeros — always zero for TSEP amount) + date (MMDDYYYY) (8 chars) total: 64 chars Key = IV = first 16 bytes of txnKey, UTF-8 encoded. The cipher output is hex-encoded (NOT base64). An HMAC-MD5 is also computed (key=txnKey, msg=txnKey) and its first 4 and last 4 hex chars wrap the cipher hex. """ creds = config.credentials merchant_id = creds.get("merchant_id", "") device_id = creds.get("device_id", "") txn_key = self._resolve_txn_key(creds) manifest = self._generate_manifest(merchant_id, device_id, txn_key) # TSEP JS URL: {base}/transit-tsep-web/jsView/{deviceID}?{manifest} tsep_base = (creds.get("tsep_base_url") or "").rstrip("/") sdk_url = f"{tsep_base}/transit-tsep-web/jsView/{device_id}?{manifest}" return IframeConfig( sdk_url=sdk_url, api_key="", merchant_id=merchant_id, mode=creds.get("tsys_env") or "test", extra={ "device_id": device_id, "manifest": manifest, "tsep_base_url": tsep_base, }, ) async def get_token_details( self, config: ProviderConfig, token: str ) -> TokenDetails: """ TSEP tokens are one-time-use. Card metadata (brand, last4) is delivered client-side via the TokenEvent / AuthenticationCompleteEvent callbacks. There is no server-side token lookup endpoint in TSEP v2. """ raise NotImplementedError( "TSYS/TSEP tokens are opaque one-time-use values. " "Card metadata is delivered via the TSEP JavaScript callback; " "use create_payment_method() with customer_context instead." ) async def create_payment_method( self, config: ProviderConfig, token: str, customer_context: Dict[str, Any], ) -> TokenDetails: """ Build a TokenDetails from the client-supplied customer_context dict. The dict must contain the fields forwarded from the TSEP TokenEvent: card_type — e.g. "VISA" masked_card — e.g. "************1234" """ card_type = customer_context.get("card_type") or "" masked_card = customer_context.get("masked_card") or "" last4 = masked_card[-4:] if len(masked_card) >= 4 else masked_card return TokenDetails( provider_payment_method_id=token, provider=self.slug, brand=card_type, last4=last4, reference_token=token, ) async def submit_charge( self, config: ProviderConfig, amount: float, currency: str, payment_method_token: str, payment_method_type: str, capture: bool, idempotency_key: Optional[str], metadata: Optional[Dict[str, Any]], ) -> ChargeResult: """ Submit a TransIT Sale (capture=True) or PreAuth (capture=False) transaction. XSD field order for Sale/PreAuth is strictly enforced: deviceID → transactionKey → cardDataSource → transactionAmount → currencyCode → cardNumber → developerID """ creds = config.credentials device_id = creds.get("device_id", "") developer_id = creds.get("developer_id") or "" api_base_url = creds.get("api_base_url") or "" txn_key = self._resolve_txn_key(creds) # TransIT expects integer cents as a plain string amount_cents = str(int(round(amount * 100))) # Use PreAuth when capture=False (authorization only), Sale when capture=True txn_type = "Sale" if capture else "PreAuth" # XSD-enforced field order payload: Dict[str, Any] = { "deviceID": device_id, "transactionKey": txn_key, "cardDataSource": "INTERNET", "transactionAmount": amount_cents, "currencyCode": "USD", "cardNumber": payment_method_token, "developerID": developer_id, } result = self._transit_post({txn_type: payload}, base_url=api_base_url) resp_key = f"{txn_type}Response" charge_resp = result.get(resp_key, result) approval_code = charge_resp.get("approvalCode") or charge_resp.get("responseCode") or "" # Prefer TSYS's own transactionID — this is the value needed for subsequent # Return (refund) and Void requests via the cardTransactionID field. # hostReferenceNumber is the bank's reference and cannot be used for that. txn_id = ( charge_resp.get("transactionID") or charge_resp.get("hostReferenceNumber") or "" ) status = "succeeded" if approval_code.upper() in {c.upper() for c in APPROVAL_CODES} else "failed" return ChargeResult( transaction_id=txn_id, status=status, amount=amount, currency=currency, raw_response=result, ) async def submit_ach_charge( self, config: ProviderConfig, routing_number: str, account_number: str, account_type: str, first_name: str, last_name: str, amount: float, ) -> ChargeResult: """ Submit a TransIT ACH (Ach) transaction. This is a TSYS-specific extra method not defined in BasePaymentProvider. XSD field order: deviceID → transactionKey → transactionAmount → accountDetails → achSecCode → originateDate → firstName → lastName → developerID """ creds = config.credentials device_id = creds.get("device_id", "") developer_id = creds.get("developer_id") or "" api_base_url = creds.get("api_base_url") or "" txn_key = self._resolve_txn_key(creds) amount_cents = str(int(round(amount * 100))) originate_date = datetime.now(timezone.utc).strftime("%m%d%Y") # XSD-enforced field order payload: Dict[str, Any] = { "deviceID": device_id, "transactionKey": txn_key, "transactionAmount": amount_cents, "accountDetails": { "routingNumber": routing_number, "accountNumber": account_number, "accountType": account_type.upper(), }, "achSecCode": "WEB", "originateDate": originate_date, "firstName": first_name, "lastName": last_name, "developerID": developer_id, } result = self._transit_post({"Ach": payload}, base_url=api_base_url) ach_resp = result.get("AchResponse", result) approval_code = ach_resp.get("approvalCode") or ach_resp.get("responseCode") or "" txn_id = ( ach_resp.get("hostReferenceNumber") or ach_resp.get("transactionID") or "" ) status = "succeeded" if approval_code.upper() in {c.upper() for c in APPROVAL_CODES} else "failed" return ChargeResult( transaction_id=txn_id, status=status, amount=amount, currency="USD", raw_response=result, ) async def refund_charge( self, config: ProviderConfig, provider_transaction_id: str, amount: float, reason: Optional[str], ) -> ChargeResult: """ Submit a TransIT Return (refund) transaction. XSD field order mirrors Sale exactly: deviceID → transactionKey → cardDataSource → transactionAmount → currencyCode → cardTransactionID → developerID cardTransactionID replaces cardNumber in the card-data slot (after currencyCode). Its value must be the TSYS transactionID from the original SaleResponse (NOT hostReferenceNumber, which is the bank's reference number). """ creds = config.credentials device_id = creds.get("device_id", "") developer_id = creds.get("developer_id") or "" api_base_url = creds.get("api_base_url") or "" txn_key = self._resolve_txn_key(creds) amount_cents = str(int(round(amount * 100))) # XSD-enforced field order mirrors the Sale payload exactly: # deviceID → transactionKey → cardDataSource → transactionAmount # → currencyCode → [card data field] → developerID # cardTransactionID replaces cardNumber in the card-data slot (after currencyCode). payload: Dict[str, Any] = { "deviceID": device_id, "transactionKey": txn_key, "cardDataSource": "INTERNET", "transactionAmount": amount_cents, "currencyCode": "USD", "cardTransactionID": provider_transaction_id, "developerID": developer_id, } result = self._transit_post({"Return": payload}, base_url=api_base_url) return_resp = result.get("ReturnResponse", result) approval_code = return_resp.get("approvalCode") or return_resp.get("responseCode") or "" txn_id = ( return_resp.get("hostReferenceNumber") or return_resp.get("transactionID") or provider_transaction_id ) status = "refunded" if approval_code.upper() in {c.upper() for c in APPROVAL_CODES} else "failed" return ChargeResult( transaction_id=txn_id, status=status, amount=amount, currency="USD", raw_response=result, ) def verify_webhook( self, config: ProviderConfig, headers: Dict[str, str], raw_body: bytes, ) -> bool: """ Validate the X-GP-Signature header. Signature = SHA-512( raw_body.decode("utf-8") + webhook_secret ) Comparison is case-insensitive (hex strings). Returns False if no webhook_secret is configured. """ webhook_secret = config.credentials.get("webhook_secret") or "" if not webhook_secret: logger.warning( "op=tsys_verify_webhook result=no_secret — rejecting webhook" ) return False # Look up the header case-insensitively sig_header = ( headers.get("X-GP-Signature") or headers.get("x-gp-signature") or "" ) if not sig_header: return False try: body_str = raw_body.decode("utf-8") except Exception: return False digest = hashlib.sha512((body_str + webhook_secret).encode("utf-8")).hexdigest() return hmac_module.compare_digest(sig_header.lower(), digest.lower()) def parse_webhook_event(self, payload: Dict[str, Any]) -> Dict[str, Any]: """ Normalise a TSYS GP webhook payload to the HubWallet standard event shape: { event_id, transaction_id, status, amount, currency, provider_slug, raw } """ event_id = payload.get("eventId") or payload.get("event_id") or "" txn_id = ( payload.get("transactionId") or payload.get("transaction_id") or payload.get("hostReferenceNumber") or "" ) # TSYS GP reports status strings like "APPROVED", "DECLINED", "VOIDED" raw_status = (payload.get("status") or payload.get("transactionStatus") or "").upper() if raw_status in {"APPROVED", "CAPTURED", "SETTLED"}: status = "succeeded" elif raw_status in {"DECLINED", "FAILED", "ERROR"}: status = "failed" elif raw_status in {"VOIDED", "VOID"}: status = "failed" else: status = "processing" # Amounts may be in cents (integer) or dollars (float string) raw_amount = payload.get("transactionAmount") or payload.get("amount") or 0 try: amount_val = float(raw_amount) # If the value looks like integer cents (≥ 100 and no decimal), convert if isinstance(raw_amount, int) and raw_amount >= 100: amount_val = amount_val / 100 except (TypeError, ValueError): amount_val = 0.0 currency = payload.get("currencyCode") or payload.get("currency") or "USD" return { "event_id": event_id, "transaction_id": txn_id, "status": status, "amount": amount_val, "currency": currency, "provider_slug": self.slug, "raw": payload, } # ------------------------------------------------------------------ # TSYS-specific helpers # ------------------------------------------------------------------ def _resolve_txn_key(self, creds: Dict[str, Any]) -> str: """ Return a valid TransIT transaction key from credentials. Supports two credential storage patterns: - Onboarding v2: stored transaction_key is used directly - Legacy / user_id+password: generate a fresh key via GenerateKey API """ stored_key = creds.get("transaction_key", "") if stored_key: return stored_key merchant_id = creds.get("merchant_id", "") user_id = creds.get("user_id", "") password = creds.get("password", "") developer_id = creds.get("developer_id", "") api_base_url = creds.get("api_base_url", "") return self._generate_transaction_key(merchant_id, user_id, password, developer_id, api_base_url) def _generate_manifest( self, merchant_id: str, device_id: str, txn_key: str ) -> str: """ Build a TSEP manifest string. Plaintext (always exactly 64 bytes = 4 × AES-128 blocks): merchantID — right-padded to 20 chars with spaces deviceID — right-padded to 24 chars with spaces "000000000000" — 12 zero chars (amount always 0 for TSEP) date — MMDDYYYY (8 chars) Cipher: AES-128-CBC, key = IV = txnKey[:16].encode("utf-8") NO extra padding — 64 bytes is already an exact multiple of 16. Adding PKCS7 padding here would produce a 168-char manifest (80-byte ciphertext = 160 hex chars) instead of the required 136-char manifest (64-byte ciphertext = 128 hex chars), and TSYS would reject the malformed query string. Output: hex-encoded ciphertext (NOT base64) HMAC: HMAC-MD5(key=txnKey, msg=txnKey).hexdigest() Final manifest = hmac[:4] + hexCipher + hmac[-4:] """ from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.backends import default_backend key_bytes = txn_key[:16].encode("utf-8") date_str = datetime.now(timezone.utc).strftime("%m%d%Y") # MMDDYYYY plaintext = ( merchant_id.ljust(20)[:20] + device_id.ljust(24)[:24] + "000000000000" + date_str ) # 64 bytes = 4 × 16-byte AES blocks — no padding required. plaintext_bytes = plaintext.encode("utf-8") cipher = Cipher( algorithms.AES(key_bytes), modes.CBC(key_bytes), backend=default_backend(), ) encryptor = cipher.encryptor() encrypted_bytes = encryptor.update(plaintext_bytes) + encryptor.finalize() hex_cipher = encrypted_bytes.hex() mac = hmac_module.new( txn_key.encode("utf-8"), msg=txn_key.encode("utf-8"), digestmod="md5", ).hexdigest() return mac[:4] + hex_cipher + mac[-4:] def _generate_transaction_key( self, merchant_id: str, user_id: str, password: str, developer_id: str = "", api_base_url: str = "", ) -> str: """ Call the TSYS GenerateKey endpoint to exchange a merchant's user_id + password for a long-lived transaction key. This is a SYNCHRONOUS operation (called once during onboarding). Credentials (user_id, password) are NEVER logged. Field order is XSD-enforced: mid → userID → password → developerID Returns the transaction key string. Raises ValueError on bad credentials, RuntimeError on network/API error. """ base_url = api_base_url if not base_url: raise RuntimeError("TSYS api_base_url credential is not configured") # XSD-enforced field order payload = { "GenerateKey": { "mid": merchant_id, "userID": user_id, "password": password, "developerID": developer_id, } } try: resp = httpx.post( base_url, json=payload, timeout=30.0, headers={"Content-Type": "application/json"}, ) resp.raise_for_status() data = resp.json() except httpx.HTTPStatusError as exc: raise RuntimeError( f"TSYS GenerateKey HTTP {exc.response.status_code} error" ) from exc except Exception as exc: raise RuntimeError(f"TSYS GenerateKey network error: {exc}") from exc # TSYS may return the response under "GenerateKeyResponse" or # "GenerateKeyRequestResponse" depending on gateway version. gen_resp = ( data.get("GenerateKeyResponse") or data.get("GenerateKeyRequestResponse") or data ) response_status = gen_resp.get("status") or gen_resp.get("responseCode") or "" if response_status.upper() != "PASS": response_msg = gen_resp.get("responseMessage") or "" # Do NOT include user_id or password in the error message raise ValueError( f"TSYS GenerateKey failed (status={response_status!r}" + (f", message={response_msg!r}" if response_msg else "") + "). Check your merchant ID and credentials." ) txn_key = gen_resp.get("transactionKey") or "" if not txn_key: raise ValueError( "TSYS GenerateKey succeeded but returned an empty transaction key." ) return txn_key def _transit_post(self, payload: Dict[str, Any], base_url: str = "") -> Dict[str, Any]: """ Synchronous HTTP POST to the TSYS TransIT API. Raises RuntimeError on network/HTTP errors. """ if not base_url: raise RuntimeError("TSYS api_base_url credential is not configured") try: resp = httpx.post( base_url, json=payload, timeout=30.0, headers={"Content-Type": "application/json"}, ) resp.raise_for_status() return resp.json() except httpx.HTTPStatusError as exc: raise RuntimeError( f"TSYS TransIT HTTP {exc.response.status_code} error" ) from exc except Exception as exc: raise RuntimeError(f"TSYS TransIT network error: {exc}") from exc