""" EkstraPAY — Credit Card API SDK (Python, dependency-free) ========================================================= Server-to-server client for merchants integrating the Credit Card API. Handles HMAC-SHA256 request signing exactly as the backend (CCApiAuth) expects. Quick start ----------- from ekstrapay_cc import EkstraPayCC, EkstraPayError cc = EkstraPayCC( api_key="YOUR_API_KEY", api_token="YOUR_API_TOKEN", base_url="https://v2.ekstrapay.com/api", ) res = cc.create_payment( amount=250.75, order_id="ORD-1001", customer_ref="cust-123", tc_no="11111111111", first_name="Ali", last_name="Veli", return_url_success="https://merchant.com/ok", return_url_fail="https://merchant.com/fail", ) # 3D -> res["redirect_url"] # native/link -> res["pay_screen"]["pay_url"] # KYC required -> res["kyc"]["code"] Signing algorithm (must match server byte-for-byte) --------------------------------------------------- cleaned = drop empty("")/None, trim strings, whole floats -> int json_str = compact JSON of cleaned (no spaces, unescaped unicode/slashes); empty object -> "[]" message = api_key + "|" + timestamp + "|" + nonce + "|" + json_str signature = HMAC_SHA256(message, api_token) (hex) body sent = json_str (the canonical string is signed AND sent) """ import hmac import hashlib import json import time import uuid import urllib.request import urllib.error __version__ = "1.0.0" class EkstraPayError(Exception): """Raised on any API/business error. Carries HTTP code and error_code.""" def __init__(self, message, http_code=None, error_code=None, data=None): super().__init__(message) self.message = message self.http_code = http_code self.error_code = error_code # e.g. "CC-AUTH-005" self.data = data class EkstraPayCC: def __init__(self, api_key, api_token, base_url="https://v2.ekstrapay.com/api", timeout=30): if not api_key or not api_token: raise ValueError("api_key and api_token are required") self.api_key = api_key self.api_token = api_token self.base_url = base_url.rstrip("/") self.timeout = timeout # ── Public API ────────────────────────────────────────────── def create_payment(self, **body): """Create a payment. Returns the `data` object of the response.""" return self._post("/v1/cc/payment/create", body) def check_payment(self, transaction_id, sync=False): """Get the current status of a transaction (sync=True also queries provider).""" return self._post("/v1/cc/payment/check", {"transaction_id": transaction_id, "sync": sync}) def cancel_payment(self, transaction_id): """Cancel a non-finalized transaction.""" return self._post("/v1/cc/payment/cancel", {"transaction_id": transaction_id}) def detail_payment(self, transaction_id): """Get the full detail of a transaction.""" return self._post("/v1/cc/payment/detail", {"transaction_id": transaction_id}) # ── Callback verification ─────────────────────────────────── @staticmethod def verify_callback(payload): """ Verify an incoming callback's `hash`. Returns True if authentic. The hash is: sha256 of these fields joined by ":" IN THIS ORDER, with null/missing optional fields as "" (positions preserved): transaction_id : order_id : customer_ref : tc_no : amount : final_amount : status `amount`/`final_amount` are the 2-decimal strings from the payload (final_amount may be null -> ""). Usage (Flask/Django/etc.): data = request.get_json() if not EkstraPayCC.verify_callback(data): abort(400) # forged / tampered """ def s(v): return "" if v is None else str(v) parts = [ s(payload.get("transaction_id")), s(payload.get("order_id")), s(payload.get("customer_ref")), s(payload.get("tc_no")), s(payload.get("amount")), s(payload.get("final_amount")), s(payload.get("status")), ] expected = hashlib.sha256(":".join(parts).encode("utf-8")).hexdigest() return hmac.compare_digest(expected, s(payload.get("hash"))) # ── Internals ─────────────────────────────────────────────── @staticmethod def _clean(body): cleaned = {} for key, val in body.items(): if isinstance(val, str): val = val.strip() if isinstance(val, bool): pass # keep booleans as-is elif isinstance(val, float) and val.is_integer(): val = int(val) # 250.0 -> 250 (match PHP json_encode) if val is None or val == "": continue cleaned[key] = val return cleaned @staticmethod def _canonical_json(cleaned): if not cleaned: return "[]" # PHP json_encode([]) === "[]" return json.dumps(cleaned, ensure_ascii=False, separators=(",", ":")) def _sign(self, json_str): ts = str(int(time.time())) nonce = uuid.uuid4().hex message = "{}|{}|{}|{}".format(self.api_key, ts, nonce, json_str) sig = hmac.new(self.api_token.encode("utf-8"), message.encode("utf-8"), hashlib.sha256).hexdigest() return ts, nonce, sig def _post(self, path, body): cleaned = self._clean(body) json_str = self._canonical_json(cleaned) ts, nonce, sig = self._sign(json_str) req = urllib.request.Request( self.base_url + path, data=json_str.encode("utf-8"), # send the exact signed canonical string method="POST", headers={ "Content-Type": "application/json", "Accept": "application/json", "X-API-Key": self.api_key, "X-Timestamp": ts, "X-Nonce": nonce, "X-Signature": sig, }, ) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: payload = json.loads(resp.read().decode("utf-8")) return self._handle(payload, resp.status) except urllib.error.HTTPError as e: try: payload = json.loads(e.read().decode("utf-8")) except Exception: payload = {} self._raise(payload, e.code) except urllib.error.URLError as e: raise EkstraPayError("Connection error: {}".format(e.reason)) @staticmethod def _handle(payload, http_code): if payload.get("status") == "error": EkstraPayCC._raise(payload, http_code) return payload.get("data", {}) @staticmethod def _raise(payload, http_code): data = payload.get("data") or {} raise EkstraPayError( payload.get("description") or payload.get("message") or "Request failed", http_code=http_code, error_code=data.get("error_code") if isinstance(data, dict) else None, data=data, )