Guides / kosa8 Cloud / Webhook recipes

Webhook recipes

kosa8 Cloud can POST an event to your server when a sandbox fails, your spending cap is reached, and more. This page has receivers to copy in Node, Python and Go: each checks the signature and forwards sandbox.error and spending_cap.reached to a Slack channel.

Add an endpoint

In the dashboard, or with an admin key. The URL must be public https. The whsec_… signing secret is in the answer, once: store it.

curl -s -X POST https://api.kosa8.com/v1/webhooks \
  -H "Authorization: Bearer $KOSA8_API_KEY" -H 'Content-Type: application/json' \
  -d '{"url":"https://hooks.example.com/kosa8","events":["sandbox.error","spending_cap.reached"]}'
# {"webhook":{"id":"wh_…","url":"https://hooks.example.com/kosa8",…,"secret":"whsec_…"},…}

# later: send a webhook.test event, and see recent deliveries
curl -s -X POST https://api.kosa8.com/v1/webhooks/wh_…/test -H "Authorization: Bearer $KOSA8_API_KEY"
curl -s https://api.kosa8.com/v1/webhooks/wh_…/deliveries -H "Authorization: Bearer $KOSA8_API_KEY"

Leave out events to get all of them: sandbox.created, sandbox.deleted, sandbox.slept, sandbox.woke, sandbox.error, function.deployed, function.invocation.failed and spending_cap.reached (sent once a month).

What arrives

A POST with a JSON body and three headers:

Kosa8-Signature: t=1794400000,v1=5f0c…e81a
Kosa8-Event: sandbox.error
Kosa8-Delivery: whd_…

{"id":"evt_…","type":"sandbox.error","created_at":1794400000000,"org_id":"org_…",
 "data":{"id":"sbx_…","error":"…","during":"create"}}

Event	`data`
`sandbox.error`	`id`, `error`, `during` (what was being done, e.g. `create`)
`spending_cap.reached`	`cost_usd`, `spending_cap_usd`, `month` (`YYYY-MM`)

v1 is the hex HMAC-SHA256 of "<t>.<raw body>", keyed with the whole secret, whsec_ included. To check it:

Read the raw body bytes before parsing them; re-serialized JSON will not match.
Compute the HMAC and compare it to v1 in constant time.
Refuse a t more than five minutes from now, so a captured request cannot be replayed later.

Answer any 2xx to acknowledge. Anything else, a timeout over 10 seconds, or a redirect is retried after 1 min, 5 min, 30 min, 2 h and 8 h, then marked failed. A retry carries the same Kosa8-Delivery, so use it to skip one you have already handled.

Slack

Each receiver below posts to a Slack incoming webhook. Create one for the channel (Slack: Apps → Incoming Webhooks → Add to Slack) and run the receiver with both secrets in its environment:

$ export KOSA8_WEBHOOK_SECRET=whsec_…
$ export SLACK_WEBHOOK_URL=https://hooks.slack.com/services/…

Other events are acknowledged and not forwarded. If Slack is down, the receiver answers 502 and kosa8 Cloud tries again later.

Node.js

Node 18 or later, no packages:

// kosa8 webhook receiver: verify Kosa8-Signature, forward two events to Slack.
// KOSA8_WEBHOOK_SECRET=whsec_… SLACK_WEBHOOK_URL=https://hooks.slack.com/services/… node server.mjs
import crypto from "node:crypto";
import http from "node:http";

const SECRET = process.env.KOSA8_WEBHOOK_SECRET;
const SLACK = process.env.SLACK_WEBHOOK_URL;

