""" CashoutGuard client for Python 3.8+ (single file, standard library only). from cashoutguard import CashoutGuard cg = CashoutGuard(os.environ["CASHOUTGUARD_SECRET"]) risk = cg.evaluate({ "event": "cashout", "account_id": str(user.id), "request_id": form.get("cg_request_id"), # from CashoutGuard.collect() in the browser "ip": client_ip, "amount": 5.0, "currency": "USD", "payout_address": wallet, }) if risk.is_blocked(): hold_for_review() Fail open: timeouts, network errors and server errors never raise. They return action "allow" with result.error set, so an outage on our side never stops your users. Docs: https://cashoutguard.com/docs """ import hashlib import hmac import json import urllib.error import urllib.parse import urllib.request VERSION = "1.0.0" DEFAULT_BASE_URL = "https://cashoutguard.com" DEFAULT_TIMEOUT = 2.0 SIGNATURE_HEADER = "X-CashoutGuard-Signature" EVENTS = ("signup", "login", "offer_click", "conversion", "cashout", "custom") class Result: """Outcome of one API call. Act on `action` (is_blocked()/needs_review()), not on `decision`.""" def __init__(self, raw=None, error=None, error_message=None, status=0): raw = raw if isinstance(raw, dict) else {} self.raw = raw self.status = status self.error = error self.error_message = error_message self.decision = raw.get("decision", "allow") if not error else "allow" # /v1/accounts/{id} has no "action": the account decision is the only verdict there. self.action = raw.get("action", self.decision) if not error else "allow" self.score = raw.get("score") self.reasons = [r for r in raw.get("reasons", []) if isinstance(r, dict)] self.event_id = raw.get("event_id") self.mode = raw.get("mode") self.account = raw.get("account") if isinstance(raw.get("account"), dict) else None def is_blocked(self): return self.action == "block" def needs_review(self): return self.action == "review" def is_allowed(self): return self.action == "allow" def ok(self): """True when the API answered with 2xx and a JSON body.""" return self.error is None def reason_codes(self): return [r.get("code") for r in self.reasons if r.get("code")] def has_reason(self, code): return code in self.reason_codes() def __repr__(self): return "" % (self.action, self.score, self.reason_codes(), self.error) class CashoutGuard: def __init__(self, secret_key, base_url=DEFAULT_BASE_URL, timeout=DEFAULT_TIMEOUT): if not secret_key or not str(secret_key).startswith("sk_"): raise ValueError("CashoutGuard: pass your secret key (sk_...), never the public key.") self.secret_key = str(secret_key) self.base_url = base_url.rstrip("/") self.timeout = timeout def evaluate(self, event): """POST /v1/evaluate. `event` is a dict: event, account_id and the optional fields in the docs.""" body = {k: v for k, v in dict(event).items() if v is not None and v != ""} if body.get("event") not in EVENTS: return Result(error="invalid_event", error_message="event must be one of %s" % ", ".join(EVENTS)) if "account_id" in body: body["account_id"] = str(body["account_id"]) return self._request("POST", "/v1/evaluate", body) def account(self, account_id): """GET /v1/accounts/{id}. A 404 returns error "not_found".""" return self._request("GET", "/v1/accounts/" + urllib.parse.quote(str(account_id), safe=""), None) def outcomes(self, ids, outcome="reversed"): """POST /v1/outcomes: what happened to conversions afterwards ("reversed" or "valid"), by transaction_id, click_id or event_id (up to 1,000 per call). The Accuracy page measures the engine on them.""" return self._request("POST", "/v1/outcomes", {"outcome": outcome, "ids": [str(i) for i in ids]}) def cashout_status(self, id, status, note=None): """POST /v1/cashouts/status: you paid, held or denied a cashout in your own admin ("paid", "held", "denied" or "pending"), so the cashout queue follows. id: the cashout_id or event_id from evaluate(), or your transaction_id.""" body = {"id": str(id), "status": status} if note is not None: body["note"] = note return self._request("POST", "/v1/cashouts/status", body) def lookup(self, ip=None, email=None): """GET /v1/lookup: reputation of an IP and/or an email without an event (never counts as an active user).""" params = {k: v for k, v in (("ip", ip), ("email", email)) if v} return self._request("GET", "/v1/lookup?" + urllib.parse.urlencode(params), None) @staticmethod def verify_webhook(raw_body, signature, secret): """True when the X-CashoutGuard-Signature header matches the raw request body.""" if not signature or not secret: return False if isinstance(raw_body, str): raw_body = raw_body.encode("utf-8") expected = hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, str(signature).strip()) def _request(self, method, path, body): data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request(self.base_url + path, data=data, method=method, headers={ "Authorization": "Bearer " + self.secret_key, "Accept": "application/json", "Content-Type": "application/json", "User-Agent": "cashoutguard-python/" + VERSION, }) try: with urllib.request.urlopen(req, timeout=self.timeout) as res: return Result(json.loads(res.read().decode("utf-8") or "{}"), status=res.status) except urllib.error.HTTPError as e: try: raw = json.loads(e.read().decode("utf-8") or "{}") except ValueError: raw = None code = "not_found" if e.code == 404 else "http_%d" % e.code msg = raw.get("error") if isinstance(raw, dict) else str(e) return Result(raw, error=code, error_message=msg, status=e.code) except Exception as e: # timeout, DNS, TLS, connection reset: fail open name = "timeout" if "timed out" in str(e).lower() else "network_error" return Result(error=name, error_message=str(e))