Webhooks

const crypto = require("crypto");

const TOLERANCE_SECONDS = 5 * 60;

// rawBody must be a Buffer of the unmodified request body bytes
// (e.g. captured via express.raw({ type: 'application/json' })).
function verifyWebhookSignature(headers, rawBody, publicKeyPem) {
  const signature = headers["flatpeak-signature"];
  const timestamp = headers["flatpeak-timestamp"];

  if (!signature || signature === "none" || !timestamp) {
    return false;
  }

  // Replay protection.
  const skew = Math.abs(
    Math.floor(Date.now() / 1000) - parseInt(timestamp, 10),
  );
  if (Number.isNaN(skew) || skew > TOLERANCE_SECONDS) {
    return false;
  }

  // Strip v1= prefix; Buffer.from supports base64url natively.
  const sigBuffer = Buffer.from(signature.replace(/^v1=/, ""), "base64url");

  // Signed message: "{timestamp}.{raw_body}".
  const message = Buffer.concat([Buffer.from(`${timestamp}.`), rawBody]);

  const verifier = crypto.createVerify("SHA256");
  verifier.update(message);
  return verifier.verify(
    {
      key: publicKeyPem,
      padding: crypto.constants.RSA_PKCS1_PSS_PADDING,
      saltLength: crypto.constants.RSA_PSS_SALTLEN_DIGEST,
    },
    sigBuffer,
  );
}

import base64
import time
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding

TOLERANCE_SECONDS = 5 * 60

def verify_webhook_signature(body: bytes, headers: dict, public_key_pem: bytes) -> bool:
    """`body` must be the raw request bytes, e.g. Flask's `request.get_data()`."""
    signature_header = headers.get("Flatpeak-Signature", "")
    timestamp = headers.get("Flatpeak-Timestamp", "")

    if not signature_header or signature_header == "none" or not timestamp:
        return False

    # Replay protection.
    try:
        if abs(int(time.time()) - int(timestamp)) > TOLERANCE_SECONDS:
            return False
    except ValueError:
        return False

    # Strip v1= prefix; pad so length is a multiple of 4 for urlsafe_b64decode.
    sig_b64 = signature_header.removeprefix("v1=")
    sig_b64 += "=" * (-len(sig_b64) % 4)
    sig_bytes = base64.urlsafe_b64decode(sig_b64)

    message = f"{timestamp}.".encode() + body

    public_key = serialization.load_pem_public_key(public_key_pem)
    try:
        public_key.verify(
            sig_bytes,
            message,
            padding.PSS(
                mgf=padding.MGF1(hashes.SHA256()),
                salt_length=padding.PSS.DIGEST_LENGTH,
            ),
            hashes.SHA256(),
        )
        return True
    except InvalidSignature:
        return False

package webhook

import (
 "crypto"
 "crypto/rsa"
 "crypto/sha256"
 "crypto/x509"
 "encoding/base64"
 "encoding/pem"
 "fmt"
 "math"
 "strconv"
 "strings"
 "time"
)

const toleranceSeconds = 5 * 60

// VerifyWebhookSignature returns nil if rawBody is authentic, an error otherwise.
// rawBody must be the unmodified request body bytes (read once from r.Body).
func VerifyWebhookSignature(rawBody []byte, timestamp, signatureHeader string, publicKeyPEM []byte) error {
 if signatureHeader == "" || signatureHeader == "none" || timestamp == "" {
  return fmt.Errorf("missing or unsigned webhook")
 }

 // Replay protection.
 ts, err := strconv.ParseInt(timestamp, 10, 64)
 if err != nil {
  return fmt.Errorf("invalid timestamp: %w", err)
 }
 if math.Abs(float64(time.Now().Unix()-ts)) > toleranceSeconds {
  return fmt.Errorf("timestamp outside tolerance window")
 }

 sigB64 := strings.TrimPrefix(signatureHeader, "v1=")
 sigBytes, err := base64.RawURLEncoding.DecodeString(sigB64)
 if err != nil {
  return fmt.Errorf("failed to decode signature: %w", err)
 }

 // Signed message: "{timestamp}.{raw_body}".
 h := sha256.New()
 h.Write([]byte(timestamp))
 h.Write([]byte("."))
 h.Write(rawBody)
 digest := h.Sum(nil)

 block, _ := pem.Decode(publicKeyPEM)
 if block == nil {
  return fmt.Errorf("failed to parse PEM block")
 }
 pub, err := x509.ParsePKIXPublicKey(block.Bytes)
 if err != nil {
  return fmt.Errorf("failed to parse public key: %w", err)
 }
 rsaPub, ok := pub.(*rsa.PublicKey)
 if !ok {
  return fmt.Errorf("public key is not RSA")
 }

 return rsa.VerifyPSS(rsaPub, crypto.SHA256, digest, sigBytes, &rsa.PSSOptions{
  SaltLength: rsa.PSSSaltLengthEqualsHash,
 })
}