Introduction

Pinwheel signs all webhook events that it sends to your endpoints. This allows you to verify that events were sent by Pinwheel, not a third party.

Verifying Pinwheel is the Sender

Each webhook event includes the x-pinwheel-signature header. This header contains a signature in the form {signature_version}={signature_digest}. The {signature_version} should be v2. The {signature_digest} is a hex-encoded string generated using the signature algorithm.

You should use a string comparison function that is safe against timing attacks when comparing signatures. Otherwise, it may be possible for an attacker to ascertain information about secret values such as your API Secret. See the examples below for constant-time comparison functions in various languages.

Signature Algorithm

The signature algorithm is specified in pseudo-code. We also include example implementations in various languages. To verify that your implementation is compliant, please leverage the test suite at the bottom of this page.

1. Retrieve the {timestamp}, stored in the x-timestamp header.
2. Retrieve the {raw_request_body}.
   1. This should be the un-parsed, un-decoded bytes from the HTTP request body. No UTF-8 decoding or JSON processing should have occurred on this value. We use the raw bytes to avoid per-language/framework processing differences.
   2. See the examples below for how to retrieve this in various web frameworks.
3. Set {key} as your API Secret.
4. Construct the {msg_prefix} by UTF-8 encoding the string {signature_version}:{timestamp}:. Notice the trailing :.
5. Construct the {msg} by byte-concatenating {msg_prefix} and {raw_request_body}.
6. Calculate the {raw_digest} by running HMAC-SHA256 with {key} on {msg}.
7. Produce {signature_digest} by hex-encoding the {raw_digest}.

Example Implementations

# Python
import hmac
import hashlib

def verify_signature(signature, timestamp, raw_body):
  msg = f"v2:{timestamp}:".encode("utf-8") + raw_body
  digest = hmac.new(
      key="YOUR_API_SECRET".encode("utf-8"),
      msg=msg,
      digestmod=hashlib.sha256,
    ).hexdigest()
  generated_signature = f"v2={digest}"
  return hmac.compare_digest(signature, generated_signature)

# Example for Flask
def webhook_view(request):
  signature = request.headers.get("x-pinwheel-signature")
  timestamp = request.headers.get("x-timestamp")
  raw_body = request.get_data(parse_form_data=True) # raw HTTP request body as a bytestring
  # for Django, `raw_body = request.body`
  # for Starlette/FastAPI, `raw_body = await request.body()`
  if not verify_signature(signature, timestamp, raw_body):
    raise Exception("Invalid webhook request signature")

// Javascript
const crypto = require('crypto');
const express = require('express');

function verifySignature(signature, timestamp, rawBody) {
  const prefix = Buffer.from(`v2:${timestamp}:`, 'utf8');
  const message = Buffer.concat([prefix, rawBody]);
  const digest = crypto.createHmac('sha256', Buffer.from(process.env.YOUR_PINWHEEL_API_SECRET, 'utf8')).update(message).digest('hex');
  const generatedSignature = `v2=${digest}`;

  const bufSig = Buffer.from(signature);
  const bufGen = Buffer.from(generatedSignature);

  return bufSig.length === bufGen.length && crypto.timingSafeEqual(bufSig, bufGen);
}

export const authMiddleware = express.json({
  verify: (request, response, buffer) => {
    const signature = request.headers['x-pinwheel-signature']
    const timestamp = request.headers['x-timestamp']

    if (verifySignature(signature, timestamp, buffer)) return;

    throw new Error('Invalid webhook request signature');
  },
});

// Java
// - leveraging the okio library (https://square.github.io/okio/)
import okio.Buffer;
import okio.ByteString;

class PinwheelWebhookUtilities {
  static ByteString SIGNATURE_PREFIX = ByteString.encodeUtf8("v2:");
  static ByteString SIGNATURE_SEP = ByteString.encodeUtf8(":");
  static int SIGNATURE_VERSION_HEADER_LEN = "v2=".length();

  public static boolean verifySignature(String signature, String timestamp, ByteString rawBody) {
    try (Buffer msg = new Buffer()) {
      msg.write(SIGNATURE_PREFIX)
          .write(ByteString.encodeUtf8(timestamp))
          .write(SIGNATURE_SEP)
          .write(rawBody);
      ByteString digest = msg.hmacSha256(ByteString.encodeUtf8("YOUR_API_SECRET"));
      ByteString expectedDigest = ByteString.decodeHex(signature.substring(SIGNATURE_VERSION_HEADER_LEN));
      return digest.equals(expectedDigest);
    }
  }
}

Test Suite

We've put together test cases to cover a variety of payload structures your signature implementation may need to accommodate. To be prepared for the set of valid webhook events our system may send your endpoints your signature algorithm implementation must be compliant with the first 4 (text-based) test cases and should be compliant with the last (non-text) test case.

We used the following parameters as fixed inputs when generating each {signature_digest}:

• {timestamp} = 860860860
• {key} = the string TEST_KEY UTF-8 encoded, i.e. 544553545f4b4559 in hex, i.e. "TEST_KEY".encode("utf-8") in Python.

You should download the test case files directly, using a command such as curl --remote-name $TEST_CASE_FILE_URL, to avoid introducing newlines or other extraneous characters into them which will affect their binary contents. Similarly, when viewing test cases make sure to use a read-only mechanism, such as cat $TEST_CASE_FILE, to avoid affecting their binary contents.

IP Address Allow List

Pinwheel will only send webhook events from a fixed number of IP addresses. For an extra layer of security, you should reject requests from elsewhere. The IP addresses are:

• 3.227.93.147/32
• 18.214.233.159/32
• 35.168.171.219/32
• 54.235.190.63/32

Please contact [email protected] for access to our Developer Dashboard.