1  """WhatsApp Business Cloud API webhook tests.
  2  
  3  Covers:
  4  * GET subscription handshake — happy path + bad verify_token → 403.
  5  * POST signature verification — bad sig → 401.
  6  * POST routing by phone_number_id — unknown id → 200 ack, no agent run.
  7  * Allowlist gate — sender outside allowlist → 200 ack, "not authorized"
  8    reply queued.
  9  * Non-text message → 200 ack, polite "text only" reply queued.
 10  
 11  The agent dispatch is patched out in every test that exercises a
 12  project-bound message because we don't want to spin up a real LLM, and
 13  because TestClient's BackgroundTasks runs synchronously after the
 14  response — letting the real chat pipeline run would either blow up on
 15  missing LLM config or take seconds.
 16  """
 17  from __future__ import annotations
 18  
 19  import hashlib
 20  import hmac
 21  import json
 22  
 23  import pytest
 24  from fastapi.testclient import TestClient
 25  
 26  from restai.config import RESTAI_DEFAULT_PASSWORD
 27  from restai.database import get_db_wrapper
 28  from restai.main import app
 29  from restai.utils.crypto import encrypt_field
 30  
 31  
 32  AUTH = ("admin", RESTAI_DEFAULT_PASSWORD)
 33  
 34  # Constants used in the fake project we set up. These are intentionally
 35  # distinctive so a stale row from a prior run is easy to spot in the DB.
 36  PHONE_NUMBER_ID = "wa_test_phone_999000111"
 37  ACCESS_TOKEN = "wa_test_access_token_xyz"
 38  APP_SECRET = "wa_test_app_secret_super_random"
 39  VERIFY_TOKEN = "wa_test_verify_token_abc"
 40  ALLOWED_SENDER = "351900000001"
 41  BLOCKED_SENDER = "999999999999"
 42  
 43  
 44  @pytest.fixture(scope="module")
 45  def client():
 46      with TestClient(app) as c:
 47          yield c
 48  
 49  
 50  @pytest.fixture(scope="module")
 51  def project_id(client):
 52      """Create (or find) an agent project, then write WhatsApp options
 53      directly into the DB so we don't depend on the project create API
 54      accepting our new keys (it does, but bypassing keeps the test
 55      hermetic to model-validator changes)."""
 56      teams = client.get("/teams", auth=AUTH).json().get("teams", []) or []
 57      if not teams:
 58          pytest.skip("no team available to bootstrap project")
 59      team_id = teams[0]["id"]
 60  
 61      name = "whatsapp_webhook_test_project"
 62      # Find existing by name via the list endpoint (per-id GET requires int).
 63      listing = client.get("/projects", auth=AUTH).json().get("projects", []) or []
 64      existing = next((p for p in listing if p.get("name") == name), None)
 65      if existing:
 66          pid = existing["id"]
 67      else:
 68          # Pick whatever LLM is registered. In CI / fresh dev DB the
 69          # default seed includes one; if not, skip cleanly.
 70          info = client.get("/info", auth=AUTH).json()
 71          llms = info.get("llms") or []
 72          llm_name = llms[0].get("name") if llms else None
 73          if not llm_name:
 74              pytest.skip("no LLMs configured — cannot bootstrap project")
 75          resp = client.post(
 76              "/projects",
 77              json={"name": name, "type": "agent", "llm": llm_name, "team_id": team_id},
 78              auth=AUTH,
 79          )
 80          if resp.status_code not in (200, 201):
 81              pytest.skip(f"could not create test project: {resp.status_code} {resp.text}")
 82          pid = resp.json()["id"]
 83  
 84      # Write WhatsApp options directly: secrets must be encrypted at rest
 85      # because the webhook handler decrypts them on the way out.
 86      db = get_db_wrapper()
 87      try:
 88          from restai.models.databasemodels import ProjectDatabase
 89          proj = db.db.query(ProjectDatabase).filter(ProjectDatabase.id == pid).first()
 90          opts = json.loads(proj.options or "{}")
 91          opts.update({
 92              "whatsapp_phone_number_id": PHONE_NUMBER_ID,
 93              "whatsapp_access_token": encrypt_field(ACCESS_TOKEN),
 94              "whatsapp_app_secret": encrypt_field(APP_SECRET),
 95              "whatsapp_verify_token": encrypt_field(VERIFY_TOKEN),
 96              "whatsapp_allowed_phone_numbers": ALLOWED_SENDER,
 97          })
 98          proj.options = json.dumps(opts)
 99          db.db.commit()
