Sign requests

"""Olympex REST API: request signing (signature v2) and a minimal client (Python 3.8+)."""
import base64
import hashlib
import hmac
import json
import os
import re
import secrets
import time
from urllib.parse import quote, unquote_plus, urlsplit

import requests

OLYMPEX_BASE_URL = "https://api-rest.olympex.io/api/v1"
SIGNATURE_VERSION = "OLPX-HMAC-SHA256-V2"
_MAX_SAFE_INTEGER = 2**53 - 1  # the server parses JSON numbers as JavaScript doubles


def _check_signable(value, where="body"):
    """Reject values whose JSON text would differ between Python and the server's JavaScript canonicalizer."""
    if value is None or isinstance(value, (bool, str)):
        return
    if isinstance(value, int):
        if abs(value) > _MAX_SAFE_INTEGER:
            raise ValueError(f"{where}: integer outside +/-(2**53 - 1); send it as a string")
        return
    if isinstance(value, float):
        if value != value or value in (float("inf"), float("-inf")):
            raise ValueError(f"{where}: NaN and Infinity are not valid JSON")
        return
    if isinstance(value, dict):
        for key, item in value.items():
            # JavaScript orders integer-like keys first, sorts by UTF-16 code units and drops "__proto__".
            if not isinstance(key, str) or not key.isascii() or key.isdigit() or key == "__proto__":
                raise ValueError(f"{where}: object keys must be non-numeric ASCII strings, got {key!r}")
            _check_signable(item, f"{where}.{key}")
        return
    if isinstance(value, (list, tuple)):
        for index, item in enumerate(value):
            _check_signable(item, f"{where}[{index}]")
        return
    raise TypeError(f"{where}: unsupported type {type(value).__name__}")


def _js_number(value):
    """Format a float the way JavaScript's JSON.stringify does (ECMAScript Number::toString): 1.0 is "1", 1e21 is "1e+21"."""
    if value == 0:
        return "0"  # also -0.0
    mantissa, _, exponent = repr(abs(value)).partition("e")  # repr gives the shortest round-trip digits, like JavaScript
    whole, _, fraction = mantissa.partition(".")
    raw = whole + fraction
    digits = raw.lstrip("0")
    point = len(whole) + int(exponent or 0) - (len(raw) - len(digits))  # value = 0.<digits> * 10**point
    digits = digits.rstrip("0")
    if len(digits) <= point <= 21:
        text = digits + "0" * (point - len(digits))
    elif 0 < point <= 21:
        text = digits[:point] + "." + digits[point:]
    elif -6 < point <= 0:
        text = "0." + "0" * -point + digits
    else:
        exp = point - 1
        text = digits[0] + ("." + digits[1:] if len(digits) > 1 else "") + ("e+" if exp >= 0 else "e-") + str(abs(exp))
    return ("-" if value < 0 else "") + text


def _canonical(value):
    if value is None:
        return "null"
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, int):
        return str(value)
    if isinstance(value, float):
        return _js_number(value)
    if isinstance(value, str):
        return json.dumps(value, ensure_ascii=False)
    if isinstance(value, dict):
        return "{" + ",".join(f"{json.dumps(key, ensure_ascii=False)}:{_canonical(value[key])}" for key in sorted(value)) + "}"
    return "[" + ",".join(_canonical(item) for item in value) + "]"


def canonical_body(body):
    """The exact string the server hashes: keys sorted at every level, no whitespace, raw UTF-8,
    numbers formatted as JavaScript formats them. Empty when there is no body."""
    if body is None:
        return ""
    _check_signable(body)
    return _canonical(body)


def canonical_query(query):
    """The query the server signs: parameters decoded ("+" is a space), sorted by key and then by value,
    re-encoded per RFC 3986 and joined with "&". Empty when there is no query."""
    pairs = []
    for part in query.split("&"):
        if part:
            key, _, value = part.partition("=")
            pairs.append((unquote_plus(key), unquote_plus(value)))
    # JavaScript compares strings by UTF-16 code units, which is the order of their UTF-16-BE bytes.
    pairs.sort(key=lambda pair: (pair[0].encode("utf-16-be"), pair[1].encode("utf-16-be")))
    return "&".join(f"{quote(key, safe='')}={quote(value, safe='')}" for key, value in pairs)


