documentverification/verifier/services/cashfree_client.py
vickytechkey 825392c845
All checks were successful
Deploy Beta (NATIVE) / deploy (push) Successful in 34s
Deploy Main Production (NATIVE) / deploy (push) Successful in 37s
Hardcode Cashfree PG API credentials in settings and client
2026-09-28 21:23:57 +05:30

526 lines
22 KiB
Python

import os
import json
import logging
import requests
from django.conf import settings
logger = logging.getLogger(__name__)
class CashfreeSecureIDClient:
"""
Service client for Cashfree Secure ID (Verification Suite / VRS) APIs.
Supports both Sandbox (TEST) and Production (PROD) environments.
Includes built-in test-data fallback for immediate interactive testing.
"""
SANDBOX_BASE_URL = "https://sandbox.cashfree.com/verification"
PRODUCTION_BASE_URL = "https://api.cashfree.com/verification"
DEFAULT_CLIENT_ID = "1370798f950e30514fa051e93868970731"
DEFAULT_CLIENT_SECRET = "cfsk_ma_prod_d441689923a71877068fee6b6e2e0d05_a626ee00"
DEFAULT_ENV = "PROD"
def __init__(self, client_id=None, client_secret=None, environment=None):
self.client_id = client_id or getattr(settings, 'CASHFREE_CLIENT_ID', '') or os.environ.get('CASHFREE_CLIENT_ID', '') or self.DEFAULT_CLIENT_ID
self.client_secret = client_secret or getattr(settings, 'CASHFREE_CLIENT_SECRET', '') or os.environ.get('CASHFREE_CLIENT_SECRET', '') or self.DEFAULT_CLIENT_SECRET
self.environment = (environment or getattr(settings, 'CASHFREE_ENV', '') or os.environ.get('CASHFREE_ENV', '') or self.DEFAULT_ENV).upper()
if self.environment == "PROD":
self.base_url = self.PRODUCTION_BASE_URL
else:
self.base_url = self.SANDBOX_BASE_URL
def _get_headers(self):
return {
"x-client-id": self.client_id,
"x-client-secret": self.client_secret,
"x-api-version": "2024-01-01",
"Content-Type": "application/json",
"Accept": "application/json",
}
def has_credentials(self):
return bool(self.client_id and self.client_secret)
def verify_credentials(self):
"""
Verify if provided Cashfree Verification Suite credentials are valid.
"""
if not self.has_credentials():
return {
"success": False,
"status_code": 400,
"message": "Cashfree Client ID or Secret Key missing.",
"data": None
}
url = f"{self.base_url}/ifsc"
payload = {"ifsc": "YESB0000262"}
try:
res = requests.post(url, headers=self._get_headers(), json=payload, timeout=10)
res_data = res.json() if res.headers.get("content-type", "").startswith("application/json") else res.text
is_valid = res.status_code == 200 or (isinstance(res_data, dict) and res_data.get("status") == "SUCCESS")
return {
"success": is_valid,
"status_code": res.status_code,
"data": res_data,
"curl": f"curl -X POST '{url}' \\\n -H 'x-client-id: {self.client_id}' \\\n -H 'x-client-secret: [REDACTED]' \\\n -H 'x-api-version: 2024-01-01' \\\n -H 'Content-Type: application/json' \\\n -d '{json.dumps(payload)}'"
}
except Exception as e:
return {
"success": False,
"status_code": 500,
"message": str(e),
"data": None
}
# ==================== PAN VERIFICATION ====================
def verify_pan(self, pan, name="", pan_type="INDIVIDUAL"):
pan = pan.strip().upper()
payload = {"pan": pan}
if name:
payload["name"] = name.strip()
url = f"{self.base_url}/pan"
curl_cmd = f"curl -X POST '{url}' \\\n -H 'x-client-id: {self.client_id or 'YOUR_CLIENT_ID'}' \\\n -H 'x-client-secret: [REDACTED]' \\\n -H 'x-api-version: 2024-01-01' \\\n -H 'Content-Type: application/json' \\\n -d '{json.dumps(payload)}'"
if self.has_credentials():
try:
res = requests.post(url, headers=self._get_headers(), json=payload, timeout=12)
res_json = res.json() if res.headers.get("content-type", "").startswith("application/json") else res.text
if res.status_code == 200:
return {
"source": "live_api",
"status_code": res.status_code,
"payload_sent": payload,
"response": res_json,
"curl": curl_cmd
}
except Exception as e:
logger.error(f"Error calling Cashfree PAN API: {e}")
# Fallback simulation using official Cashfree Sandbox test data
valid_pans = {
"ABCPV1234D": {"name": "JOHN DOE", "type": "INDIVIDUAL"},
"XYZPP4321W": {"name": "ALICE SMITH", "type": "INDIVIDUAL"},
"AZJPG7110R": {"name": "RAHUL SHARMA", "type": "INDIVIDUAL"},
"ABCCD8000T": {"name": "ACME SOLUTIONS PVT LTD", "type": "BUSINESS"},
"XYZHP2000L": {"name": "TECH CORP INDIA", "type": "BUSINESS"},
"AAAHU4383C": {"name": "GLOBAL ENTERPRISES", "type": "BUSINESS"},
"AMJCL2021N": {"name": "CASHFREE INDIA PRIVATE LTD", "type": "BUSINESS"},
}
invalid_pans = ["DEFPV0126D", "TUVPP5678W", "LMNCD8010T", "EFGHH2020L"]
if pan in valid_pans:
reg_name = valid_pans[pan]["name"]
score = 100.0 if not name or name.upper() == reg_name else (85.0 if name.upper() in reg_name or reg_name in name.upper() else 70.0)
result = "DIRECT_MATCH" if score == 100.0 else "GOOD_PARTIAL_MATCH"
mock_res = {
"reference_id": 982341,
"pan": pan,
"registered_name": reg_name,
"type": valid_pans[pan]["type"],
"valid": True,
"message": "PAN verification successful",
"name_match_score": score,
"name_match_result": result,
"status": "SUCCESS"
}
status_code = 200
elif pan in invalid_pans:
mock_res = {
"reference_id": 982342,
"pan": pan,
"valid": False,
"message": "PAN number does not exist or is invalid in income tax database",
"status": "INVALID"
}
status_code = 400
else:
mock_res = {
"reference_id": 982343,
"pan": pan,
"registered_name": name.upper() or "VERIFIED NAME",
"type": pan_type,
"valid": True,
"message": "PAN details fetched successfully (Sandbox Mode)",
"name_match_score": 100.0 if name else 0.0,
"name_match_result": "DIRECT_MATCH" if name else "NOT_APPLICABLE",
"status": "SUCCESS"
}
status_code = 200
return {
"source": "sandbox_simulated",
"status_code": status_code,
"payload_sent": payload,
"response": mock_res,
"curl": curl_cmd
}
# ==================== AADHAAR VERIFICATION & MASKING ====================
def verify_aadhaar(self, aadhaar_number, action="verify"):
aadhaar_number = aadhaar_number.replace(" ", "").strip()
payload = {"aadhaar_number": aadhaar_number}
url = f"{self.base_url}/aadhaar" if action == "verify" else f"{self.base_url}/aadhaar/masking"
curl_cmd = f"curl -X POST '{url}' \\\n -H 'x-client-id: {self.client_id or 'YOUR_CLIENT_ID'}' \\\n -H 'x-client-secret: [REDACTED]' \\\n -H 'x-api-version: 2024-01-01' \\\n -H 'Content-Type: application/json' \\\n -d '{json.dumps(payload)}'"
if self.has_credentials():
try:
res = requests.post(url, headers=self._get_headers(), json=payload, timeout=12)
res_json = res.json() if res.headers.get("content-type", "").startswith("application/json") else res.text
if res.status_code == 200:
return {
"source": "live_api",
"status_code": res.status_code,
"payload_sent": payload,
"response": res_json,
"curl": curl_cmd
}
except Exception as e:
logger.error(f"Error calling Cashfree Aadhaar API: {e}")
# Fallback simulation
valid_aadhaars = ["655675523712", "655675523711"]
invalid_aadhaars = ["655675523710", "655675523709"]
if action == "mask":
masked = "XXXX-XXXX-" + (aadhaar_number[-4:] if len(aadhaar_number) >= 4 else "3712")
mock_res = {
"reference_id": 554312,
"aadhaar_number": masked,
"status": "SUCCESS",
"message": "Aadhaar masking performed successfully per UIDAI regulations",
"qr_code_masked": True,
"first_8_digits_obscured": True
}
status_code = 200
else:
if aadhaar_number in valid_aadhaars or len(aadhaar_number) == 12:
mock_res = {
"reference_id": 554310,
"aadhaar_number": "XXXX-XXXX-" + (aadhaar_number[-4:] if len(aadhaar_number) >= 4 else "3712"),
"valid": True,
"status": "SUCCESS",
"gender": "MALE",
"state": "Karnataka",
"age_band": "20-30",
"mobile_number_masked": "XXXXXX9912",
"message": "Aadhaar verification successful"
}
status_code = 200
elif aadhaar_number in invalid_aadhaars:
mock_res = {
"reference_id": 554311,
"aadhaar_number": aadhaar_number,
"valid": False,
"status": "INVALID",
"message": "Invalid Aadhaar number or not found in UIDAI database"
}
status_code = 400
else:
mock_res = {
"reference_id": 554313,
"valid": True,
"status": "SUCCESS",
"gender": "FEMALE",
"state": "Maharashtra",
"age_band": "25-35",
"mobile_number_masked": "XXXXXX8877",
"message": "Aadhaar details retrieved (Sandbox Mode)"
}
status_code = 200
return {
"source": "sandbox_simulated",
"status_code": status_code,
"payload_sent": payload,
"response": mock_res,
"curl": curl_cmd
}
# ==================== GSTIN VERIFICATION ====================
def verify_gstin(self, gstin):
gstin = gstin.strip().upper()
payload = {"gstin": gstin}
url = f"{self.base_url}/gstin"
curl_cmd = f"curl -X POST '{url}' \\\n -H 'x-client-id: {self.client_id or 'YOUR_CLIENT_ID'}' \\\n -H 'x-client-secret: [REDACTED]' \\\n -H 'x-api-version: 2024-01-01' \\\n -H 'Content-Type: application/json' \\\n -d '{json.dumps(payload)}'"
if self.has_credentials():
try:
res = requests.post(url, headers=self._get_headers(), json=payload, timeout=12)
res_json = res.json() if res.headers.get("content-type", "").startswith("application/json") else res.text
if res.status_code == 200:
return {
"source": "live_api",
"status_code": res.status_code,
"payload_sent": payload,
"response": res_json,
"curl": curl_cmd
}
except Exception as e:
logger.error(f"Error calling Cashfree GSTIN API: {e}")
# Fallback simulation
valid_gstins = {
"29AAICP2912R1ZR": {
"legal_name": "CASHFREE PAYMENTS INDIA PRIVATE LIMITED",
"trade_name": "Cashfree Payments",
"state_jurisdiction": "Karnataka",
"principal_address": "Koramangala 5th Block, Bengaluru, Karnataka 560095",
"taxpayer_type": "Regular",
"registration_date": "2015-09-01"
},
"27ABCFGH5678D1Z": {
"legal_name": "APEX ENTERPRISES INDIA PVT LTD",
"trade_name": "Apex Digital",
"state_jurisdiction": "Maharashtra",
"principal_address": "Bandra West, Mumbai, Maharashtra 400050",
"taxpayer_type": "Regular",
"registration_date": "2018-04-15"
},
"36JKLPQR9012E2Y": {
"legal_name": "HYDERABAD TECHSOLUTIONS LLP",
"trade_name": "TechSolutions",
"state_jurisdiction": "Telangana",
"principal_address": "HITEC City, Hyderabad, Telangana 500081",
"taxpayer_type": "Regular",
"registration_date": "2020-01-10"
}
}
invalid_gstins = ["29AAIZP2912R1ZR", "05BBICA2912R1ZR"]
if gstin in valid_gstins:
data = valid_gstins[gstin]
mock_res = {
"reference_id": 771239,
"gstin": gstin,
"gstin_status": "ACTIVE",
"legal_name_of_business": data["legal_name"],
"trade_name": data["trade_name"],
"taxpayer_type": data["taxpayer_type"],
"registration_date": data["registration_date"],
"state_jurisdiction": data["state_jurisdiction"],
"principal_place_of_business_address": data["principal_address"],
"valid": True,
"status": "SUCCESS"
}
status_code = 200
elif gstin in invalid_gstins:
mock_res = {
"reference_id": 771240,
"gstin": gstin,
"valid": False,
"status": "INVALID",
"message": "GSTIN number is invalid or not registered in GST portal"
}
status_code = 400
else:
mock_res = {
"reference_id": 771241,
"gstin": gstin,
"gstin_status": "ACTIVE",
"legal_name_of_business": "VERIFIED BUSINESS ENTITY",
"trade_name": "Verified Brand",
"taxpayer_type": "Regular",
"registration_date": "2021-06-01",
"state_jurisdiction": "India",
"principal_place_of_business_address": "Main Business Street, Industrial Area, India",
"valid": True,
"status": "SUCCESS"
}
status_code = 200
return {
"source": "sandbox_simulated",
"status_code": status_code,
"payload_sent": payload,
"response": mock_res,
"curl": curl_cmd
}
# ==================== IFSC VERIFICATION & LOOKUP ====================
def verify_ifsc(self, ifsc):
ifsc = ifsc.strip().upper()
payload = {"ifsc": ifsc}
url = f"{self.base_url}/ifsc"
curl_cmd = f"curl -X POST '{url}' \\\n -H 'x-client-id: {self.client_id or 'YOUR_CLIENT_ID'}' \\\n -H 'x-client-secret: [REDACTED]' \\\n -H 'x-api-version: 2024-01-01' \\\n -H 'Content-Type: application/json' \\\n -d '{json.dumps(payload)}'"
if self.has_credentials():
try:
res = requests.post(url, headers=self._get_headers(), json=payload, timeout=12)
res_json = res.json() if res.headers.get("content-type", "").startswith("application/json") else res.text
if res.status_code == 200:
return {
"source": "live_api",
"status_code": res.status_code,
"payload_sent": payload,
"response": res_json,
"curl": curl_cmd
}
except Exception as e:
logger.error(f"Error calling Cashfree IFSC API: {e}")
# Fallback simulation
ifsc_database = {
"YESB0000262": {
"bank": "YES BANK",
"branch": "KORAMANGALA",
"address": "GROUND FLOOR, NO 43, 80 FEET ROAD, KORAMANGALA 4TH BLOCK, BENGALURU 560034",
"city": "BENGALURU",
"state": "KARNATAKA",
"micr": "560532002",
"modes": ["IMPS", "NEFT", "RTGS", "UPI"]
},
"HDFC0000001": {
"bank": "HDFC BANK",
"branch": "MUMBAI - FORT",
"address": "MANEKJI WADIA BLDG, GROUND FLOOR, NANIK MOTWANI MARG, FORT, MUMBAI 400023",
"city": "MUMBAI",
"state": "MAHARASHTRA",
"micr": "400240002",
"modes": ["IMPS", "NEFT", "RTGS", "UPI"]
},
"SBIN0008752": {
"bank": "STATE BANK OF INDIA",
"branch": "INDIRANAGAR",
"address": "100 FEET ROAD, INDIRANAGAR 1ST STAGE, BENGALURU 560038",
"city": "BENGALURU",
"state": "KARNATAKA",
"micr": "560002045",
"modes": ["IMPS", "NEFT", "RTGS", "UPI"]
},
"SCBL0036078": {
"bank": "STANDARD CHARTERED BANK",
"branch": "MG ROAD BENGALURU",
"address": "26-27 RAHEJA TOWERS, MG ROAD, BENGALURU 560001",
"city": "BENGALURU",
"state": "KARNATAKA",
"micr": "560036002",
"modes": ["IMPS", "NEFT", "RTGS", "UPI"]
}
}
invalid_ifscs = ["CNRR0002640", "INVALID001"]
if ifsc in ifsc_database:
b_info = ifsc_database[ifsc]
mock_res = {
"reference_id": 331892,
"ifsc": ifsc,
"valid": True,
"bank": b_info["bank"],
"branch": b_info["branch"],
"address": b_info["address"],
"city": b_info["city"],
"state": b_info["state"],
"micr": b_info["micr"],
"supported_modes": b_info["modes"],
"status": "SUCCESS"
}
status_code = 200
elif ifsc in invalid_ifscs:
mock_res = {
"reference_id": 331893,
"ifsc": ifsc,
"valid": False,
"status": "INVALID",
"message": "IFSC code is invalid or branch does not exist in RBI registry"
}
status_code = 400
else:
mock_res = {
"reference_id": 331894,
"ifsc": ifsc,
"valid": True,
"bank": "INDIAN NATIONAL BANK",
"branch": "MAIN BRANCH",
"address": "COMMERCIAL PLAZA, METRO STATION ROAD, INDIA",
"city": "METRO CITY",
"state": "INDIA",
"micr": "110002999",
"supported_modes": ["IMPS", "NEFT", "RTGS", "UPI"],
"status": "SUCCESS"
}
status_code = 200
return {
"source": "sandbox_simulated",
"status_code": status_code,
"payload_sent": payload,
"response": mock_res,
"curl": curl_cmd
}
# ==================== BANK ACCOUNT VERIFICATION (BAV SYNC) ====================
def verify_bank_account(self, bank_account, ifsc, name="", phone=""):
bank_account = bank_account.strip()
ifsc = ifsc.strip().upper()
payload = {
"bank_account": bank_account,
"ifsc": ifsc
}
if name:
payload["name"] = name.strip()
if phone:
payload["phone"] = phone.strip()
url = f"{self.base_url}/bank-account/sync"
curl_cmd = f"curl -X POST '{url}' \\\n -H 'x-client-id: {self.client_id or 'YOUR_CLIENT_ID'}' \\\n -H 'x-client-secret: [REDACTED]' \\\n -H 'x-api-version: 2024-01-01' \\\n -H 'Content-Type: application/json' \\\n -d '{json.dumps(payload)}'"
if self.has_credentials():
try:
res = requests.post(url, headers=self._get_headers(), json=payload, timeout=15)
res_json = res.json() if res.headers.get("content-type", "").startswith("application/json") else res.text
if res.status_code == 200:
return {
"source": "live_api",
"status_code": res.status_code,
"payload_sent": payload,
"response": res_json,
"curl": curl_cmd
}
except Exception as e:
logger.error(f"Error calling Cashfree Bank Account API: {e}")
# Fallback simulation
valid_accounts = ["026291800001191", "00011020001772", "000890289871772", "1233943142", "388108022658"]
failed_accounts = ["000100289877623", "026291800001190"]
if bank_account in valid_accounts or (len(bank_account) >= 9 and bank_account not in failed_accounts):
registered_name = name.upper() if name else "JOHN DOE"
mock_res = {
"reference_id": 441091,
"bank_account": bank_account,
"ifsc": ifsc,
"account_status": "VALID",
"account_name": registered_name,
"name_match_score": 100.0 if name else 0.0,
"name_match_result": "DIRECT_MATCH" if name else "NOT_APPLICABLE",
"utr": "427189012345",
"bank_name": "YES BANK" if "YESB" in ifsc else "HDFC BANK",
"city": "BENGALURU",
"message": "Bank account is active and verified via penny drop",
"status": "SUCCESS"
}
status_code = 200
else:
mock_res = {
"reference_id": 441092,
"bank_account": bank_account,
"ifsc": ifsc,
"account_status": "INVALID",
"message": "Bank account number is invalid or rejected by destination bank",
"status": "FAILED"
}
status_code = 400
return {
"source": "sandbox_simulated",
"status_code": status_code,
"payload_sent": payload,
"response": mock_res,
"curl": curl_cmd
}