The listo command line

npm install -g listo-cli

brew install listo-delivery/tap/listo

curl -fsSL https://cli.listo.delivery/install.sh | sh

// Verifies the signature on an incoming Listo webhook before trusting a byte of
// it. Reads the raw request body on stdin, the way your web framework hands it
// to you, and refuses anything that does not check out.
//
//   npm i listo-delivery
//   LISTO_SIGNATURE='t=...,v1=...' LISTO_WEBHOOK_SECRETS=whsec_a,whsec_b \
//     node verify_webhook_signature.ts < body.json
//
import { readFileSync } from "node:fs";
import { Listo, ListoError } from "listo-delivery";

// The RAW bytes, before any JSON parsing. Parsing and re-serializing changes
// them (key order, escapes, whitespace) and the signature covers what was sent,
// not what your JSON library would have sent.
const payload = readFileSync(0);

try {
  // Checks the signature and the 300 second window in BOTH directions (the
  // replay this scheme exists to stop), then parses the body. `secrets` is a
  // LIST because a rotation has two secrets live at once: accept either while
  // you swap over, then drop the old one. No API key is needed.
  const event = Listo.webhooks.constructEvent(
    payload,
    process.env.LISTO_SIGNATURE,
    process.env.LISTO_WEBHOOK_SECRETS!.split(","),
  );

  // Events arrive at least once and in no guaranteed order, so treat `id` as the
  // thing you deduplicate on and the object's own timestamps as the thing you
  // order by. Answer 2xx quickly and do the work afterwards.
  console.log(`${event.type} ${event.id}`);

  switch (event.type) {
    case "delivery.status_changed":
    case "delivery.updated":
      console.log(`order ${event.data.external_id} is ${event.data.status}`);
      break;
    case "delivery.courier_location":
      console.log(`courier at ${event.data.location.lat},${event.data.location.lng}`);
      break;
    default:
      // A new type is an additive change, so ignore what you do not recognise
      // rather than failing on it.
      console.log("ignored");
  }
} catch (error) {
  // `invalid_signature` (a ListoSignatureError) and `invalid_payload` (a signed
  // body that is not a JSON object) are both a ListoError.
  if (!(error instanceof ListoError)) throw error;

  // Answer 400 and stop. Do not parse it, do not act on it, do not log it as an
  // event you received.
  console.error("signature rejected");
  process.exit(1);
}

# Verifies the signature on an incoming Listo webhook before trusting a byte of
# it. Reads the raw request body on stdin, the way your web framework hands it
# to you, and refuses anything that does not check out.
#
#   pip install listo-delivery
#   LISTO_SIGNATURE='t=...,v1=...' LISTO_WEBHOOK_SECRETS=whsec_a,whsec_b \
#     python verify_webhook_signature.py < body.json
#
import os
import sys

from listo_delivery import ListoError, webhooks
from listo_delivery.models import (
    DeliveryCourierLocationEvent,
    DeliveryStatusChangedEvent,
    DeliveryUpdatedEvent,
)

# The RAW bytes, before any JSON parsing. Parsing and re-serializing changes
# them (key order, escapes, whitespace) and the signature covers what was sent,
# not what your JSON library would have sent.
payload = sys.stdin.buffer.read()

try:
    # Checks the signature and the 300 second window in BOTH directions (the
    # replay this scheme exists to stop), then parses the body. `secrets` is a
    # LIST because a rotation has two secrets live at once: accept either while
    # you swap over, then drop the old one. No API key is needed.
    event = webhooks.construct_event(
        payload,
        os.environ.get("LISTO_SIGNATURE"),
        os.environ["LISTO_WEBHOOK_SECRETS"].split(","),
    )
except ListoError:
    # `invalid_signature` (a ListoSignatureError) and `invalid_payload` (a signed
    # body the SDK cannot read) are both a ListoError. Answer 400 and stop. Do
    # not parse it, do not act on it, do not log it as an event you received.
    sys.exit("signature rejected")

# Events arrive at least once and in no guaranteed order, so treat `id` as the
# thing you deduplicate on and the object's own timestamps as the thing you
# order by. Answer 2xx quickly and do the work afterwards.
kind = event.type_
print(f"{kind} {event.id}")

match event:
    case DeliveryStatusChangedEvent() | DeliveryUpdatedEvent():
        print(f"order {event.data.external_id} is {event.data.status}")
    case DeliveryCourierLocationEvent():
        print(f"courier at {event.data.location.lat},{event.data.location.lng}")
    case _:
        # A new type is an additive change, so ignore what you do not recognise
        # rather than failing on it.
        print("ignored")

