Texts and webhooks

"""Wait for the next text on a line that carries a code, and print the code.

Texts are read without their bodies first, so a text with no code is never
opened; the one with a code is opened once, which Ostra writes to its audit log.
The key needs messages:read and messages:read_content.
Python 3.9 or later, standard library only.
"""
import json
import os
import sys
import time
import urllib.parse
import urllib.request

API = os.environ["OSTRA_API_BASE_URL"]
KEY = os.environ["OSTRA_API_KEY"]
LINE = os.environ.get("LINE", "LAX-SANDBOX-00")
GIVE_UP = time.time() + int(os.environ.get("WAIT_SECONDS", "120"))


def get(path, query=None):
    url = API + path + ("?" + urllib.parse.urlencode(query) if query else "")
    req = urllib.request.Request(url, headers={"Authorization": "Bearer " + KEY})
    with urllib.request.urlopen(req, timeout=100) as r:
        return r.status, r.headers, (json.load(r) if r.status == 200 else None)


cursor = None  # None: wait for the next text from now
while time.time() < GIVE_UP:
    query = {"timeout": 60, "limit": 1}
    if cursor:
        query["cursor"] = cursor
    status, headers, page = get(f"/v1/lines/{LINE}/messages/wait", query)
    if status == 204:  # nothing arrived; wait on from where the server says
        cursor = headers["Ostra-Cursor"]
        continue
    cursor = page["next_cursor"]
    text = page["messages"][0]
    if not text["has_code"]:
        continue
    _, _, opened = get(f"/v1/lines/{LINE}/messages/{text['id']}")
    if opened["message"]["code"]:
        print(opened["message"]["code"])
        sys.exit(0)

sys.exit("no code arrived")

LINE defaults to the sandbox line. In a test, start this before your service sends the text, so the wait from now sees it.

// Wait for the next text on a line that carries a code, and print the code.
// Texts are read without their bodies first; only the one with a code is
// opened, once, and Ostra writes that view to its audit log.
// The key needs messages:read and messages:read_content. Node 18 or later.
const api = process.env.OSTRA_API_BASE_URL
const auth = { authorization: `Bearer ${process.env.OSTRA_API_KEY}` }
const line = process.env.LINE ?? 'LAX-SANDBOX-00'
const giveUp = Date.now() + Number(process.env.WAIT_SECONDS ?? 120) * 1000

let cursor // undefined: wait for the next text from now
while (Date.now() < giveUp) {
  const query = new URLSearchParams({ timeout: '60', limit: '1', ...(cursor ? { cursor } : {}) })
  const res = await fetch(`${api}/v1/lines/${line}/messages/wait?${query}`, { headers: auth })
  if (res.status === 204) {
    cursor = res.headers.get('ostra-cursor') // nothing arrived; wait on from here
    continue
  }
  if (!res.ok) throw new Error(`${res.status}: ${await res.text()}`)
  const page = await res.json()
  cursor = page.next_cursor
  const [text] = page.messages
  if (!text.has_code) continue
  const opened = await (await fetch(`${api}/v1/lines/${line}/messages/${text.id}`, { headers: auth })).json()
  if (opened.message.code) {
    console.log(opened.message.code)
    process.exit(0)
  }
}
console.error('no code arrived')
process.exit(1)

#!/bin/sh
# An endpoint for texts on every line you hold, with the code in each event.
# include_code needs messages:read_content. The answer's secret (ows_...) is
# shown once: keep it where your receiver can read it.
curl -sS -X POST "$OSTRA_API_BASE_URL/v1/webhooks" \
  -H "Authorization: Bearer $OSTRA_API_KEY" \
  -H 'content-type: application/json' \
  -d '{"url": "https://hooks.example.com/ostra", "events": ["message.received"], "include_code": true}'

{
  "id": "evt_6f9c2d1e4b8a4c7f9e0d1a2b3c4d5e6f",
  "type": "message.received",
  "created_at": "2026-10-12T17:05:02.114Z",
  "organization_id": "3d6f2a8e-1b4c-4e9a-8f7d-0c5b2a1e9d4f",
  "sandbox": true,
  "data": {
    "line_id": "LAX-SANDBOX-00",
    "message": {
      "id": "6c1f0e52-7d1b-4f8e-9a51-2b7c4e0d93aa",
      "sender": "+13105550100",
      "sender_label": null,
      "received_at": "2026-10-12T17:05:02.114Z",
      "source": "simulated",
      "service": null,
      "has_code": true,
      "code": "482915"
    }
  }
}

t=<Ostra-Timestamp>,v1=<hex HMAC-SHA256 of "<t>.<raw body>">

// A webhook receiver that checks every delivery's signature before trusting it.
// OSTRA_WEBHOOK_SECRET is the endpoint's whole secret, ows_ included. Node 18 or later, no packages.
import { createHmac, timingSafeEqual } from 'node:crypto'
import { createServer } from 'node:http'

