Sign a request

{timestamp}
{METHOD}
{path-with-query}
{sha256-hex-of-raw-body}
{idempotency-key or empty string}

x-api-signature = base64( Ed25519-sign(private key, canonical) )

import crypto from "node:crypto";
import fs from "node:fs";

// Your private key — generated by you, never sent to MegTrust.
const PRIVATE_KEY = crypto.createPrivateKey(fs.readFileSync("megtrust-private.pem"));
const KEY_ID = process.env.MEGTRUST_KEY_ID;
const BASE = "https://stagingvault.megtrust.com";

export async function call(method, path, body, idempotencyKey) {
  // A fresh signed query nonce distinguishes identical calls in the same second.
  const url = new URL(path, BASE);
  url.searchParams.set("_request_id", crypto.randomUUID());
  path = url.pathname + url.search;
  const payload = body ? JSON.stringify(body) : "";
  const timestamp = Math.floor(Date.now() / 1000).toString();
  const bodyHash = crypto.createHash("sha256").update(payload).digest("hex");
  const idem = idempotencyKey ?? "";

  // 5 lines, in this exact order.
  const canonical = [timestamp, method.toUpperCase(), path, bodyHash, idem].join("\n");
  const signature = crypto.sign(null, Buffer.from(canonical), PRIVATE_KEY).toString("base64");

  const res = await fetch(BASE + path, {
    method,
    headers: {
      "content-type": "application/json",
      "x-api-key-id": KEY_ID,
      "x-api-timestamp": timestamp,
      "x-api-signature": signature,
      ...(idem ? { "idempotency-key": idem } : {}),
    },
    ...(payload ? { body: payload } : {}),
    redirect: "error",
    signal: AbortSignal.timeout(30_000),
  });
  return { status: res.status, body: await res.json() };
}

import base64, hashlib, json, os, time, uuid

import requests  # pip install requests cryptography
from cryptography.hazmat.primitives.serialization import load_pem_private_key

BASE = os.environ.get("MEGTRUST_BASE_URL", "https://stagingvault.megtrust.com")
KEY_ID = os.environ["MEGTRUST_KEY_ID"]
with open(os.environ.get("MEGTRUST_PRIVATE_KEY_PATH", "megtrust-private.pem"), "rb") as f:
    PRIVATE_KEY = load_pem_private_key(f.read(), password=None)


def call(method, path, body=None, idempotency_key=None):
    # A fresh signed query nonce on every attempt, retries included.
    path += ("&" if "?" in path else "?") + f"_request_id={uuid.uuid4()}"
    payload = b"" if body is None else json.dumps(body).encode()
    timestamp = str(int(time.time()))

    # 5 lines, in this exact order.
    canonical = "\n".join([
        timestamp,
        method.upper(),
        path,
        hashlib.sha256(payload).hexdigest(),
        idempotency_key or "",
    ])
    signature = base64.b64encode(PRIVATE_KEY.sign(canonical.encode())).decode()

    headers = {
        "content-type": "application/json",
        "x-api-key-id": KEY_ID,
        "x-api-timestamp": timestamp,
        "x-api-signature": signature,
    }
    if idempotency_key:
        headers["idempotency-key"] = idempotency_key
    res = requests.request(method, BASE + path, data=payload or None, headers=headers,
                           timeout=30, allow_redirects=False)
    return res.status_code, res.json()