Webhooks

@RestController
@RequestMapping("/webhooks/clickterm")
public class ClickTermWebhookController {

    private static final String TIMESTAMP_HEADER = "X-Clickterm-Timestamp";
    private static final String SIGNATURE_HEADER = "X-Clickterm-Signature";
    private static final long MAX_SKEW_SECONDS = 300L;

    @Value("${clickterm.webhook-secret}")
    private String webhookSecret;

    @PostMapping
    public ResponseEntity<Void> receiveWebhook(
        @RequestHeader(TIMESTAMP_HEADER) String timestamp,
        @RequestHeader(SIGNATURE_HEADER) String signature,
        @RequestBody byte[] rawBody
    ) throws Exception {
        long now = Instant.now().getEpochSecond();
        long requestTs = Long.parseLong(timestamp);

        if (Math.abs(now - requestTs) > MAX_SKEW_SECONDS) {
            return ResponseEntity.status(HttpStatus.UNAUTHORIZED).build();
        }

        String signedPayload = timestamp + "." + new String(rawBody, StandardCharsets.UTF_8);

        Mac mac = Mac.getInstance("HmacSHA256");
        mac.init(new SecretKeySpec(
            webhookSecret.getBytes(StandardCharsets.UTF_8),
            "HmacSHA256"
        ));

        String digest = HexFormat.of().formatHex(
            mac.doFinal(signedPayload.getBytes(StandardCharsets.UTF_8))
        );
        // Important: prepend "sha256=" to match the header format
        String expected = "sha256=" + digest;

        if (!MessageDigest.isEqual(
            signature.getBytes(StandardCharsets.UTF_8),
            expected.getBytes(StandardCharsets.UTF_8)
        )) {
            return ResponseEntity.status(HttpStatus.UNAUTHORIZED).build();
        }

        // Parse rawBody and process the event here.
        return ResponseEntity.ok().build();
    }
}

import crypto from "node:crypto";
import express from "express";

const app = express();

app.use(
  express.json({
    verify: (req, _res, buf) => {
      req.rawBody = buf;
    },
  })
);

const TIMESTAMP_HEADER = "X-Clickterm-Timestamp";
const SIGNATURE_HEADER = "X-Clickterm-Signature";
const MAX_SKEW_SECONDS = 300;

app.post("/webhooks/clickterm", (req, res) => {
  const secret = process.env.CLICKTERM_WEBHOOK_SECRET;
  const timestamp = req.header(TIMESTAMP_HEADER);
  const signature = req.header(SIGNATURE_HEADER);

  if (!secret || !timestamp || !signature) {
    return res.sendStatus(400);
  }

  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - Number(timestamp)) > MAX_SKEW_SECONDS) {
    return res.sendStatus(401);
  }

  const signingPayload = Buffer.concat([
    Buffer.from(`${timestamp}.`, "utf8"),
    req.rawBody ?? Buffer.from("", "utf8"),
  ]);

  const digest = crypto
    .createHmac("sha256", secret)
    .update(signingPayload)
    .digest("hex");

  // Important: prepend "sha256=" to match the header format
  const expected = `sha256=${digest}`;
  const isValid =
    signature.length === expected.length &&
    crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected));

  if (!isValid) {
    return res.sendStatus(401);
  }

  const { eventType, data } = req.body;

  if (eventType === "CLICKWRAP_EVENT_VERIFIED") {
    // Process the verified clickwrap event here.
    console.log(data.clickwrapEventId, data.clickwrapEventStatus);
  }

  return res.sendStatus(200);
});

import hashlib
import hmac
import os
import time

from flask import Flask, jsonify, request

app = Flask(__name__)

TIMESTAMP_HEADER = "X-Clickterm-Timestamp"
SIGNATURE_HEADER = "X-Clickterm-Signature"
MAX_SKEW_SECONDS = 300


@app.post("/webhooks/clickterm")
def clickterm_webhook():
    secret = os.environ.get("CLICKTERM_WEBHOOK_SECRET")
    timestamp = request.headers.get(TIMESTAMP_HEADER)
    signature = request.headers.get(SIGNATURE_HEADER)

    if not secret or not timestamp or not signature:
        return ("", 400)

    if abs(int(time.time()) - int(timestamp)) > MAX_SKEW_SECONDS:
        return ("", 401)

    raw_body = request.get_data()
    signed_payload = timestamp.encode("utf-8") + b"." + raw_body
    digest = hmac.new(
        secret.encode("utf-8"),
        signed_payload,
        hashlib.sha256,
    ).hexdigest()
    # Important: prepend "sha256=" to match the header format
    expected = f"sha256={digest}"

    if not hmac.compare_digest(signature, expected):
        return ("", 401)

    payload = request.get_json()

    if payload["eventType"] == "CLICKWRAP_EVENT_VERIFIED":
        data = payload["data"]
        print(data["clickwrapEventId"], data["clickwrapEventStatus"])

    return jsonify({"received": True}), 200