Get notified with webhooks

curl -X POST https://yungle.co/api/v1/webhooks \
  -H "Authorization: Bearer $YUNGLE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://example.com/hooks/yungle",
    "events": ["transfer.ready", "transfer.downloaded"],
    "description": "Delivery notifications"
  }'

import { YungleClient } from 'yungle-client';

const yungle = new YungleClient({ apiKey: process.env.YUNGLE_API_KEY });

const { webhook, secret } = await yungle.createWebhook({
  url: 'https://example.com/hooks/yungle',
  events: ['transfer.ready', 'transfer.downloaded'],
  description: 'Delivery notifications',
});

from yungle import Yungle

yungle = Yungle()

created = yungle.create_webhook(
    "https://example.com/hooks/yungle",
    ["transfer.ready", "transfer.downloaded"],
    description="Delivery notifications",
)
secret = created["secret"]

import express from 'express';
import { verifyWebhook } from 'yungle-client';

const app = express();

// express.raw keeps the exact bytes; the signature covers them.
app.post('/hooks/yungle', express.raw({ type: 'application/json' }), async (req, res) => {
  const raw = req.body.toString('utf8');
  const ok = await verifyWebhook(raw, req.get('yungle-signature') ?? '', process.env.YUNGLE_WEBHOOK_SECRET);
  if (!ok) return res.status(400).end();
  const event = JSON.parse(raw);
  res.status(200).end();
  // handle event…
});

import json
import os

from flask import Flask, request
from yungle import verify_webhook

app = Flask(__name__)

@app.post("/hooks/yungle")
def yungle_hook():
    raw = request.get_data()  # the exact bytes, before any parsing
    if not verify_webhook(raw, request.headers.get("Yungle-Signature", ""), os.environ["YUNGLE_WEBHOOK_SECRET"]):
        return "", 400
    event = json.loads(raw)
    # handle event…
    return "", 200

import { createHmac, timingSafeEqual } from 'node:crypto';

export function verifyYungle(rawBody, header, secret, toleranceSeconds = 300) {
  const parts = Object.fromEntries((header ?? '').split(',').map((p) => p.split('=')));
  const t = Number(parts.t);
  if (!Number.isInteger(t) || Math.abs(Date.now() / 1000 - t) > toleranceSeconds || !parts.v1) return false;
  const expected = createHmac('sha256', secret).update(`${t}.${rawBody}`).digest();
  const given = Buffer.from(parts.v1, 'hex');
  return given.length === expected.length && timingSafeEqual(given, expected);
}

import hashlib
import hmac
import time

def verify_yungle(raw_body: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
    try:
        parts = dict(p.split("=", 1) for p in header.split(","))
        t, given = int(parts["t"]), parts["v1"]
    except (KeyError, ValueError):
        return False
    if abs(time.time() - t) > tolerance:
        return False
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, given)

curl -X POST https://yungle.co/api/v1/webhooks/01JB9D1A2B3C4D5E6F7G8H9J0K/test \
  -H "Authorization: Bearer $YUNGLE_API_KEY"

const { deliveryId } = await yungle.testWebhook('01JB9D1A2B3C4D5E6F7G8H9J0K');

delivery_id = yungle.test_webhook("01JB9D1A2B3C4D5E6F7G8H9J0K")["deliveryId"]