簽章與驗證

X-ListingsAPI-Signature: sha256=<digest>   digest = base64( HMAC-SHA256(secret, raw_request_body) )

{  "event": "endpoint.verification",  "nonce": "9f2c1e642b1a4b0e9f1a2c7d5e8a1b3c",  "timestamp": "2026-08-25T14:32:10Z"}

const crypto = require('crypto');const express = require('express'); const app = express();const SECRET = process.env.LISTINGSAPI_WEBHOOK_SECRET; // Keep the raw bytes. express.json() would otherwise leave you// with only the parsed object, and re-serializing it changes the// bytes the HMAC covers.app.use(express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }),); function signatureIsValid(rawBody, header) {const digest = crypto  .createHmac('sha256', SECRET)  .update(rawBody)  .digest('base64');const expected = 'sha256=' + digest; const a = Buffer.from(header || '', 'utf8');const b = Buffer.from(expected, 'utf8');// timingSafeEqual throws on a length mismatch, so check first.return a.length === b.length && crypto.timingSafeEqual(a, b);} // The only work allowed before the 200: put the raw bytes somewhere durable// and wait for the write to be acknowledged. Swap the INSERT for your real// queue or table write; it must be committed before you respond, because// nothing is retried if your process dies after the 200 goes out.async function storeForProcessing(rawBody, signature) {await db.query(  'INSERT INTO webhook_inbox (signature, body, received_at) VALUES ($1, $2, now())',  [signature, rawBody.toString('utf8')],);} app.post('/webhooks/listingsapi', async (req, res) => {const header = req.get('X-ListingsAPI-Signature');if (!signatureIsValid(req.rawBody, header)) {  return res.status(401).send('bad signature');} const body = req.body; // The verification handshake: answer with HEX, not base64.if (body.event === 'endpoint.verification') {  const answer = crypto    .createHmac('sha256', SECRET)    .update(body.nonce)    .digest('hex');  return res.status(200).send(answer);} // A real event: durable write first, acknowledge second.try {  await storeForProcessing(req.rawBody, header);} catch (err) {  // Nothing was persisted, so do not claim success.  return res.status(500).send('could not persist');}return res.status(200).json({ received: true });});

import base64import hashlibimport hmacimport os from flask import Flask, abort, request app = Flask(__name__)SECRET = os.environ["LISTINGSAPI_WEBHOOK_SECRET"].encode() HEADER = "X-ListingsAPI-Signature"  def signature_is_valid(raw_body: bytes, header: str | None) -> bool:  digest = hmac.new(SECRET, raw_body, hashlib.sha256).digest()  expected = "sha256=" + base64.b64encode(digest).decode()  return hmac.compare_digest(header or "", expected)  def store_for_processing(raw_body: bytes, signature: str) -> None:  """The only work allowed before the 200: put the raw bytes somewhere  durable and return only once the write is committed. Swap the INSERT for  your real queue or table write - nothing is retried if the process dies  after the response goes out."""  with db.begin() as tx:      tx.execute(          "INSERT INTO webhook_inbox (signature, body) VALUES (%s, %s)",          (signature, raw_body),      )  @app.post("/webhooks/listingsapi")def webhooks():  raw = request.get_data()  # raw bytes, exactly as received  header = request.headers.get(HEADER)  if not signature_is_valid(raw, header):      abort(401)   body = request.get_json(silent=True) or {}   # The verification handshake: answer with HEX, not base64.  if body.get("event") == "endpoint.verification":      nonce = body["nonce"].encode()      return hmac.new(SECRET, nonce, hashlib.sha256).hexdigest(), 200   # A real event: durable write first, acknowledge second.  try:      store_for_processing(raw, header or "")  except Exception:      # Nothing was persisted, so do not claim success.      return {"error": "could not persist"}, 500  return {"received": True}, 200

whsec_test_0123456789abcdef0123456789abcdef

{"event":"connection.location_connected","timestamp":"2026-08-25T14:32:10Z","account_id":11073,"location_id":"279381","data":{"platform":"google","connected_account_id":"ba03bc4b-9f8c-4d3b-8e93-9628034c63cc"}}

X-ListingsAPI-Signature: sha256=+FxQAwiCAfQdicZcHlv9fNOrkjf0yK99Jje97VMS6Y4=

9f2c1e642b1a4b0e9f1a2c7d5e8a1b3c

d5aff73f9739b49aaabccf264c8777a697998ba07c493fbaec646b5370354145

import base64, hashlib, hmac secret = b"whsec_test_0123456789abcdef0123456789abcdef"body = open("body.json", "rb").read()          # the 209 bytes abovenonce = b"9f2c1e642b1a4b0e9f1a2c7d5e8a1b3c" # request signature: base64, over the raw bodyprint("sha256=" + base64.b64encode(hmac.new(secret, body, hashlib.sha256).digest()).decode())# challenge answer: lowercase hex, over the nonce aloneprint(hmac.new(secret, nonce, hashlib.sha256).hexdigest())