def sign_request(method, url, body, credentials, timestamp=None, nonce=None):
    """Return the canonical body and the signed headers for one request. `url` is the full URL you send,
    query included. Pass body=None for GET and DELETE. Sign every attempt again: nonces are single-use."""
    method = method.upper()
    parts = urlsplit(url)
    path, query = parts.path, canonical_query(parts.query)
    body_string = canonical_body(body)
    timestamp = str(int(time.time()) if timestamp is None else timestamp)
    nonce = secrets.token_hex(12) if nonce is None else nonce
    if not re.fullmatch(r"[0-9a-fA-F]{24}", nonce):
        raise ValueError("nonce must be 24 hexadecimal characters")
    digest = hashlib.sha256(body_string.encode("utf-8")).digest()
    body_hash = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
    message = "\n".join([SIGNATURE_VERSION, timestamp, nonce, method, path, query, body_hash])
    # The HMAC key is the secret key string itself (UTF-8 bytes), not its base64url-decoded bytes.
    signature = hmac.new(credentials["secret_key"].encode("utf-8"), message.encode("utf-8"), hashlib.sha256).hexdigest()
    value_info = f"{timestamp}\n{nonce}\n{body_hash}"
    return {
        "method": method,
        "path": path,
        "query": query,
        "body_string": body_string,
        "body_hash": body_hash,
        "timestamp": timestamp,
        "nonce": nonce,
        "message": message,
        "headers": {
            "content-type": "application/json",
            "x-api-key-id": credentials["api_key_id"],
            "x-value-info": base64.b64encode(value_info.encode("utf-8")).decode("ascii"),
            "x-passphrase": credentials["passphrase"],
            "x-signature": signature,
        },
    }


class OlympexApiError(Exception):
    def __init__(self, status, code, message, details=None, request_id=None):
        super().__init__(f"{status} {code}: {message}")
        self.status = status
        self.code = code
        self.message = message
        self.details = details or []
        self.request_id = request_id


def credentials_from_env():
    try:
        return {
            "api_key_id": os.environ["OLYMPEX_API_KEY_ID"],
            "secret_key": os.environ["OLYMPEX_SECRET_KEY"],
            "passphrase": os.environ["OLYMPEX_PASSPHRASE"],
        }
    except KeyError as missing:
        raise RuntimeError(f"Set {missing.args[0]} (and the other OLYMPEX_* variables).") from None


def olympex_request(method, path, body=None, credentials=None, timeout=35):
    """Sign and send one request, then return `data` or raise OlympexApiError.

    POST and PATCH need a body; GET and DELETE take none and are signed over the empty string.
    `path` may carry a query string, for example "/tokens?chainId=137". The signature covers the method, the path and the query.
    """
    method = method.upper()
    has_body = method in ("POST", "PATCH")
    if has_body and body is None:
        raise ValueError(f"{method} {path} needs a body")
    if not has_body and body is not None:
        raise ValueError(f"{method} requests take no body")
    url = OLYMPEX_BASE_URL + path
    signed = sign_request(method, url, body, credentials or credentials_from_env())
    response = requests.request(
        method,
        url,
        # bytes: http.client sends a str body as Latin-1
        data=signed["body_string"].encode("utf-8") if has_body else None,
        headers=signed["headers"],
        timeout=timeout,
    )
    try:
        payload = response.json()
    except ValueError:
        payload = None
    if not isinstance(payload, dict):
        payload = {}
    if response.ok and payload.get("success") is True:
        return payload.get("data")
    error = payload.get("error")
    if isinstance(error, dict):
        raise OlympexApiError(
            response.status_code,
            error.get("code"),
            error.get("message"),
            error.get("details"),
            (payload.get("meta") or {}).get("requestId"),
        )
    # Gateway responses (401, 403, 429, 500, 503) carry only {"message": ...}.
    raise OlympexApiError(response.status_code, f"HTTP_{response.status_code}", payload.get("message") or response.text)

