## File: your_app/services/payment_service.py import logging import pprint import traceback import re from authorizenet import apicontractsv1 from authorizenet.apicontrollers import ( createTransactionController, createCustomerProfileController, createCustomerPaymentProfileController ) from authorizenet.constants import constants from .. import schemas from config import load_config # Assuming you have this logger = logging.getLogger(__name__) # Load Authorize.net credentials ApplicationConfig = load_config() # API_LOGIN_ID = ApplicationConfig.API_LOGIN_ID # TRANSACTION_KEY = ApplicationConfig.TRANSACTION_KEY # Authorize.net credentials (Sandbox Test Credentials) API_LOGIN_ID = '9U6w96gZmX' TRANSACTION_KEY = '94s6Qy458mMNJr7G' # --- MODIFICATION: Set the environment globally --- # Set this to SANDBOX for testing, PRODUCTION for live constants.show_url_on_request = True # Very useful for debugging constants.environment = constants.SANDBOX def _get_authnet_error_message(response): """ Robust error parsing function that correctly handles the API's response format. """ if response is None: return "No response from payment gateway." try: if hasattr(response, 'messages') and response.messages is not None: if hasattr(response, 'transactionResponse') and response.transactionResponse and hasattr(response.transactionResponse, 'errors') and response.transactionResponse.errors: error = response.transactionResponse.errors[0] return f"Error {error.errorCode}: {error.errorText}" if hasattr(response.messages, 'message'): message_list = response.messages.message if not isinstance(message_list, list): message_list = [message_list] if message_list: msg = message_list[0] code = msg.code if hasattr(msg, 'code') else 'Unknown' text = msg.text if hasattr(msg, 'text') else 'No details provided.' return f"Error {code}: {text}" except Exception as e: logger.error(f"Error while parsing Auth.Net error message: {e}") return "An unparsable error occurred with the payment gateway." return "An unknown error occurred with the payment gateway." def create_customer_profile(customer: schemas.Customer, card_info: schemas.CardCreate): """ Creates a new customer profile in Authorize.Net with their first payment method. This version sanitizes and trims all customer data before sending. """ logger.info(f"Attempting to create Auth.Net profile for customer ID: {customer.id}") merchantAuth = apicontractsv1.merchantAuthenticationType(name=API_LOGIN_ID, transactionKey=TRANSACTION_KEY) # --- DATA SANITIZATION LOGIC --- def sanitize(text, max_len, allow_spaces=False, is_zip=False): if not text: return "" if is_zip: pattern = r'[^a-zA-Z0-9-]' # Allow hyphens for ZIP+4 else: pattern = r'[^a-zA-Z0-9]' if not allow_spaces else r'[^a-zA-Z0-9\s]' sanitized = re.sub(pattern, '', str(text)) return sanitized.strip()[:max_len] # API max lengths: name=50, address=60, city=40, state=40, zip=20, email=255 first_name = sanitize(customer.customer_first_name, 50) or "N/A" last_name = sanitize(customer.customer_last_name, 50) or "N/A" address = sanitize(customer.customer_address, 60, allow_spaces=True) or "N/A" city = sanitize(customer.customer_town, 40) or "N/A" # ========= CHANGE 1.A: ADD STATE HERE ========= state = sanitize(customer.customer_state, 40) or "MA" # Defaulting to MA for safety zip_code = sanitize(customer.customer_zip, 20, is_zip=True) email = (customer.customer_email or f"no-email-{customer.id}@example.com")[:255] creditCard = apicontractsv1.creditCardType( cardNumber=card_info.card_number, expirationDate=card_info.expiration_date, cardCode=card_info.cvv ) billTo = apicontractsv1.customerAddressType( firstName=first_name, lastName=last_name, address=address, city=city, state='MA', # And include it in the object zip=zip_code, country="USA" ) paymentProfile = apicontractsv1.customerPaymentProfileType( billTo=billTo, payment=apicontractsv1.paymentType(creditCard=creditCard) ) customerProfile = apicontractsv1.customerProfileType( merchantCustomerId=str(customer.id), email=email, paymentProfiles=[paymentProfile] ) request = apicontractsv1.createCustomerProfileRequest( merchantAuthentication=merchantAuth, profile=customerProfile, # ========= CHANGE 2.A: USE liveMode ========= validationMode="testMode" ) controller = createCustomerProfileController(request) # ... rest of the function is the same ... try: controller.execute() response = controller.getresponse() if response is not None and response.messages.resultCode == "Ok": profile_id = response.customerProfileId payment_id = response.customerPaymentProfileIdList[0] if response.customerPaymentProfileIdList else None return str(profile_id), str(payment_id) if payment_id else None else: error_msg = _get_authnet_error_message(response) logger.error(f"Failed to create customer profile (API Error): {error_msg}") logger.error(f"SANITIZED DATA SENT: FirstName='{first_name}', LastName='{last_name}', Address='{address}', City='{city}', State='{state}', Zip='{zip_code}', Email='{email}'") raise ValueError(error_msg) except Exception as e: logger.error(f"A critical exception occurred during the API call: {traceback.format_exc()}") raise ValueError("Could not connect to the payment gateway.") def add_payment_profile_to_customer(customer_profile_id: str, customer: schemas.Customer, card_info: schemas.CardCreate): logger.info(f"Adding new payment profile to Auth.Net customer profile ID: {customer_profile_id}") merchantAuth = apicontractsv1.merchantAuthenticationType(name=API_LOGIN_ID, transactionKey=TRANSACTION_KEY) def sanitize(text, max_len, allow_spaces=False, is_zip=False): if not text: return "" if is_zip: pattern = r'[^a-zA-Z0-9-]' else: pattern = r'[^a-zA-Z0-9]' if not allow_spaces else r'[^a-zA-Z0-9\s]' sanitized = re.sub(pattern, '', str(text)) return sanitized.strip()[:max_len] first_name = sanitize(customer.customer_first_name, 50) or "N/A" last_name = sanitize(customer.customer_last_name, 50) or "N/A" address = sanitize(customer.customer_address, 60, allow_spaces=True) or "N/A" city = sanitize(customer.customer_town, 40) or "N/A" # ========= CHANGE 1.B: ADD STATE HERE ========= state = sanitize(customer.customer_state, 40) or "MA" # Defaulting to MA for safety zip_code = sanitize(customer.customer_zip, 20, is_zip=True) creditCard = apicontractsv1.creditCardType( cardNumber=card_info.card_number, expirationDate=card_info.expiration_date, cardCode=card_info.cvv ) billTo = apicontractsv1.customerAddressType( firstName=first_name, lastName=last_name, address=address, city=city, state=state, # And include it in the object zip=zip_code, country="USA" ) paymentProfile = apicontractsv1.customerPaymentProfileType( billTo=billTo, payment=apicontractsv1.paymentType(creditCard=creditCard) ) request = apicontractsv1.createCustomerPaymentProfileRequest( merchantAuthentication=merchantAuth, customerProfileId=customer_profile_id, paymentProfile=paymentProfile, # ========= CHANGE 2.B: USE liveMode ========= validationMode="testMode" ) controller = createCustomerPaymentProfileController(request) try: controller.execute() response = controller.getresponse() if response is not None and response.messages.resultCode == "Ok": return str(response.customerPaymentProfileId) else: error_msg = _get_authnet_error_message(response) logger.error(f"Failed to add payment profile: {error_msg}") logger.error(f"SANITIZED DATA SENT FOR ADD PROFILE: FirstName='{first_name}', LastName='{last_name}', Address='{address}', City='{city}', State='{state}', Zip='{zip_code}'") logger.error(f"Card info: number='{card_info.card_number}', exp='{card_info.expiration_date}', cvv='{card_info.cvv}'") # Mask if sensitive logger.error(pprint.pformat(vars(billTo))) logger.error(pprint.pformat(vars(request))) raise ValueError(error_msg) except Exception as e: logger.error(f"A critical exception occurred during the API call: {traceback.format_exc()}") raise ValueError("Could not connect to the payment gateway.") def authorize_customer_profile(customer_profile_id: str, payment_profile_id: str, transaction_req: schemas.TransactionAuthorizeByCardID): """ Creates an AUTH_ONLY transaction against a customer profile. This holds the funds but does not capture them. """ logger.info(f"Authorizing profile {customer_profile_id} / payment {payment_profile_id} for ${transaction_req.preauthorize_amount}") merchantAuth = apicontractsv1.merchantAuthenticationType(name=API_LOGIN_ID, transactionKey=TRANSACTION_KEY) profile_to_authorize = apicontractsv1.profileTransAuthOnlyType( customerProfileId=customer_profile_id, paymentProfileId=payment_profile_id ) transactionRequest = apicontractsv1.transactionRequestType( transactionType="authOnlyTransaction", amount=f"{transaction_req.preauthorize_amount:.2f}", profile=profile_to_authorize ) if transaction_req.tax_amount and transaction_req.tax_amount > 0: transactionRequest.tax = apicontractsv1.extendedAmountType(amount=f"{transaction_req.tax_amount:.2f}", name="Sales Tax") createtransactionrequest = apicontractsv1.createTransactionRequest( merchantAuthentication=merchantAuth, transactionRequest=transactionRequest ) controller = createTransactionController(createtransactionrequest) controller.execute() # The response is returned directly to the router to be parsed there return controller.getresponse() def capture_authorized_transaction(transaction_req: schemas.TransactionCapture): """Captures a previously authorized transaction.""" logger.info(f"Capturing transaction {transaction_req.auth_net_transaction_id} for {transaction_req.charge_amount}") merchantAuth = apicontractsv1.merchantAuthenticationType(name=API_LOGIN_ID, transactionKey=TRANSACTION_KEY) transactionRequest = apicontractsv1.transactionRequestType( transactionType="priorAuthCaptureTransaction", amount=f"{transaction_req.charge_amount:.2f}", refTransId=transaction_req.auth_net_transaction_id ) createtransactionrequest = apicontractsv1.createTransactionRequest( merchantAuthentication=merchantAuth, transactionRequest=transactionRequest ) controller = createTransactionController(createtransactionrequest) controller.execute() return controller.getresponse()