const secret = process.env.OSTRA_WEBHOOK_SECRET
const TOLERANCE = 300 // seconds a signature's time may be from this clock
// Event id -> when handled. An event can arrive more than once: retries run for
// 24 hours, and a failed delivery can be sent again by hand for 30 days. This
// in-memory map is for the sample; keep handled ids in your own database.
const seen = new Map()
const KEEP_MS = 31 * 24 * 3600 * 1000

/** Ostra-Signature is t=<unix seconds>,v1=<hex HMAC-SHA256 of "<t>.<raw body>">. */
export function verifyOstraWebhook(secret, header, rawBody, nowSeconds = Math.floor(Date.now() / 1000)) {
  const parts = Object.fromEntries(header.split(',').map((p) => [p.slice(0, p.indexOf('=')), p.slice(p.indexOf('=') + 1)]))
  const t = Number(parts.t)
  if (!Number.isInteger(t) || Math.abs(nowSeconds - t) > TOLERANCE) return false
  const want = createHmac('sha256', secret).update(`${t}.`).update(rawBody).digest()
  const got = Buffer.from(parts.v1 ?? '', 'hex')
  return got.length === want.length && timingSafeEqual(got, want)
}

createServer((req, res) => {
  const chunks = []
  req.on('data', (c) => chunks.push(c))
  req.on('end', () => {
    const raw = Buffer.concat(chunks) // the exact bytes received, before any parsing
    if (!verifyOstraWebhook(secret, String(req.headers['ostra-signature'] ?? ''), raw)) {
      res.writeHead(400).end()
      return
    }
    res.writeHead(204).end() // answer within 10 seconds; do the work after
    // Deduplicate on the id inside the signed body, never on a header the signature does not cover.
    const event = JSON.parse(raw)
    const now = Date.now()
    for (const [id, at] of seen) if (now - at > KEEP_MS) seen.delete(id)
    if (seen.has(event.id)) return
    seen.set(event.id, now)
    if (event.type === 'message.received') {
      const m = event.data.message
      // Hand m.code to your test here. Codes are sensitive: do not log them.
      console.log(`${event.data.line_id}: text from ${m.sender}, ${m.code ? 'code received' : 'no code'}`)
    } else {
      console.log(event.type)
    }
  })
}).listen(Number(process.env.PORT ?? 8080), '127.0.0.1', () => console.log(`listening on ${process.env.PORT ?? 8080}`))

Ostra's checks send this receiver a delivery the real server signed, then the same delivery again, again with a changed event id header, with one byte of the body changed, with an old timestamp and with no signature: it handles the first once and refuses the last three.

"""A webhook receiver that checks every delivery's signature before trusting it.

OSTRA_WEBHOOK_SECRET is the endpoint's whole secret, ows_ included.
Python 3.9 or later, standard library only.
"""
import hashlib
import hmac
import json
import os
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

SECRET = os.environ["OSTRA_WEBHOOK_SECRET"].encode()
TOLERANCE = 300  # seconds a signature's time may be from this clock
# Event id -> when handled. An event can arrive more than once: retries run for
# 24 hours, and a failed delivery can be sent again by hand for 30 days. This
# in-memory dict is for the sample; keep handled ids in your own database.
SEEN = {}
KEEP = 31 * 24 * 3600


def verify_ostra_webhook(secret, header, raw_body, now=None):
    """Ostra-Signature is t=<unix seconds>,v1=<hex HMAC-SHA256 of "<t>.<raw body>">."""
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
    try:
        t = int(parts.get("t", ""))
    except ValueError:
        return False
    if abs((now or time.time()) - t) > TOLERANCE:
        return False
    want = hmac.new(secret, f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(want, parts.get("v1", ""))


class Receiver(BaseHTTPRequestHandler):
    def do_POST(self):
        raw = self.rfile.read(int(self.headers.get("content-length", 0)))  # the exact bytes received
        if not verify_ostra_webhook(SECRET, self.headers.get("ostra-signature", ""), raw):
            self.send_response(400)
            self.end_headers()
            return
        self.send_response(204)  # answer within 10 seconds; do the work after
        self.end_headers()
        # Deduplicate on the id inside the signed body, never on a header the signature does not cover.
        event = json.loads(raw)
        now = time.time()
        for event_id, at in list(SEEN.items()):
            if now - at > KEEP:
                del SEEN[event_id]
        if event["id"] in SEEN:
            return
        SEEN[event["id"]] = now
        if event["type"] == "message.received":
            message = event["data"]["message"]
            # Hand message["code"] to your test here. Codes are sensitive: do not log them.
            got = "code received" if message.get("code") else "no code"
            print(f"{event['data']['line_id']}: text from {message['sender']}, {got}", flush=True)
        else:
            print(event["type"], flush=True)

    def log_message(self, *args):
        pass


port = int(os.environ.get("PORT", "8080"))
print(f"listening on {port}", flush=True)
HTTPServer(("127.0.0.1", port), Receiver).serve_forever()

Both receivers listen on this machine only. Ostra delivers only to https URLs, so run them behind your own HTTPS server.