from sign_request import olympex_request

# GET: no body. The signer signs the empty string.
chain_ids = olympex_request("GET", "/chains")["chainIds"]
# POST: the signer canonicalizes and signs the body.
supported = olympex_request("POST", "/support-chain", {"chainId": 137})
print(137 in chain_ids, supported)  # True True

# Olympex REST API: sign (signature v2) and send one request with openssl and curl.
# Works in bash 3.2+ and zsh, with OpenSSL 1.1+/3.x or LibreSSL.
# Needs OLYMPEX_API_KEY_ID, OLYMPEX_SECRET_KEY and OLYMPEX_PASSPHRASE in the environment.
OLYMPEX_BASE_URL="https://api-rest.olympex.io/api/v1"

# Usage: olympex_request METHOD PATH [BODY]
#   olympex_request GET '/tokens?chainId=137'
#   olympex_request POST /support-chain '{"chainId":137}'
# POST and PATCH need a body that is already canonical JSON (keys sorted at every level, no whitespace):
# it is hashed byte for byte and the server hashes its own canonical form.
# GET and DELETE take no body and are signed over the empty string.
# The signature also covers the method, the path (with its /api/v1 prefix) and the query. The query is signed as
# written, so write it in canonical form: parameters sorted by key and then by value, each percent-encoded per RFC 3986.
# OLYMPEX_TIMESTAMP and OLYMPEX_NONCE, if set, replace the current time and the random nonce. They exist only to
# reproduce the known-answer vectors: never export them, or every later request fails with 403 (stale time, reused nonce).
olympex_request() {
  local method="$1" endpoint="$2" body="${3-}" query=""
  local ts nonce body_hash sign_path value_info message signature
  : "${OLYMPEX_API_KEY_ID:?set OLYMPEX_API_KEY_ID}" "${OLYMPEX_SECRET_KEY:?set OLYMPEX_SECRET_KEY}" "${OLYMPEX_PASSPHRASE:?set OLYMPEX_PASSPHRASE}"
  case "$method" in
    POST|PATCH) [ -n "$body" ] || { printf 'olympex_request: %s needs a body\n' "$method" >&2; return 2; } ;;
    GET|DELETE) [ -z "$body" ] || { printf 'olympex_request: %s takes no body\n' "$method" >&2; return 2; } ;;
    *) printf 'olympex_request: unsupported method %s\n' "$method" >&2; return 2 ;;
  esac
  # Never name a variable `path`: zsh ties it to PATH.
  sign_path="/${OLYMPEX_BASE_URL#https://*/}${endpoint%%\?*}"
  case "$endpoint" in *\?*) query="${endpoint#*\?}" ;; esac
  ts="${OLYMPEX_TIMESTAMP:-$(date +%s)}"
  nonce="${OLYMPEX_NONCE:-$(openssl rand -hex 12)}"
  body_hash="$(printf '%s' "$body" | openssl dgst -sha256 -binary | openssl base64 -A | tr '+/' '-_' | tr -d '=')"
  value_info="$(printf '%s\n%s\n%s' "$ts" "$nonce" "$body_hash" | openssl base64 -A)"
  message="$(printf 'OLPX-HMAC-SHA256-V2\n%s\n%s\n%s\n%s\n%s\n%s' "$ts" "$nonce" "$method" "$sign_path" "$query" "$body_hash")"
  signature="$(printf '%s' "$message" | openssl dgst -sha256 -hmac "$OLYMPEX_SECRET_KEY" -binary | od -An -v -tx1 | tr -d ' \n')"
  curl -sS --max-time 35 -X "$method" "$OLYMPEX_BASE_URL$endpoint" \
    -H "content-type: application/json" \
    -H "x-api-key-id: $OLYMPEX_API_KEY_ID" \
    -H "x-value-info: $value_info" \
    -H "x-passphrase: $OLYMPEX_PASSPHRASE" \
    -H "x-signature: $signature" \
    ${body:+--data-raw} ${body:+"$body"}
}