100      finally:
101          db.db.close()
102  
103      return pid
104  
105  
106  def _sign(body: bytes, secret: str = APP_SECRET) -> str:
107      return "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
108  
109  
110  def _make_payload(from_phone: str = ALLOWED_SENDER, msg_type: str = "text",
111                     text: str = "hello bot") -> bytes:
112      msg = {"from": from_phone, "id": "wamid.test", "type": msg_type}
113      if msg_type == "text":
114          msg["text"] = {"body": text}
115      return json.dumps({
116          "object": "whatsapp_business_account",
117          "entry": [{
118              "id": "biz_account_id",
119              "changes": [{
120                  "field": "messages",
121                  "value": {
122                      "messaging_product": "whatsapp",
123                      "metadata": {"phone_number_id": PHONE_NUMBER_ID},
124                      "messages": [msg],
125                  },
126              }],
127          }],
128      }).encode()
129  
130  
131  # ─── GET handshake ──────────────────────────────────────────────────────
132  
133  def test_verify_handshake_ok(client, project_id):
134      r = client.get("/webhooks/whatsapp", params={
135          "hub.mode": "subscribe",
136          "hub.challenge": "challenge_value_42",
137          "hub.verify_token": VERIFY_TOKEN,
138      })
139      assert r.status_code == 200, r.text
140      assert r.text == "challenge_value_42"
141  
142  
143  def test_verify_handshake_bad_token(client, project_id):
144      r = client.get("/webhooks/whatsapp", params={
145          "hub.mode": "subscribe",
146          "hub.challenge": "challenge_value_42",
147          "hub.verify_token": "definitely_wrong_token",
148      })
149      assert r.status_code == 403
150  
151  
152  def test_verify_handshake_missing_params(client):
153      r = client.get("/webhooks/whatsapp", params={"hub.mode": "subscribe"})
154      assert r.status_code == 400
155  
156  
157  # ─── POST routing & signature ───────────────────────────────────────────
158  
159  def test_post_unknown_phone_number_id_acks_silently(client):
160      """If no project owns the phone_number_id, signature is never checked
161      (we don't have a secret to check it with). Still ack 200 so Meta
162      doesn't retry — it's an inbound for a project that was deleted."""
163      body = json.dumps({
164          "object": "whatsapp_business_account",
165          "entry": [{"changes": [{"value": {
166              "messaging_product": "whatsapp",
167              "metadata": {"phone_number_id": "phone_id_we_do_not_know"},
168              "messages": [{"from": ALLOWED_SENDER, "type": "text", "text": {"body": "hi"}}],
169          }}]}],
170      }).encode()
171      r = client.post(
172          "/webhooks/whatsapp",
173          content=body,
174          headers={"X-Hub-Signature-256": "sha256=00", "content-type": "application/json"},
175      )
176      assert r.status_code == 200
177      assert r.json() == {"status": "ok"}
178  
179  
180  def test_post_bad_signature_returns_401(client, project_id, monkeypatch):
181      body = _make_payload()
182      sent_messages = []
183      monkeypatch.setattr(
184          "restai.routers.whatsapp_webhook.send_message",
185          lambda *a, **kw: sent_messages.append(a) or {"ok": True},
186      )
187  
188      r = client.post(
189          "/webhooks/whatsapp",
190          content=body,
191          headers={"X-Hub-Signature-256": "sha256=tampered_sig", "content-type": "application/json"},
192      )
193      assert r.status_code == 401
194      assert sent_messages == [], "no outbound on bad sig"
195  
196  
197  def test_post_text_message_dispatches_agent(client, project_id, monkeypatch):
198      """Valid signature + allowlisted sender + text → agent runs and
199      reply is sent. We mock both the agent dispatch and the outbound
200      send so the test stays hermetic."""
201      sent_messages = []
202      monkeypatch.setattr(
203          "restai.routers.whatsapp_webhook.send_message",
204          lambda token, pid, to, text: sent_messages.append((token, pid, to, text)) or {"ok": True},
205      )
206  
207      async def fake_run_agent(project_id, text, from_phone):
208          return f"echo: {text}"
209      monkeypatch.setattr(
210          "restai.routers.whatsapp_webhook._run_agent",
211          fake_run_agent,
212      )
213  
214      body = _make_payload(text="hello bot")
215      sig = _sign(body)
216      r = client.post(
217          "/webhooks/whatsapp",
218          content=body,
219          headers={"X-Hub-Signature-256": sig, "content-type": "application/json"},
220      )
221      assert r.status_code == 200
222      assert sent_messages, "agent reply should have been queued"
223      token, pid, to, reply = sent_messages[-1]
224      assert token == ACCESS_TOKEN
225      assert pid == PHONE_NUMBER_ID
226      assert to == ALLOWED_SENDER
227      assert reply == "echo: hello bot"
228  
229  
230  def test_post_blocked_sender_replies_unauthorized(client, project_id, monkeypatch):
231      sent_messages = []
232      monkeypatch.setattr(
233          "restai.routers.whatsapp_webhook.send_message",
234          lambda token, pid, to, text: sent_messages.append((to, text)) or {"ok": True},
235      )
236      agent_called = []
237      async def fake_run_agent(project_id, text, from_phone):
238          agent_called.append(from_phone)
239          return "should not happen"
240      monkeypatch.setattr("restai.routers.whatsapp_webhook._run_agent", fake_run_agent)
241  
242      body = _make_payload(from_phone=BLOCKED_SENDER, text="hello")
243      sig = _sign(body)
244      r = client.post(
245          "/webhooks/whatsapp",
246          content=body,
247          headers={"X-Hub-Signature-256": sig, "content-type": "application/json"},
248      )
249      assert r.status_code == 200
250      assert agent_called == [], "agent must not run for blocked senders"
251      assert sent_messages, "polite reject should be queued"
252      to, reply = sent_messages[-1]
253      assert to == BLOCKED_SENDER
254      assert "not authorized" in reply.lower()
255  
256  
257  def test_post_non_text_message_replies_text_only_notice(client, project_id, monkeypatch):
258      sent_messages = []
259      monkeypatch.setattr(
260          "restai.routers.whatsapp_webhook.send_message",
261          lambda token, pid, to, text: sent_messages.append((to, text)) or {"ok": True},
262      )
263      agent_called = []
264      async def fake_run_agent(project_id, text, from_phone):
265          agent_called.append(from_phone)
266          return "should not happen"
267      monkeypatch.setattr("restai.routers.whatsapp_webhook._run_agent", fake_run_agent)
268  
269      body = _make_payload(msg_type="image")
270      sig = _sign(body)
271      r = client.post(
272          "/webhooks/whatsapp",
273          content=body,
274          headers={"X-Hub-Signature-256": sig, "content-type": "application/json"},
275      )
276      assert r.status_code == 200
277      assert agent_called == [], "agent must not run for non-text messages"
278      assert sent_messages, "text-only notice should be queued"
279      to, reply = sent_messages[-1]
280      assert to == ALLOWED_SENDER
281      assert "text" in reply.lower()
282  
283  
284  # ─── verify_signature unit tests (no network) ──────────────────────────
285  
286  def test_verify_signature_rejects_missing_header():
287      from restai.whatsapp import verify_signature
288      assert verify_signature(b"body", None, "secret") is False
289      assert verify_signature(b"body", "", "secret") is False
290  
291  
292  def test_verify_signature_rejects_bad_prefix():
293      from restai.whatsapp import verify_signature
294      assert verify_signature(b"body", "md5=abcd", "secret") is False
295  
296  
297  def test_verify_signature_accepts_valid_hmac():
298      from restai.whatsapp import verify_signature
299      body = b'{"hello":"world"}'
300      sig = "sha256=" + hmac.new(b"secret", body, hashlib.sha256).hexdigest()
301      assert verify_signature(body, sig, "secret") is True
302  
303  
304  def test_verify_signature_constant_time_mismatch():
305      from restai.whatsapp import verify_signature
306      body = b'{"hello":"world"}'
307      sig = "sha256=" + hmac.new(b"different_key", body, hashlib.sha256).hexdigest()
308      assert verify_signature(body, sig, "secret") is False