# Verifies the signature on an incoming Listo webhook before trusting a byte of
# it. Reads the raw request body on stdin, the way your web framework hands it
# to you, and refuses anything that does not check out.
#
#   gem install listo-delivery
#   LISTO_SIGNATURE='t=...,v1=...' LISTO_WEBHOOK_SECRETS=whsec_a,whsec_b \
#     ruby verify_webhook_signature.rb < body.json
#
require "listo_delivery"

# The RAW bytes, before any JSON parsing. Parsing and re-serializing changes
# them (key order, escapes, whitespace) and the signature covers what was sent,
# not what your JSON library would have sent.
payload = $stdin.read

begin
  # Checks the signature and the 300 second window in BOTH directions (the
  # replay this scheme exists to stop), then parses the body. `secrets` is a
  # LIST because a rotation has two secrets live at once: accept either while
  # you swap over, then drop the old one. No API key is needed.
  event = ListoDelivery::Webhooks.construct_event(
    payload,
    ENV.fetch("LISTO_SIGNATURE", nil),
    ENV.fetch("LISTO_WEBHOOK_SECRETS").split(",")
  )
rescue ListoDelivery::Error
  # `invalid_signature` (SignatureError) and `invalid_payload` (a signed body
  # that is not a JSON object) are both a ListoDelivery::Error. Answer 400 and
  # stop. Do not parse it, do not act on it, do not log it as an event you
  # received.
  abort("signature rejected")
end

# Events arrive at least once and in no guaranteed order, so treat `id` as the
# thing you deduplicate on and the object's own timestamps as the thing you
# order by. Answer 2xx quickly and do the work afterwards.
puts "#{event.type} #{event.id}"

case event.type
when "delivery.status_changed", "delivery.updated"
  puts "order #{event.data.external_id} is #{event.data.status}"
when "delivery.courier_location"
  puts "courier at #{event.data.location.lat},#{event.data.location.lng}"
else
  # A new type is an additive change, so ignore what you do not recognise
  # rather than failing on it.
  puts "ignored"
end

# Verifies the signature on an incoming Listo webhook before trusting a byte of
# it. Reads the raw request body on stdin, the way your web framework hands it
# to you, and refuses anything that does not check out.
#
#   LISTO_SIGNATURE='t=...,v1=...' LISTO_WEBHOOK_SECRETS=whsec_a,whsec_b \
#     ruby verify_webhook_signature.rb < body.json
#
require "json"
require "openssl"

# The window a signature stays good for, matching the one Listo signs against.
# It is what stops a request someone captured from being replayed at you later.
TOLERANCE_SECONDS = 300

# `secrets` is a LIST because a rotation has two secrets live at once: accept
# either while you swap over, then drop the old one.
def listo_signature_valid?(payload, header, secrets, now: Time.now)
  match = /\At=(\d+),v1=(\h+)\z/.match(header.to_s)
  return false if match.nil?

  timestamp = match[1].to_i
  signature = match[2]

  # Checked in BOTH directions. A timestamp far in the past is the replay this
  # scheme exists to stop; one far in the future is the same attack against a
  # clock nobody looked at.
  return false if (now.to_i - timestamp).abs > TOLERANCE_SECONDS

  secrets.any? do |secret|
    expected = OpenSSL::HMAC.hexdigest("SHA256", secret, "#{timestamp}.#{payload}")
    OpenSSL.secure_compare(expected, signature)
  end
end

# The RAW bytes, before any JSON parsing. Parsing and re-serializing changes
# them (key order, escapes, whitespace) and the signature covers what was sent,
# not what your JSON library would have sent.
payload = $stdin.read
header = ENV.fetch("LISTO_SIGNATURE", nil)
secrets = ENV.fetch("LISTO_WEBHOOK_SECRETS").split(",")

unless listo_signature_valid?(payload, header, secrets)
  # Answer 400 and stop. Do not parse it, do not act on it, do not log it as an
  # event you received.
  abort("signature rejected")
end

event = JSON.parse(payload)

# Events arrive at least once and in no guaranteed order, so treat `id` as the
# thing you deduplicate on and the object's own timestamps as the thing you
# order by. Answer 2xx quickly and do the work afterwards.
puts "#{event["type"]} #{event["id"]}"

case event["type"]
when "delivery.status_changed", "delivery.updated"
  delivery = event["data"]
  puts "order #{delivery["external_id"]} is #{delivery["status"]}"
when "delivery.courier_location"
  location = event.dig("data", "location")
  puts "courier at #{location["lat"]},#{location["lng"]}"
else
  # A new type is an additive change, so ignore what you do not recognise
  # rather than failing on it.
  puts "ignored"
end