#!/usr/bin/env python3 """Panvaya scheduled Ocean webhook receiver example (Python 3.9+, no packages). Run locally: export PANVAYA_WEBHOOK_SECRET='replace-with-your-own-long-random-receiver-secret' python3 panvaya-webhook-receiver.py Defaults: bind 127.0.0.1:8089, POST /panvaya/ocean, SQLite inbox ./panvaya-webhook-inbox.sqlite3. Optional: PANVAYA_RECEIVER_HOST, PANVAYA_RECEIVER_PORT, PANVAYA_RECEIVER_DB. Use a separate receiver secret, not your Panvaya API key. Configure the destination with authType=API_KEY, authHeaderName=X-Custom-Webhook-Secret, authSecret=the same receiver secret. Use the downloadable Postman collection's "Test local receiver" request for offline testing with the developer sample. Send it twice with the same delivery_id to verify deduplication. No carrier call or API credit is involved. This is a local integration example. For deployment, port its handler into your production HTTP framework and use a trusted public HTTPS endpoint on port 443. Panvaya cannot deliver to localhost/private addresses. Current deliveries use receiver credentials over TLS, without a Panvaya HMAC signature. 202 means committed to the local inbox, not written into a TMS/ERP. A downstream worker must map inbox payloads to your system and track its own processing state. Keep raw payloads private and implement appropriate retention. Share a durable inbox across replicas in production; do not use independent per-instance files. Retries/replays reuse X-Panvaya-Delivery-Id and can have different live AIS data. Deduplicate by delivery ID, not by payload hash or timestamp. Return 2xx only after durable acceptance; DB failures return retryable 503. Never log raw bodies, receiver secrets or API keys. See https://panvaya.com/developers/webhooks. """ import hmac import json import logging import os import sqlite3 import uuid from contextlib import closing from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer MAX_BODY_BYTES = 5_000_000 LOG = logging.getLogger("panvaya.receiver") def reject_non_json_constant(_value): raise ValueError("Non-JSON numeric constant") def create_server(address, secret, database_path): """Create a receiver; caller owns the server's lifetime and SQLite file.""" if not isinstance(secret, str) or len(secret.encode("utf-8")) < 32: raise ValueError("Set a receiver secret of at least 32 bytes") if str(database_path) == ":memory:": raise ValueError("Use a persistent SQLite file for delivery deduplication") secret_bytes = secret.encode("utf-8") # Restrict newly created inbox files to the current operating-system user. descriptor = os.open(database_path, os.O_CREAT | os.O_RDWR, 0o600) os.close(descriptor) with closing(sqlite3.connect(database_path, timeout=5)) as db: db.execute("""CREATE TABLE IF NOT EXISTS webhook_inbox ( delivery_id TEXT PRIMARY KEY, subscription_id TEXT NOT NULL, received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, body TEXT NOT NULL )""") db.commit() class Handler(BaseHTTPRequestHandler): def setup(self): super().setup() self.connection.settimeout(10) def log_message(self, *_args): # BaseHTTPRequestHandler includes the caller-controlled URL in logs. pass def reply(self, status, message): body = json.dumps({"status": message}).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Connection", "close") self.end_headers() self.close_connection = True try: self.wfile.write(body) except (BrokenPipeError, ConnectionResetError): # A sender retry will resolve the committed delivery ID. pass def single_header(self, name): values = self.headers.get_all(name, []) return values[0] if len(values) == 1 else None def do_POST(self): if self.path != "/panvaya/ocean": self.reply(404, "not_found") return supplied = self.single_header("X-Custom-Webhook-Secret") if supplied is None or not hmac.compare_digest(supplied.encode("utf-8"), secret_bytes): self.reply(401, "unauthorized") return if self.headers.get("Transfer-Encoding") is not None: self.reply(400, "unsupported_transfer_encoding") return content_type = self.single_header("Content-Type") or "" if content_type.split(";", 1)[0].strip().lower() != "application/json": self.reply(415, "expected_json") return length_text = self.single_header("Content-Length") if length_text is None or not length_text.isascii() or not length_text.isdecimal(): self.reply(400, "invalid_content_length") return if len(length_text) > 10: self.reply(413, "body_too_large") return length = int(length_text) if length > MAX_BODY_BYTES: self.reply(413, "body_too_large") return try: delivery_id = str(uuid.UUID(self.single_header("X-Panvaya-Delivery-Id") or "")) subscription_id = str(uuid.UUID(self.single_header("X-Panvaya-Subscription-Id") or "")) except ValueError: self.reply(400, "invalid_delivery_identity") return if self.single_header("X-Panvaya-Product") != "WEBHOOK_SEA_TRACKING": self.reply(400, "unsupported_product") return try: raw = self.rfile.read(length) if len(raw) != length: self.reply(400, "incomplete_body") return text = raw.decode("utf-8") payload = json.loads(text, parse_constant=reject_non_json_constant) data = payload.get("data") if isinstance(payload, dict) else None if not isinstance(data, dict) or "errors" not in payload: raise ValueError() if not isinstance(data.get("reference"), dict) or not isinstance(data.get("containers"), list): raise ValueError() except (ValueError, UnicodeDecodeError, RecursionError): self.reply(400, "invalid_ocean_response") return except (TimeoutError, OSError): self.reply(408, "body_timeout") return try: with closing(sqlite3.connect(database_path, timeout=5)) as db: with db: saved = db.execute( "INSERT INTO webhook_inbox (delivery_id, subscription_id, body) VALUES (?, ?, ?) " "ON CONFLICT(delivery_id) DO NOTHING", (delivery_id, subscription_id, text), ).rowcount == 1 if not saved: owner = db.execute("SELECT subscription_id FROM webhook_inbox WHERE delivery_id = ?", (delivery_id,)).fetchone() if owner[0] != subscription_id: self.reply(409, "delivery_identity_conflict") return LOG.info("webhook_received deliveryId=%s outcome=%s", delivery_id, "accepted" if saved else "duplicate") self.reply(202 if saved else 200, "accepted" if saved else "duplicate") except sqlite3.Error: LOG.error("webhook_inbox_unavailable deliveryId=%s", delivery_id) self.reply(503, "inbox_unavailable") server = ThreadingHTTPServer(address, Handler) server.daemon_threads = True return server def main(): logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") secret = os.environ.get("PANVAYA_WEBHOOK_SECRET", "") address = (os.environ.get("PANVAYA_RECEIVER_HOST", "127.0.0.1"), int(os.environ.get("PANVAYA_RECEIVER_PORT", "8089"))) database = os.environ.get("PANVAYA_RECEIVER_DB", "panvaya-webhook-inbox.sqlite3") with create_server(address, secret, database) as server: LOG.info("webhook_receiver_started port=%s", server.server_port) try: server.serve_forever() except KeyboardInterrupt: LOG.info("webhook_receiver_stopped") if __name__ == "__main__": main()