""" Foliox License - Python Automation client, v2 (for YOUR server, never for your customers' machines). Creates and manages license keys from your own backend - a checkout webhook, a Discord bot, a cron job. from automation_client import AutomationClient api = AutomationClient("https://YOUR-SERVER", token=os.environ["LICENSE_API_TOKEN"]) keys = api.create_keys("YOUR_PRODUCT_ID", count=1, duration_days=30, customer_email="buyer@example.com") print(keys[0]["key"]) IMPORTANT - where this may run: the token can mint keys, so it belongs on a server you control (an environment variable or secret manager), never inside an app, installer or web page your customers receive. Give each token only the permissions its job needs and an expiry, and revoke it from the dashboard if it ever leaks. The software your customers run uses ``license_client.LicenseClient`` instead, which needs no secret at all. What this client does for you (and why): * HTTPS only (loopback excepted, for development), certificates always verified, TLS 1.2+, redirects never followed - a redirect would carry the token to another host. * The token is sent only in the ``Authorization`` header, is never written into an error message, log line or ``repr()``, and is checked for shape when the client is created so a typo fails immediately. * Retries use exponential backoff with jitter inside a hard deadline - but only where a retry is safe. Reads and the idempotent actions (revoke, restore, reset device) are retried on network errors, 429 and 5xx. Creating keys is *not* idempotent, so it is retried only when the request provably never reached the server (a refused connection, a failed DNS lookup or HTTP 429) - never after a timeout, which could mint duplicates. * Keys you pass in are validated and URL-encoded before they go into a request path. * Every failure is a typed exception (``NotFoundError``, ``PlanLimitError``, ``RateLimitedError`` ...) carrying the HTTP status, the server's error ``code`` and a message. Nothing else is ever swallowed. Python 3.9+, standard library only. """ from __future__ import annotations import http.client import json import random import re import socket import ssl import time import urllib.error import urllib.parse import urllib.request from typing import Any, Callable, Dict, List, Optional __version__ = "2.0.0" __all__ = [ "AutomationClient", "AutomationError", "AutomationConfigError", "AuthenticationError", "PermissionDeniedError", "PlanLimitError", "NotFoundError", "ValidationFailedError", "RateLimitedError", "ServerError", "NetworkError", ] _MAX_BODY = 1024 * 1024 _MAX_RETRY_AFTER = 10.0 _LOOPBACK_HOSTS = {"localhost", "127.0.0.1", "::1"} _TOKEN_RE = re.compile(r"^lk_[A-Za-z0-9_-]{43}$") _PRODUCT_RE = re.compile(r"^[a-z0-9-]{1,64}$") _KEY_RE = re.compile(r"^[0-9A-Z-]{10,64}$") _ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" _KEY_SHAPE = re.compile(r"^[%s]{5}(-[%s]{5}){3}-[%s]$" % (_ALPHABET, _ALPHABET, _ALPHABET)) # --------------------------------------------------------------------------- # Errors # --------------------------------------------------------------------------- class AutomationError(Exception): """Base class. ``status`` is the HTTP status (None for network failures), ``code`` the server's error code.""" def __init__(self, message: str, *, status: Optional[int] = None, code: Optional[str] = None, details: Any = None, fields: Any = None): super().__init__(message) self.message = message self.status = status self.code = code self.details = details self.fields = fields def __str__(self) -> str: parts = [self.message] if self.status: parts.append(f"(HTTP {self.status}{', ' + self.code if self.code else ''})") return " ".join(parts) class AutomationConfigError(AutomationError, ValueError): """The client was configured with something unsafe or malformed (raised at construction).""" class AuthenticationError(AutomationError): """401: the token is missing, malformed, expired or revoked.""" class PermissionDeniedError(AutomationError): """403: the token lacks the permission (``code == 'INSUFFICIENT_SCOPE'``) or the account is suspended (``'ACCOUNT_SUSPENDED'``).""" class PlanLimitError(AutomationError): """402: the plan's limit was reached. ``details`` says which (``{'limit': 'keys', 'max': 500, 'plan': 'free', ...}``).""" class NotFoundError(AutomationError): """404: no such product or key (or it belongs to someone else - the two look the same on purpose).""" class ValidationFailedError(AutomationError): """400: the request was rejected as malformed; ``fields`` says what was wrong.""" class RateLimitedError(AutomationError): """429: too many requests. ``retry_after`` is the server's hint in seconds, if it sent one.""" def __init__(self, message: str, *, retry_after: Optional[float] = None, **kwargs: Any): super().__init__(message, **kwargs) self.retry_after = retry_after class ServerError(AutomationError): """5xx: the service had a problem. Safe to try again later.""" class NetworkError(AutomationError): """The service could not be reached, or its answer was unusable.""" # --------------------------------------------------------------------------- # Transport # --------------------------------------------------------------------------- class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D401 return None class _Retryable(Exception): def __init__(self, error: AutomationError, retry_after: Optional[float] = None): super().__init__(str(error)) self.error = error self.retry_after = retry_after def _retry_after_seconds(headers) -> Optional[float]: raw = headers.get("Retry-After") if headers else None try: return max(0.0, min(float(raw), _MAX_RETRY_AFTER)) if raw is not None else None except (TypeError, ValueError): return None class AutomationClient: def __init__( self, base_url: str, token: str, *, timeout: float = 15.0, max_retries: int = 2, backoff_base: float = 0.5, backoff_cap: float = 8.0, total_deadline: float = 45.0, allow_insecure_loopback: bool = True, user_agent: Optional[str] = None, _sleep: Callable[[float], None] = time.sleep, ): try: parts = urllib.parse.urlsplit(str(base_url or "").strip()) host = (parts.hostname or "").lower() except ValueError as exc: raise AutomationConfigError("base_url must be a full address such as https://license.example.com") from exc if parts.scheme not in ("https", "http") or not host: raise AutomationConfigError("base_url must be a full address such as https://license.example.com") if parts.username or parts.password or parts.query or parts.fragment: raise AutomationConfigError("base_url must not contain credentials, a query string or a fragment") if parts.scheme == "http" and not (allow_insecure_loopback and host in _LOOPBACK_HOSTS): raise AutomationConfigError("base_url must use https:// - an API token must never cross the network in the clear") if not isinstance(token, str) or not _TOKEN_RE.match(token): raise AutomationConfigError("token does not look like an API token (it starts with 'lk_'); create one under API tokens in your dashboard") if not (0 < timeout <= 120): raise AutomationConfigError("timeout must be between 0 and 120 seconds") self._base = f"{parts.scheme}://{parts.netloc}{parts.path.rstrip('/')}" self._token = token self._timeout = float(timeout) self._max_retries = max(0, min(int(max_retries), 5)) self._backoff_base = max(0.0, float(backoff_base)) self._backoff_cap = max(self._backoff_base, float(backoff_cap)) self._total_deadline = max(1.0, float(total_deadline)) self._sleep = _sleep self._user_agent = user_agent or f"foliox-automation-python/{__version__}" context = ssl.create_default_context() context.minimum_version = ssl.TLSVersion.TLSv1_2 self._opener = urllib.request.build_opener(urllib.request.HTTPSHandler(context=context), _NoRedirect()) def __repr__(self) -> str: # never reveals the token return f"AutomationClient(base_url={self._base!r}, token='{self._token[:5]}...')" # -- API ---------------------------------------------------------------------------------------------------- def products(self) -> List[Dict[str, Any]]: """Your products: ``[{'id', 'name', 'slug', 'description', 'default_duration_days'}]``. Needs ``products:read``.""" return self._request("GET", "/products")["products"] def create_keys( self, product: str, *, count: int = 1, duration_days: Optional[int] = None, expires_at: Optional[str] = None, customer_email: Optional[str] = None, notes: Optional[str] = None, key_type: Optional[str] = None, ) -> List[Dict[str, Any]]: """Generates ``count`` (1-100) keys for ``product`` (its product ID). Needs ``keys:create``. ``duration_days`` runs from the first activation; ``expires_at`` (ISO 8601, UTC) is a fixed date. With a ``customer_email`` the service e-mails the key to the buyer (on plans that include customer e-mails). ``key_type`` is ``"device"`` (the default: each key locks to the first device that activates it) or ``"public"`` (no device binding: any number of people can use the same key at once). """ if not isinstance(product, str) or not _PRODUCT_RE.match(product): raise AutomationConfigError("product must be a product ID (lowercase letters, digits and hyphens)") if not isinstance(count, int) or isinstance(count, bool) or not 1 <= count <= 100: raise AutomationConfigError("count must be a whole number from 1 to 100") if key_type is not None and key_type not in ("device", "public"): raise AutomationConfigError("key_type must be 'device' or 'public'") body: Dict[str, Any] = {"product": product, "count": count} if duration_days is not None: body["durationDays"] = duration_days if expires_at is not None: body["expiresAt"] = expires_at if customer_email is not None: body["customerEmail"] = customer_email if notes is not None: body["notes"] = notes if key_type is not None: body["keyType"] = key_type return self._request("POST", "/keys", body, idempotent=False)["keys"] def get_key(self, key: str) -> Dict[str, Any]: """One key's status, expiry and device binding. Needs ``keys:read``.""" return self._request("GET", f"/keys/{self._key_path(key)}")["key"] def revoke_key(self, key: str) -> Dict[str, Any]: """Revokes a key: it stops validating immediately. Needs ``keys:revoke``. Safe to repeat.""" return self._request("POST", f"/keys/{self._key_path(key)}/revoke", {})["key"] def restore_key(self, key: str) -> Dict[str, Any]: """Undoes a revoke. Needs ``keys:revoke``. Safe to repeat.""" return self._request("POST", f"/keys/{self._key_path(key)}/restore", {})["key"] def reset_device(self, key: str) -> Dict[str, Any]: """Unbinds the key from its device so the customer can activate on a new one. Needs ``keys:reset``. Safe to repeat.""" return self._request("POST", f"/keys/{self._key_path(key)}/reset-device", {})["key"] # -- internals ---------------------------------------------------------------------------------------------- @staticmethod def _key_path(key: str) -> str: clean = re.sub(r"[^0-9A-Z-]", "", str(key or "").strip().upper()) if not _KEY_RE.match(clean) or not _KEY_SHAPE.match(clean): raise AutomationConfigError("that is not a license key (expected XXXXX-XXXXX-XXXXX-XXXXX-X)") return urllib.parse.quote(clean, safe="") def _request(self, method: str, path: str, body: Optional[dict] = None, *, idempotent: bool = True) -> dict: deadline = time.monotonic() + self._total_deadline last: Optional[_Retryable] = None for attempt in range(self._max_retries + 1): remaining = deadline - time.monotonic() if remaining <= 0: break try: return self._once(method, path, body, min(self._timeout, remaining), idempotent) except _Retryable as exc: last = exc if attempt == self._max_retries: break wait = exc.retry_after if exc.retry_after is not None else random.uniform(0.0, min(self._backoff_cap, self._backoff_base * (2**attempt))) if time.monotonic() + wait >= deadline: break self._sleep(wait) raise last.error if last else NetworkError("no time left to reach the license server") def _once(self, method: str, path: str, body: Optional[dict], timeout: float, idempotent: bool) -> dict: data = json.dumps(body, separators=(",", ":")).encode("utf-8") if body is not None else None headers = {"Authorization": f"Bearer {self._token}", "Accept": "application/json", "User-Agent": self._user_agent} if data is not None: headers["Content-Type"] = "application/json" request = urllib.request.Request(f"{self._base}/api/automation/v1{path}", data=data, method=method, headers=headers) try: with self._opener.open(request, timeout=timeout) as response: status, response_headers, raw = response.status, response.headers, response.read(_MAX_BODY + 1) except urllib.error.HTTPError as err: try: status, response_headers, raw = err.code, err.headers, err.read(_MAX_BODY + 1) except (OSError, http.client.HTTPException): status, response_headers, raw = err.code, err.headers, b"" finally: err.close() except ssl.SSLCertVerificationError as exc: raise NetworkError("TLS certificate verification failed - refusing to send the token") from exc except urllib.error.URLError as exc: reason = exc.reason if isinstance(reason, ssl.SSLCertVerificationError): raise NetworkError("TLS certificate verification failed - refusing to send the token") from exc # a refused connection or failed lookup means the request never left this machine: always safe to retry never_sent = isinstance(reason, (ConnectionRefusedError, socket.gaierror)) error = NetworkError(f"could not reach the license server ({reason.__class__.__name__})") if idempotent or never_sent: raise _Retryable(error) from exc raise error from exc except (socket.timeout, TimeoutError, ConnectionError, ssl.SSLError, OSError, http.client.HTTPException) as exc: error = NetworkError(f"could not reach the license server ({exc.__class__.__name__})") if idempotent: raise _Retryable(error) from exc raise error from exc if len(raw) > _MAX_BODY: raise NetworkError("the answer from the server was unexpectedly large") try: parsed = json.loads(raw.decode("utf-8")) if raw else {} except (UnicodeDecodeError, ValueError): parsed = None if 200 <= status < 300: if not isinstance(parsed, dict): raise NetworkError("the answer from the server was not JSON - is base_url your license server?") return parsed if 300 <= status < 400: raise NetworkError("the server redirected the request - check that base_url is exactly your license server's address (the token was not forwarded)", status=status) info = (parsed or {}).get("error", {}) if isinstance(parsed, dict) else {} message = str(info.get("message") or f"HTTP {status}") common = {"status": status, "code": info.get("code"), "details": info.get("details"), "fields": info.get("fields")} if status == 401: raise AuthenticationError(message, **common) if status == 402: raise PlanLimitError(message, **common) if status == 403: raise PermissionDeniedError(message, **common) if status == 404: raise NotFoundError(message, **common) if status == 429: hint = _retry_after_seconds(response_headers) error = RateLimitedError(message, retry_after=hint, **common) raise _Retryable(error, hint) # rejected before it was processed, so even creating keys may be retried if status >= 500: error = ServerError(message, **common) if idempotent and status in (500, 502, 503, 504): raise _Retryable(error) raise error if status == 400: raise ValidationFailedError(message, **common) raise AutomationError(message, **common)