# Cryptanio 1.0.0 62 / public API v1 """Cryptanio public API /v1. Python 3.10+, standard library only. Server-side client: do not share API keys with browser code. No redirects, automatic retries, pagination, or SSE replay. Monetary amounts stay strings. """ import http.client import json import math import re import socket from urllib.parse import quote, urlencode, urlsplit _MAX_RESPONSE_BYTES = 8 * 1024 * 1024 _WATCH_FILTERS = {"project", "chain", "limit", "before"} _EVENT_FILTERS = {"project", "watch_id", "chain", "type", "address", "tx_id", "limit", "before"} _INVOICE_FILTERS = {"project", "chain", "status", "limit", "before"} class ApiError(Exception): def __init__(self, status, code, message, retry_after=None): super().__init__(message) self.status = status self.code = code self.message = message self.retry_after = retry_after class Cryptanio: def __init__(self, api_key="", *, base_url="https://api.cryptanio.com", timeout=30): if not isinstance(api_key, str) or "\r" in api_key or "\n" in api_key: raise ValueError("API key must be a string without line breaks.") if isinstance(timeout, bool) or not isinstance(timeout, (int, float)) or not math.isfinite(timeout) or timeout <= 0 or timeout > 300: raise ValueError("Timeout must be positive finite seconds, at most 300.") try: if not isinstance(base_url, str) or any(c.isspace() for c in base_url) or "?" in base_url or "#" in base_url: raise ValueError() url = urlsplit(base_url) port = url.port loopback = url.hostname in {"localhost", "127.0.0.1", "::1"} if not url.hostname or url.username is not None or url.password is not None or url.query or url.fragment or url.path not in {"", "/"}: raise ValueError() if url.scheme != "https" and not (url.scheme == "http" and loopback): raise ValueError() except (ValueError, TypeError): raise ValueError("Base URL must be an HTTPS origin; HTTP is allowed only on loopback.") from None self._api_key = api_key self._hostname = url.hostname self._port = port self._https = url.scheme == "https" self._timeout = timeout def create_watch(self, body): return self._request("POST", "/v1/watches", body=body) def list_watches(self, filters=None): return self._request("GET", "/v1/watches", query=self._query(filters, _WATCH_FILTERS)) def get_watch(self, id): return self._request("GET", "/v1/watches/" + self._id(id)) def delete_watch(self, id): return self._request("DELETE", "/v1/watches/" + self._id(id)) def list_events(self, filters=None): return self._request("GET", "/v1/events", query=self._query(filters, _EVENT_FILTERS)) def list_payments(self, filters=None): return self._request("GET", "/v1/payments", query=self._query(filters, _EVENT_FILTERS)) def create_invoice(self, body, idempotency_key): if not isinstance(idempotency_key, str) or re.fullmatch(r"[A-Za-z0-9_-]{16,80}", idempotency_key) is None: raise ValueError("Idempotency key must contain 16–80 ASCII letters, digits, underscores or hyphens.") return self._request("POST", "/v1/invoices", body=body, idempotency_key=idempotency_key) def list_invoices(self, filters=None): return self._request("GET", "/v1/invoices", query=self._query(filters, _INVOICE_FILTERS)) def get_invoice(self, id): return self._request("GET", "/v1/invoices/" + self._id(id)) def cancel_invoice(self, id): return self._request("POST", "/v1/invoices/" + self._id(id) + "/cancel") def create_endpoint(self, body): return self._request("POST", "/v1/endpoints", body=body) def list_endpoints(self): return self._request("GET", "/v1/endpoints") def delete_endpoint(self, id): return self._request("DELETE", "/v1/endpoints/" + self._id(id)) def stream_events(self): """Return native HTTPResponse byte stream. Use `with` or close(). Timeout applies to each socket operation (including stream reads), not the total stream lifetime. No automatic parsing or reconnect. """ return self._request("GET", "/v1/stream", stream=True) def get_health(self): return self._request("GET", "/v1/health", authenticated=False) def list_chains(self): return self._request("GET", "/v1/chains", authenticated=False) @staticmethod def _id(id): if not isinstance(id, str) or not id.strip() or id in {".", ".."}: raise ValueError("ID must be a nonempty string, not a relative path segment.") return quote(id, safe="") @staticmethod def _query(filters, allowed): if filters is None: filters = {} if not isinstance(filters, dict): raise ValueError("Filters must be a dictionary.") values = {} for name, value in filters.items(): if name not in allowed: raise ValueError("Unknown list filter.") if value is None: continue if name == "limit": if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 200: raise ValueError("Limit must be an integer between 1 and 200.") elif not isinstance(value, str): raise ValueError("Filter values must be strings (except limit).") values[name] = value return urlencode(values) def _redact(self, message): return message.replace(self._api_key, "[redacted]") if self._api_key else message def _request(self, method, path, *, body=None, query="", authenticated=True, idempotency_key=None, stream=False): if authenticated and not self._api_key: raise ValueError("An API key is required for this operation.") headers = {"Accept": "text/event-stream" if stream else "application/json", "Connection": "close"} if authenticated: headers["Authorization"] = "Bearer " + self._api_key if idempotency_key: headers["Idempotency-Key"] = idempotency_key encoded = None if body is not None: if not isinstance(body, dict): raise ValueError("Body must be a dictionary.") if "amount" in body and not isinstance(body["amount"], str): raise ValueError("Amount must be a decimal string.") try: encoded = json.dumps(body, ensure_ascii=False, allow_nan=False).encode("utf-8") except (TypeError, ValueError): raise ValueError("Body must be JSON serializable.") from None headers["Content-Type"] = "application/json" elif method == "POST" and not path.endswith("/cancel"): raise ValueError("Body must be a dictionary.") connection_type = http.client.HTTPSConnection if self._https else http.client.HTTPConnection connection = connection_type(self._hostname, self._port, timeout=self._timeout) response = None keep_stream = False status, retry_after = 0, None try: connection.request(method, path + ("?" + query if query else ""), body=encoded, headers=headers) response = connection.getresponse() status, retry_after = response.status, response.getheader("Retry-After") if 300 <= status < 400: raise ApiError(status, "redirect_not_followed", "API redirects are not followed.", retry_after) if stream and 200 <= status < 300: if not (response.getheader("Content-Type") or "").lower().startswith("text/event-stream"): raise ApiError(status, "unreadable_response", "Expected an event stream.", retry_after) # Connection: close lets HTTPResponse own the socket after headers. keep_stream = True return response raw = response.read(_MAX_RESPONSE_BYTES + 1) if len(raw) > _MAX_RESPONSE_BYTES: raise ApiError(status, "response_too_large", "API response exceeds 8 MiB.", retry_after) if status == 204 and not raw: return None try: data = json.loads(raw.decode("utf-8")) except (UnicodeError, ValueError): raise ApiError(status, "unreadable_response", "The API returned invalid JSON.", retry_after) from None if not 200 <= status < 300: error = data.get("error") if isinstance(data, dict) else None error = error if isinstance(error, dict) else {} code = error.get("code") if isinstance(error.get("code"), str) else "http_error" message = error.get("message") if isinstance(error.get("message"), str) else "The API request failed." raise ApiError(status, self._redact(code), self._redact(message), retry_after) return data except (socket.timeout, TimeoutError): raise ApiError(status, "timeout", "Request timed out.", retry_after) from None except (OSError, http.client.HTTPException, UnicodeError): raise ApiError(status, "transport_error", "Unable to read the API response.", retry_after) from None finally: if not keep_stream: if response is not None: response.close() connection.close()