Webhooks

POST /pikoman/webhook HTTP/1.1
Content-Type: application/json
User-Agent: Pikoman-Webhooks/1 (+https://pikoman.se/docs/webhooks/)
webhook-id: msg_4f1c9e0b7a2d48e6b3c5a9d1e7f20864
webhook-timestamp: 1761400000
webhook-signature: v1a,Q2JpZ3JhbnQ…
pikoman-key-id: wh_9a3f1c0e5b7d2468

{"time_ms":1761400000123,"app_id":"app-yourapp","events":[
  {"name":"channel_occupied","channel":"presence-room"},
  {"name":"member_added","channel":"presence-room","user_id":"42","connection":"c_8f3a1c9e2b7d4a6051e0c3b9"},
  {"name":"client_event","channel":"chat-1","event":"client-message","data":{"text":"hi"},"connection":"c_8f3a…","user_id":"42"}
]}

GET https://next-api.pikoman.se/.well-known/pikoman-webhook-key/app-yourapp

{"app_id":"app-yourapp","keys":[{"kid":"wh_9a3f1c0e5b7d2468","alg":"Ed25519","public_key":"<base64, 32 bytes>","whpk":"whpk_<the same>"}]}

import { createPublicKey, verify } from "node:crypto";

// Your app's public key from GET https://next-api.pikoman.se/.well-known/pikoman-webhook-key/<your app id>
// ("public_key", base64). Not a secret: it can only check, never sign.
const PIKOMAN_KEY = createPublicKey({
  key: Buffer.concat([
    Buffer.from("302a300506032b6570032100", "hex"), // Ed25519 public key header (SPKI)
    Buffer.from(process.env.PIKOMAN_WEBHOOK_PUBLIC_KEY, "base64"),
  ]),
  format: "der",
  type: "spki",
});
const seen = new Map(); // webhook-id -> when, to drop a batch you already handled

/** The events of a webhook request, or null when it is not from Pikoman. `body` is the raw bytes. */
export function pikomanEvents(headers, body) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatures = (headers["webhook-signature"] ?? "").split(" ");
  if (!id || !timestamp) return null;
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return null; // older than 5 minutes
  const signed = Buffer.concat([Buffer.from(`${id}.${timestamp}.`), body]);
  const valid = signatures.some((s) =>
    s.startsWith("v1a,") && verify(null, signed, PIKOMAN_KEY, Buffer.from(s.slice(4), "base64")));
  if (!valid || seen.has(id)) return null;
  const batch = JSON.parse(body);
  if (batch.app_id !== process.env.PIKOMAN_APP) return null; // for another app: not yours
  seen.set(id, Date.now());
  return batch.events;
}

// Express: the raw body, not parsed JSON, since the signature covers the exact bytes.
app.post("/pikoman/webhook", express.raw({ type: "application/json" }), (req, res) => {
  const events = pikomanEvents(req.headers, req.body);
  if (!events) return res.sendStatus(401);
  res.sendStatus(204); // answer at once; do the work after
  for (const event of events) console.log(event.name, event.channel);
});

<?php
// Your app's public key from https://next-api.pikoman.se/.well-known/pikoman-webhook-key/<your app id>, base64.
$publicKey = base64_decode(getenv('PIKOMAN_WEBHOOK_PUBLIC_KEY'));

$body = file_get_contents('php://input');            // the raw body
$id = $_SERVER['HTTP_WEBHOOK_ID'] ?? '';
$timestamp = $_SERVER['HTTP_WEBHOOK_TIMESTAMP'] ?? '';
$signatures = explode(' ', $_SERVER['HTTP_WEBHOOK_SIGNATURE'] ?? '');

$valid = false;
if ($id !== '' && abs(time() - (int) $timestamp) <= 300) {
    foreach ($signatures as $signature) {
        if (str_starts_with($signature, 'v1a,')) {
            $raw = base64_decode(substr($signature, 4), true);
            if ($raw !== false && strlen($raw) === SODIUM_CRYPTO_SIGN_BYTES
                && sodium_crypto_sign_verify_detached($raw, "$id.$timestamp.$body", $publicKey)) {
                $valid = true;
            }
        }
    }
}
$batch = $valid ? json_decode($body, true) : null;
if (!$valid || ($batch['app_id'] ?? '') !== getenv('PIKOMAN_APP')) {
    http_response_code(401);
    exit;
}
// Drop an id you have already handled (a retry), e.g. in a table with a unique key.
http_response_code(204);
foreach ($batch['events'] as $event) {
    error_log($event['name'] . ' ' . $event['channel']);
}

# pip install cryptography
import base64, json, os, time
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.exceptions import InvalidSignature

PIKOMAN_KEY = Ed25519PublicKey.from_public_bytes(
    base64.b64decode(os.environ["PIKOMAN_WEBHOOK_PUBLIC_KEY"]))

def pikoman_events(headers, body: bytes):
    """The events of a webhook request, or None when it is not from Pikoman."""
    msg_id = headers.get("webhook-id", "")
    timestamp = headers.get("webhook-timestamp", "0")
    if not msg_id or abs(time.time() - int(timestamp)) > 300:
        return None
    signed = f"{msg_id}.{timestamp}.".encode() + body
    for signature in headers.get("webhook-signature", "").split(" "):
        if signature.startswith("v1a,"):
            try:
                PIKOMAN_KEY.verify(base64.b64decode(signature[4:]), signed)
                batch = json.loads(body)
                if batch.get("app_id") != os.environ["PIKOMAN_APP"]:
                    return None  # for another app: not yours
                return batch["events"]
            except (InvalidSignature, ValueError):
                pass
    return None

# Flask: request.get_data() is the raw body.
# events = pikoman_events(request.headers, request.get_data())

// ed25519-dalek = "2", base64 = "0.22"
use base64::{Engine, engine::general_purpose::STANDARD};
use ed25519_dalek::{Signature, VerifyingKey};

/// Whether a webhook request is from Pikoman: `key` is your app's 32-byte public key from
/// `/.well-known/pikoman-webhook-key/<your app id>`, `now` Unix seconds, `body` the raw bytes.
/// Parse the body after, and check that its `app_id` is your app's.
pub fn from_pikoman(key: &VerifyingKey, id: &str, timestamp: &str, signatures: &str,
                    body: &[u8], now: u64) -> bool {
    let Ok(sent) = timestamp.parse::<u64>() else { return false };
    if id.is_empty() || now.abs_diff(sent) > 300 {
        return false;
    }
    let mut signed = format!("{id}.{timestamp}.").into_bytes();
    signed.extend_from_slice(body);
    signatures.split(' ').filter_map(|s| s.strip_prefix("v1a,")).any(|s| {
        STANDARD.decode(s).ok()
            .and_then(|bytes| Signature::from_slice(&bytes).ok())
            .is_some_and(|signature| key.verify_strict(&signed, &signature).is_ok())
    })
}