# AntPay hosted checkout - everything your server needs. Python 3.8 or newer, standard library only. # Keep the three values below on your server. Never put them in a browser or a mobile app. # [setup] import hashlib import hmac import json import os import time import urllib.error import urllib.parse import urllib.request import re from http.server import BaseHTTPRequestHandler API_BASE = os.environ.get("ANTPAY_API_BASE", "https://api.antpay.network") API_KEY = os.environ.get("ANTPAY_API_KEY", "") # your payment key POSTBACK_SECRET = os.environ.get("ANTPAY_POSTBACK_SECRET", "") # issued with your account, 64 characters SIGNING_SECRET = os.environ.get("ANTPAY_SIGNING_SECRET", "") # only if request signing is on for your account # A reference you can always read back: letters, digits, dot, underscore, dash; 1 to 120 characters; no "..". REFERENCE = re.compile(r"^(?!.*[.][.])[A-Za-z0-9._-]{1,120}$") # Wait at least 80 seconds for an answer. The API can take up to about 75 s on a create; giving up earlier and # sending the create again while the first is still running is how one order ends up with two checkouts. TIMEOUT_S = 80 # [/setup] # [sign] # Request signing (only if it is switched on for your account). # canonical = "{timestamp}.{METHOD}.{path with query}.{body exactly as sent}" def sign_request(method, path, body="", timestamp=None): timestamp = timestamp or str(int(time.time())) # SECONDS canonical = f"{timestamp}.{method.upper()}.{path}.{body}" signature = hmac.new(SIGNING_SECRET.encode(), canonical.encode("utf-8"), hashlib.sha256).hexdigest() return {"X-Timestamp": timestamp, "X-Signature": signature} # the secret is used as the 64-character STRING # [/sign] # [call] def call(method, path, payload=None): body = "" if payload is None else json.dumps(payload, separators=(",", ":")) # serialise ONCE headers = {"X-Api-Key": API_KEY} if body: headers["Content-Type"] = "application/json" if SIGNING_SECRET: headers.update(sign_request(method, path, body)) req = urllib.request.Request(API_BASE + path, data=body.encode("utf-8") if body else None, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=TIMEOUT_S) as res: return res.status, json.loads(res.read().decode("utf-8")) except urllib.error.HTTPError as e: # 4xx / 5xx still carry a JSON body text = e.read().decode("utf-8") try: return e.code, json.loads(text) except ValueError: return e.code, None # [/call] # [create] # Step 1: create the checkout on your server, store the checkoutId, redirect the customer. def create_checkout(order_id, total): if not REFERENCE.match(order_id): raise ValueError('reference: use letters, digits, ".", "_" and "-" only (1 to 120 characters)') status, data = call("POST", "/api/v1/checkouts", { "reference": order_id, # your order id, unique per API key "amount": total, # whole rupees, minimum 300 "paymentMethod": "qr", "paymentType": "onetime", "returnUrl": "https://shop.example.com/orders/" + urllib.parse.quote(order_id, safe=""), }) if status == 200 and data and data.get("success"): return {"checkoutId": data["checkoutId"], "checkoutUrl": data["checkoutUrl"], "expiresAt": data["expiresAt"]} # 4xx: nothing was created (a duplicate reference already exists: look it up). # 5xx or no answer: the checkout may exist. Ask by reference before you retry. raise RuntimeError(f"checkout not created (HTTP {status}): {data}") # [/create] # [status] # Ask for the payment by YOUR reference. Use it after an unclear create, and as a fallback to the postback. # Returns None ONLY when the API itself says there is no such payment. Any other 404 means the address was refused # (a wrong path, or a reference with characters that cannot travel in a URL): that is an error, not "no payment". def get_by_reference(reference): status, data = call("GET", "/api/v1/transactions/" + urllib.parse.quote(reference, safe="")) if status == 200 and data and data.get("success"): return data["transaction"] # ["status"] is pending | completed | failed if status == 404 and data and data.get("error") == "Transaction not found": return None raise RuntimeError(f"status call failed (HTTP {status}, {data.get('error') if data else None})") def get_checkout(checkout_id): status, data = call("GET", "/api/v1/checkouts/" + urllib.parse.quote(checkout_id, safe="")) if status == 200 and data and data.get("success"): return data["checkout"] if status == 404 and data and data.get("error") == "Checkout session not found": return None raise RuntimeError(f"status call failed (HTTP {status}, {data.get('error') if data else None})") # [/status] # [verify] # A postback is genuine when X-Mint-Signature equals "sha256=" + HMAC-SHA256(raw body, your postback secret). # raw_body must be the bytes exactly as received, not a re-encoded object. def verify_postback(raw_body, signature_header, secret=None): secret = POSTBACK_SECRET if secret is None else secret if not secret or not isinstance(signature_header, str): return False expected = "sha256=" + hmac.new(secret.encode(), raw_body, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, signature_header) # [/verify] # [postback] # Step 4: receive the postback. Verify, record once, answer 2xx within 10 seconds. # on_result(payload) is YOUR function: it must be safe to call twice with the same transactionId + status. def postback_handler(on_result): class Handler(BaseHTTPRequestHandler): def do_POST(self): raw = self.rfile.read(int(self.headers.get("Content-Length", "0"))) if not verify_postback(raw, self.headers.get("X-Mint-Signature")): self.send_response(401) # not from AntPay, or the body was changed self.end_headers() return p = json.loads(raw.decode("utf-8")) try: # p["status"] is "completed" or "failed" (never "pending"). Find your order by p["checkoutId"]: # the id you stored when you created the checkout. Do not match on the reference alone. on_result(p) self.send_response(200) except Exception: self.send_response(500) # we try again: 3 attempts in total, 5 s apart self.end_headers() def log_message(self, *args): pass return Handler # [/postback]