Withdrawal examples

export STABYL_BASE_URL="https://api-staging.stabyl.com/v1"
export STABYL_API_KEY="sb_test..."
export STABYL_SIGNING_SECRET="sb_testsig1_..."

const {
  createHash,
  createPrivateKey,
  randomUUID,
  sign,
} = require("node:crypto");

const baseUrl = requiredEnv("STABYL_BASE_URL").replace(/\/+$/, "");
const apiKey = requiredEnv("STABYL_API_KEY");
const signingKey = signingKeyFromSecret(
  requiredEnv("STABYL_SIGNING_SECRET"),
);

function requiredEnv(name) {
  const value = process.env[name];
  if (!value) throw new Error(`${name} is required`);
  return value;
}

function signingKeyFromSecret(secret) {
  const marker = "sig1_";
  const markerIndex = secret.indexOf(marker);
  if (markerIndex < 0) throw new Error("unsupported signing secret");

  const seed = Buffer.from(
    secret.slice(markerIndex + marker.length),
    "base64url",
  );
  if (seed.length !== 32) throw new Error("invalid signing secret");

  // RFC 8410 PKCS#8 prefix for a raw 32-byte Ed25519 private seed.
  const pkcs8Prefix = Buffer.from(
    "302e020100300506032b657004220420",
    "hex",
  );
  return createPrivateKey({
    key: Buffer.concat([pkcs8Prefix, seed]),
    format: "der",
    type: "pkcs8",
  });
}

async function responseJson(response) {
  const text = await response.text();
  const payload = text ? JSON.parse(text) : null;
  if (!response.ok) {
    throw new Error(`Stabyl returned ${response.status}: ${text}`);
  }
  return payload;
}

async function signedPost(path, payload) {
  // Serialize once. Hash, sign, and send these exact bytes.
  const rawBody = Buffer.from(JSON.stringify(payload), "utf8");
  const url = `${baseUrl}${path}`;
  const parsedUrl = new URL(url);
  const target = `${parsedUrl.pathname}${parsedUrl.search}`;
  const timestamp = Date.now().toString();
  const nonce = randomUUID();
  const bodyHash = createHash("sha256").update(rawBody).digest("hex");
  const canonical = [
    "STABYL-SIGNATURE-V1",
    timestamp,
    nonce,
    "POST",
    target,
    bodyHash,
  ].join("\n");
  const signature = sign(null, Buffer.from(canonical, "utf8"), signingKey)
    .toString("base64url");

  return responseJson(await fetch(url, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "X-Api-Key": apiKey,
      "X-Stabyl-Timestamp": timestamp,
      "X-Stabyl-Nonce": nonce,
      "X-Stabyl-Signature": signature,
    },
    body: rawBody,
  }));
}

async function main() {
  const intent = {
    currency: "NGN",
    items: [{
      client_item_id: "beneficiary-a",
      amount: "25000.00",
      destination: {
        type: "bank_account",
        bank_code: "044",
        bank_name: "Access Bank",
        account_number: "0123456789",
        account_name: "Example Recipient",
      },
      description: "Supplier settlement",
    }],
  };

  const quote = await signedPost(
    "/partner/wallets/withdrawal/quotes",
    intent,
  );
  const available = quote.data.quotes.find(
    (candidate) => candidate.status === "available",
  );
  if (!available) throw new Error("no available withdrawal route");
  console.log("quote:", available);

  // Generate this once for the business intent. Preserve it and the unchanged
  // submitBody if an uncertain HTTP result must be retried.
  const submitBody = {
    ...intent,
    idempotency_key: randomUUID(),
  };
  const submitted = await signedPost(
    "/partner/wallets/withdrawals",
    submitBody,
  );
  console.log("submitted withdrawal:", submitted.data.id);
  console.log(
    "store this ID and wait for withdrawal.success or withdrawal.failed",
  );
}

main().catch((error) => {
  console.error(error);
  process.exitCode = 1;
});

python -m pip install "cryptography>=42" "requests>=2.31"

import base64
import hashlib
import json
import os
import time
import uuid
from urllib.parse import urlsplit

import requests
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey


BASE_URL = os.environ["STABYL_BASE_URL"].rstrip("/")
API_KEY = os.environ["STABYL_API_KEY"]


def signing_key_from_secret(secret: str) -> Ed25519PrivateKey:
    marker = "sig1_"
    marker_index = secret.find(marker)
    if marker_index < 0:
        raise ValueError("unsupported signing secret")

    encoded = secret[marker_index + len(marker):]
    padded = encoded + "=" * (-len(encoded) % 4)
    seed = base64.urlsafe_b64decode(padded)
    if len(seed) != 32:
        raise ValueError("invalid signing secret")
    return Ed25519PrivateKey.from_private_bytes(seed)


SIGNING_KEY = signing_key_from_secret(os.environ["STABYL_SIGNING_SECRET"])


def response_json(response: requests.Response) -> dict:
    if not response.ok:
        raise RuntimeError(
            f"Stabyl returned {response.status_code}: {response.text}"
        )
    return response.json()


def signed_post(path: str, payload: dict) -> dict:
    # Serialize once. Hash, sign, and send these exact bytes.
    raw_body = json.dumps(
        payload,
        separators=(",", ":"),
        ensure_ascii=False,
    ).encode("utf-8")
    url = f"{BASE_URL}{path}"
    parsed_url = urlsplit(url)
    target = parsed_url.path
    if parsed_url.query:
        target += f"?{parsed_url.query}"

    timestamp = str(int(time.time() * 1000))
    nonce = str(uuid.uuid4())
    body_hash = hashlib.sha256(raw_body).hexdigest()
    canonical = "\n".join([
        "STABYL-SIGNATURE-V1",
        timestamp,
        nonce,
        "POST",
        target,
        body_hash,
    ]).encode("utf-8")
    signature = base64.urlsafe_b64encode(
        SIGNING_KEY.sign(canonical)
    ).rstrip(b"=").decode("ascii")

    response = requests.post(
        url,
        headers={
            "Content-Type": "application/json",
            "X-Api-Key": API_KEY,
            "X-Stabyl-Timestamp": timestamp,
            "X-Stabyl-Nonce": nonce,
            "X-Stabyl-Signature": signature,
        },
        data=raw_body,
        timeout=30,
    )
    return response_json(response)


def main() -> None:
    intent = {
        "currency": "NGN",
        "items": [{
            "client_item_id": "beneficiary-a",
            "amount": "25000.00",
            "destination": {
                "type": "bank_account",
                "bank_code": "044",
                "bank_name": "Access Bank",
                "account_number": "0123456789",
                "account_name": "Example Recipient",
            },
            "description": "Supplier settlement",
        }],
    }

    quote = signed_post("/partner/wallets/withdrawal/quotes", intent)
    available = next(
        (
            candidate
            for candidate in quote["data"]["quotes"]
            if candidate["status"] == "available"
        ),
        None,
    )
    if available is None:
        raise RuntimeError("no available withdrawal route")
    print("quote:", available)

    # Generate this once for the business intent. Preserve it and the unchanged
    # submit_body if an uncertain HTTP result must be retried.
    submit_body = {
        **intent,
        "idempotency_key": str(uuid.uuid4()),
    }
    submitted = signed_post("/partner/wallets/withdrawals", submit_body)
    transaction_id = submitted["data"]["id"]
    print("submitted withdrawal:", transaction_id)
    print(
        "store this ID and wait for withdrawal.success or withdrawal.failed"
    )


if __name__ == "__main__":
    main()