// verifyKosa8 checks the HMAC over "<t>.<raw body>" and refuses a t more than
// toleranceS seconds away (a replay).
export function verifyKosa8(rawBody, header, secret, toleranceS = 300) {
  const parts = {};
  for (const kv of String(header || "").split(",")) {
    const i = kv.indexOf("=");
    if (i > 0) parts[kv.slice(0, i).trim()] = kv.slice(i + 1).trim();
  }
  const t = Number(parts.t);
  if (!Number.isInteger(t) || Math.abs(Date.now() / 1000 - t) > toleranceS) return false;
  const want = crypto.createHmac("sha256", secret).update(`${parts.t}.${rawBody}`).digest();
  const got = Buffer.from(parts.v1 || "", "hex");
  return got.length === want.length && crypto.timingSafeEqual(got, want);
}

function slackText(evt) {
  const d = evt.data || {};
  if (evt.type === "sandbox.error") {
    return `:warning: kosa8 sandbox ${d.id} failed during ${d.during}: ${d.error}`;
  }
  if (evt.type === "spending_cap.reached") {
    return `:money_with_wings: kosa8 Cloud reached its spending cap for ${d.month}: ` +
      `$${Number(d.cost_usd).toFixed(2)} of $${d.spending_cap_usd}. New work is refused until the cap is raised.`;
  }
  return null; // other events: acknowledged, not forwarded
}

const server = http.createServer((req, res) => {
  const chunks = [];
  req.on("data", (c) => chunks.push(c));
  req.on("end", async () => {
    const raw = Buffer.concat(chunks).toString("utf8"); // the exact bytes signed
    if (req.method !== "POST" || !verifyKosa8(raw, req.headers["kosa8-signature"], SECRET)) {
      res.writeHead(401).end();
      return;
    }
    const evt = JSON.parse(raw);
    const text = slackText(evt);
    if (text) {
      const r = await fetch(SLACK, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ text }) });
      if (!r.ok) { res.writeHead(502).end(); return; } // kosa8 retries later
    }
    res.writeHead(200).end("ok");
  });
});
server.listen(Number(process.env.PORT || 8080));

Python

Python 3.8 or later, standard library only:

# kosa8 webhook receiver: verify Kosa8-Signature, forward two events to Slack.
# KOSA8_WEBHOOK_SECRET=whsec_… SLACK_WEBHOOK_URL=https://hooks.slack.com/services/… python3 server.py
import hashlib
import hmac
import json
import os
import time
import urllib.request
from http.server import BaseHTTPRequestHandler, HTTPServer

SECRET = os.environ["KOSA8_WEBHOOK_SECRET"]
SLACK = os.environ["SLACK_WEBHOOK_URL"]


def verify_kosa8(raw: bytes, header: str, secret: str, tolerance_s: int = 300) -> bool:
    """Check the HMAC over b"<t>." + raw body; refuse a t more than tolerance_s away."""
    parts = dict(p.strip().split("=", 1) for p in (header or "").split(",") if "=" in p)
    ts = parts.get("t", "")
    if not ts.isdigit() or abs(time.time() - int(ts)) > tolerance_s:
        return False
    want = hmac.new(secret.encode(), ts.encode() + b"." + raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(want, parts.get("v1", ""))


def slack_text(evt: dict):
    d = evt.get("data") or {}
    if evt.get("type") == "sandbox.error":
        return f":warning: kosa8 sandbox {d.get('id')} failed during {d.get('during')}: {d.get('error')}"
    if evt.get("type") == "spending_cap.reached":
        return (f":money_with_wings: kosa8 Cloud reached its spending cap for {d.get('month')}: "
                f"${float(d.get('cost_usd', 0)):.2f} of ${d.get('spending_cap_usd')}. New work is refused until the cap is raised.")
    return None  # other events: acknowledged, not forwarded


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        raw = self.rfile.read(int(self.headers.get("Content-Length", 0)))  # the exact bytes signed
        if not verify_kosa8(raw, self.headers.get("Kosa8-Signature"), SECRET):
            self.send_response(401)
            self.end_headers()
            return
        text = slack_text(json.loads(raw))
        if text:
            req = urllib.request.Request(SLACK, data=json.dumps({"text": text}).encode(),
                                         headers={"Content-Type": "application/json"})
            try:
                urllib.request.urlopen(req, timeout=10).read()
            except OSError:
                self.send_response(502)  # kosa8 retries later
                self.end_headers()
                return
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"ok")


