#!/usr/bin/env python3 """WhatsApp Cloud API webhook doctor. Read-only by default. Checks each link of the chain between Meta and your webhook receiver (n8n WhatsApp Trigger, Make Watch Events, or a custom endpoint) and prints which link is broken. Only write action: --apply runs POST /{waba-id}/subscribed_apps, and only when the app is missing from the WABA's subscribed_apps. It never deletes anything. Secrets come from environment variables and are never printed: META_APP_TOKEN app access token (APP_ID|APP_SECRET style) META_SYSTEM_TOKEN system user token with whatsapp_business_management Python 3 standard library only. """ import argparse import json import os import sys import urllib.error import urllib.parse import urllib.request CHALLENGE = "12345" TEST_PATH_HINT = "/webhook-test/" SECRETS = [] # token values, scrubbed from any text we print # ---------------------------------------------------------------- HTTP layer def live_transport(method, url, params, form): """The only function that touches the network. Mock mode replaces it.""" if params: url = url + ("&" if "?" in url else "?") + urllib.parse.urlencode(params) data = urllib.parse.urlencode(form).encode() if form is not None else None req = urllib.request.Request(url, data=data, method=method) try: with urllib.request.urlopen(req, timeout=20) as r: return r.status, r.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: return e.code, e.read().decode("utf-8", "replace") except (urllib.error.URLError, OSError) as e: return 0, "network error: %s" % getattr(e, "reason", e) class MockTransport: """Maps each call to a fixture file in a directory. Records calls.""" def __init__(self, directory, base): self.dir = directory self.base = base.rstrip("/") self.calls = [] self.applied = False def _load(self, name): path = os.path.join(self.dir, name) if not os.path.isfile(path): return 404, json.dumps({"error": {"message": "no fixture " + name, "code": 0}}) with open(path, encoding="utf-8") as f: return 200, f.read() def __call__(self, method, url, params, form): self.calls.append((method, url.split("?")[0])) if not url.startswith(self.base + "/"): status, text = self._load("callback_probe.json") if status != 200: return status, text fx = json.loads(text) return fx["status"], fx["body"] parts = [p for p in url[len(self.base):].split("/") if p][1:] # drop version edge = parts[1] if len(parts) > 1 else "" if method == "POST" and edge == "subscribed_apps": self.applied = True return self._load("apply_response.json") if edge == "subscribed_apps": if self.applied and os.path.isfile(os.path.join(self.dir, "waba_subscribed_apps_after_apply.json")): return self._load("waba_subscribed_apps_after_apply.json") return self._load("waba_subscribed_apps.json") names = {"": "app_info.json", "subscriptions": "app_subscriptions.json", "phone_numbers": "waba_phone_numbers.json"} return self._load(names.get(edge, "unknown.json")) TRANSPORT = live_transport def http_get(url, params=None): return TRANSPORT("GET", url, params, None) def http_post(url, form): return TRANSPORT("POST", url, None, form) def redact(text): for s in SECRETS: if s: text = text.replace(s, "***") return text # ------------------------------------------------------------- Graph helpers def graph(cfg, method, path, token, params=None): """Returns (data, error_string). Token goes in a param, never printed.""" url = "%s/%s/%s" % (cfg.graph_base.rstrip("/"), cfg.graph_version, path) if method == "GET": p = dict(params or {}) p["access_token"] = token status, text = http_get(url, p) else: f = dict(params or {}) f["access_token"] = token status, text = http_post(url, f) try: data = json.loads(text) except ValueError: return None, redact("HTTP %s, non-JSON reply: %s" % (status, text[:120])) if status != 200 or (isinstance(data, dict) and "error" in data): e = data.get("error", {}) if isinstance(data, dict) else {} msg = e.get("message", text[:120]) return None, redact("HTTP %s, Graph error code %s: %s" % (status, e.get("code", "?"), msg)) return data, None def result(name, status, detail, fix=""): return {"name": name, "status": status, "detail": detail, "fix": fix} def field_names(fields): out = [] for f in fields or []: out.append(f.get("name") if isinstance(f, dict) else str(f)) return out def norm_url(u): return (u or "").strip().rstrip("/") # -------------------------------------------------------------------- checks def check_app_sanity(cfg, tok): name = "1. App id is valid and the app token works" if not tok: return result(name, "SKIP", "META_APP_TOKEN not set.", "export META_APP_TOKEN='APP_ID|APP_SECRET'") data, err = graph(cfg, "GET", cfg.app_id, tok, {"fields": "name"}) if err: return result(name, "FAIL", err, "Check --app-id and META_APP_TOKEN (format APP_ID|APP_SECRET).") return result(name, "PASS", "App name: %s" % data.get("name", "(no name field)")) def check_subscription(cfg, tok): name = "2. App webhook subscription (whatsapp_business_account)" if not tok: return result(name, "SKIP", "META_APP_TOKEN not set.", "export META_APP_TOKEN='APP_ID|APP_SECRET'") data, err = graph(cfg, "GET", cfg.app_id + "/subscriptions", tok) if err: return result(name, "FAIL", err, "Reading app subscriptions needs an app access token for this app.") entry = next((s for s in data.get("data", []) if s.get("object") == "whatsapp_business_account"), None) if not entry: return result(name, "FAIL", "No subscription for object whatsapp_business_account.", "The app has no WhatsApp webhook registered. n8n: activate (publish) the workflow so the " "trigger registers its production URL. Make: activate the Watch Events scenario once. " "Custom endpoint: set the callback URL and verify token in the App Dashboard " "(WhatsApp > Configuration), then subscribe to 'messages'.") problems, fixes = [], [] got, want = entry.get("callback_url", ""), cfg.expected_callback_url if norm_url(got) != norm_url(want): if TEST_PATH_HINT in got: problems.append("callback_url is an n8n TEST url: %s" % got) fixes.append("The subscription points at a /webhook-test/ URL, so production messages go nowhere. " "Do NOT run more manual tests. Activate the workflow so n8n re-registers the production " "URL, then re-run this tool. If it still shows the test URL, set the callback to the " "production URL in the App Dashboard (WhatsApp > Configuration > Edit).") else: problems.append("callback_url %s does not match expected %s" % (got or "(empty)", want)) fixes.append("The app delivers to a different URL (another workflow, scenario or old tunnel). " "Set the callback to your production URL, or fix --expected-callback-url if this one is right.") if entry.get("active") is not True: problems.append("active is %s" % entry.get("active")) fixes.append("Subscription is not active. Re-save the callback in the App Dashboard so Meta re-verifies it.") names = field_names(entry.get("fields")) if "messages" not in names: problems.append("'messages' not in fields %s" % names) fixes.append("Subscribe the 'messages' field (App Dashboard > WhatsApp > Configuration > Webhook fields).") detail = "callback_url=%s active=%s fields=%s" % (got or "(empty)", entry.get("active"), names) if problems: return result(name, "FAIL", detail + " | " + "; ".join(problems), " ".join(fixes)) return result(name, "PASS", detail) def get_waba_subscribed(cfg, tok): return graph(cfg, "GET", cfg.waba_id + "/subscribed_apps", tok) def app_in_subscribed(cfg, data): for item in data.get("data", []): inner = item.get("whatsapp_business_api_data", item) if str(inner.get("id")) == str(cfg.app_id): return True return False def check_waba_subscribed(cfg, tok): name = "3. App is in the WABA's subscribed_apps" if not tok: return result(name, "SKIP", "META_SYSTEM_TOKEN not set.", "export META_SYSTEM_TOKEN=... (whatsapp_business_management)") data, err = get_waba_subscribed(cfg, tok) if err: return result(name, "FAIL", err, "Check --waba-id and that the system user has access to this WABA.") if app_in_subscribed(cfg, data): return result(name, "PASS", "App %s is subscribed to WABA %s." % (cfg.app_id, cfg.waba_id)) return result(name, "FAIL", "App %s is NOT in subscribed_apps of WABA %s." % (cfg.app_id, cfg.waba_id), "Re-run with --apply (POST /%s/subscribed_apps). Note it subscribes the app that owns " "the system user token, so that must be app %s." % (cfg.waba_id, cfg.app_id)) def check_phone(cfg, tok): name = "4. Phone number belongs to this WABA" if not tok: return result(name, "SKIP", "META_SYSTEM_TOKEN not set.", "export META_SYSTEM_TOKEN=...") data, err = graph(cfg, "GET", cfg.waba_id + "/phone_numbers", tok, {"fields": "id,display_phone_number,verified_name,quality_rating"}) if err: return result(name, "FAIL", err, "Check --waba-id and token permissions.") for p in data.get("data", []): if str(p.get("id")) == str(cfg.phone_number_id): bits = ["display_phone_number=%s" % p.get("display_phone_number", "?")] if p.get("verified_name"): bits.append("verified_name=%s" % p["verified_name"]) if p.get("quality_rating"): bits.append("quality_rating=%s" % p["quality_rating"]) return result(name, "PASS", " ".join(bits)) more = " (list is paginated; this tool reads page 1 only)" if data.get("paging", {}).get("next") else "" ids = [str(p.get("id")) for p in data.get("data", [])] return result(name, "FAIL", "Phone number id %s not in WABA %s. Numbers there: %s%s" % (cfg.phone_number_id, cfg.waba_id, ids or "none", more), "You may be using the wrong WABA id or the wrong phone number id (not the display number). " "Messages sent to a number on a different WABA never reach this app's webhook.") def check_callback(cfg): name = "5. Callback URL answers the verify GET" if not cfg.verify_token: return result(name, "SKIP", "No --verify-token given.", "Pass --verify-token to run the hub.challenge probe.") parts = urllib.parse.urlsplit(cfg.expected_callback_url) if parts.scheme != "https": return result(name, "FAIL", "Expected callback URL is not https.", "Meta needs a public HTTPS callback URL.") q = urllib.parse.parse_qsl(parts.query) + [("hub.mode", "subscribe"), ("hub.verify_token", cfg.verify_token), ("hub.challenge", CHALLENGE)] url = urllib.parse.urlunsplit(parts._replace(query=urllib.parse.urlencode(q))) status, body = http_get(url) body = redact(body) if status == 200 and body.strip() == CHALLENGE: return result(name, "PASS", "GET %s?hub.* returned 200 with the challenge." % cfg.expected_callback_url) return result(name, "FAIL", "HTTP %s, body %r (wanted 200 and %s)." % (status, body[:80], CHALLENGE), "The endpoint is unreachable, not public, wrong path, wrong verify token, or the workflow is not " "active (n8n production URLs only answer when the workflow is active).") def manual_checks(): return [ result("6. App is in Live mode", "MANUAL", "Not read by this tool.", "App Dashboard top bar: switch to Live. In development mode only test numbers deliver (see the Make thread)."), result("7. You message the real number, not the test number", "MANUAL", "Not read by this tool.", "Send to the display number shown in check 4, from a different WhatsApp account."), ] def run_apply(cfg, tok, results): """Only POST this tool ever makes. Skipped when already subscribed.""" sub = next(r for r in results if r["name"].startswith("3.")) if sub["status"] != "FAIL" or "NOT in subscribed_apps" not in sub["detail"]: print("--apply: nothing to do (WABA subscription not reported missing).") return if not tok: print("--apply: META_SYSTEM_TOKEN is required.") return data, err = graph(cfg, "POST", cfg.waba_id + "/subscribed_apps", tok) if err: print("--apply: POST failed: " + err) return print("--apply: POST /%s/subscribed_apps -> %s" % (cfg.waba_id, json.dumps(data))) new = check_waba_subscribed(cfg, tok) results[results.index(sub)] = new print("--apply: re-check of check 3 -> " + new["status"]) # ---------------------------------------------------------------------- main def parse_args(argv): p = argparse.ArgumentParser(description="Diagnose 'WhatsApp webhook active but no messages arrive'. Read-only unless --apply.") p.add_argument("--app-id", required=True) p.add_argument("--waba-id", required=True) p.add_argument("--phone-number-id", required=True) p.add_argument("--expected-callback-url", required=True) p.add_argument("--verify-token", help="enables the hub.challenge probe") p.add_argument("--graph-version", default="v21.0", help="check the current docs for supported versions") p.add_argument("--graph-base", default="https://graph.facebook.com") p.add_argument("--mock", metavar="DIR", help="read fixtures from DIR instead of calling the network") p.add_argument("--apply", action="store_true", help="POST /{waba-id}/subscribed_apps if the app is missing") return p.parse_args(argv) def main(argv=None): global TRANSPORT cfg = parse_args(argv) app_tok = os.environ.get("META_APP_TOKEN", "") sys_tok = os.environ.get("META_SYSTEM_TOKEN", "") SECRETS[:] = [app_tok, sys_tok, cfg.verify_token or ""] if cfg.mock: TRANSPORT = MockTransport(cfg.mock, cfg.graph_base) app_tok, sys_tok = app_tok or "mock", sys_tok or "mock" print("[MOCK MODE] reading fixtures from %s, no network calls" % cfg.mock) else: TRANSPORT = live_transport results = [check_app_sanity(cfg, app_tok), check_subscription(cfg, app_tok), check_waba_subscribed(cfg, sys_tok), check_phone(cfg, sys_tok), check_callback(cfg)] if cfg.apply: run_apply(cfg, sys_tok, results) results += manual_checks() for r in results: print("[%s] %s" % (r["status"], r["name"])) print(" " + r["detail"]) if r["status"] in ("FAIL", "MANUAL") and r["fix"]: print(" FIX: " + r["fix"]) failed = [r for r in results if r["status"] == "FAIL"] if failed: print("\nFirst broken link: " + failed[0]["name"]) return 1 skipped = [r for r in results if r["status"] == "SKIP"] print("\nNo failing checks%s. Review the MANUAL items." % (" (%d skipped)" % len(skipped) if skipped else "")) return 0 if __name__ == "__main__": sys.exit(main())