HTTPServer(("", int(os.environ.get("PORT", "8080"))), Handler).serve_forever()

Go

// kosa8 webhook receiver: verify Kosa8-Signature, forward two events to Slack.
// KOSA8_WEBHOOK_SECRET=whsec_… SLACK_WEBHOOK_URL=https://hooks.slack.com/services/… go run .
package main

import (
	"bytes"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"strconv"
	"strings"
	"time"
)

// VerifyKosa8 checks the HMAC over "<t>." + raw body and refuses a t more
// than tolerance away (a replay).
func VerifyKosa8(raw []byte, header, secret string, tolerance time.Duration) bool {
	var ts, sig string
	for _, p := range strings.Split(header, ",") {
		k, v, ok := strings.Cut(strings.TrimSpace(p), "=")
		if !ok {
			continue
		}
		switch k {
		case "t":
			ts = v
		case "v1":
			sig = v
		}
	}
	t, err := strconv.ParseInt(ts, 10, 64)
	if err != nil {
		return false
	}
	if d := time.Since(time.Unix(t, 0)); d > tolerance || d < -tolerance {
		return false
	}
	got, err := hex.DecodeString(sig)
	if err != nil {
		return false
	}
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(ts + "."))
	mac.Write(raw)
	return hmac.Equal(got, mac.Sum(nil))
}

type event struct {
	Type string         `json:"type"`
	Data map[string]any `json:"data"`
}

func slackText(e event) string {
	switch e.Type {
	case "sandbox.error":
		return fmt.Sprintf(":warning: kosa8 sandbox %v failed during %v: %v", e.Data["id"], e.Data["during"], e.Data["error"])
	case "spending_cap.reached":
		cost, _ := e.Data["cost_usd"].(float64)
		return fmt.Sprintf(":money_with_wings: kosa8 Cloud reached its spending cap for %v: $%.2f of $%v. New work is refused until the cap is raised.",
			e.Data["month"], cost, e.Data["spending_cap_usd"])
	}
	return "" // other events: acknowledged, not forwarded
}

func main() {
	secret, slack := os.Getenv("KOSA8_WEBHOOK_SECRET"), os.Getenv("SLACK_WEBHOOK_URL")
	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		raw, err := io.ReadAll(io.LimitReader(r.Body, 1<<20)) // the exact bytes signed
		if err != nil || r.Method != http.MethodPost || !VerifyKosa8(raw, r.Header.Get("Kosa8-Signature"), secret, 5*time.Minute) {
			http.Error(w, "bad signature", http.StatusUnauthorized)
			return
		}
		var e event
		if err := json.Unmarshal(raw, &e); err != nil {
			http.Error(w, "bad body", http.StatusBadRequest)
			return
		}
		if text := slackText(e); text != "" {
			body, _ := json.Marshal(map[string]string{"text": text})
			resp, err := http.Post(slack, "application/json", bytes.NewReader(body))
			if err != nil || resp.StatusCode >= 300 {
				http.Error(w, "slack failed", http.StatusBadGateway) // kosa8 retries later
				return
			}
			resp.Body.Close()
		}
		fmt.Fprint(w, "ok")
	})
	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
	}
	http.ListenAndServe(":"+port, nil)
}

Already using a framework? Keep the verify function and call it with the raw body: in Express use express.raw({ type: "application/json" }) on that route, in Flask request.get_data(), in FastAPI await request.body().

Test it

Send yourself a signed event without waiting for a real one:

$ curl -s -X POST https://api.kosa8.com/v1/webhooks/wh_…/test -H "Authorization: Bearer $KOSA8_API_KEY"

That sends webhook.test, which the receivers above acknowledge without posting to Slack. To see the Slack message, set a spending cap just under this month's spend in Settings: the next create is refused and spending_cap.reached is sent.