use axum::body::Body;
use axum::extract::Query;
use axum::http::{HeaderMap, Request, StatusCode};
use axum::response::IntoResponse;
use chrono::{DateTime, Utc};
use hmac::{Hmac, Mac};
use serde::Deserialize;
use serde_json::{Value, json};
use sha2::Sha256;
use crate::engine::runtime::IngressEnvelope;
use crate::ingress::{
CanonicalAttachment, CanonicalButton, ProviderIds, build_canonical_payload,
canonical_session_key, default_metadata, empty_entities,
};
use crate::provider_core_only;
use crate::routing::TenantRuntimeHandle;
use crate::runner::ingress_util::{collect_body, mark_processed};
type HmacSha256 = Hmac<Sha256>;
pub async fn verify(Query(query): Query<VerifyQuery>) -> impl IntoResponse {
let expected = std::env::var("WHATSAPP_VERIFY_TOKEN").ok();
match (&query.mode, &query.challenge, &query.verify_token, expected) {
(Some(mode), Some(challenge), Some(token), Some(expected))
if mode == "subscribe" && token == &expected =>
{
(StatusCode::OK, challenge.clone())
}
_ => (StatusCode::FORBIDDEN, String::new()),
}
}
pub async fn webhook(
TenantRuntimeHandle { tenant, runtime }: TenantRuntimeHandle,
request: Request<Body>,
) -> Result<StatusCode, StatusCode> {
if provider_core_only::is_enabled() {
tracing::warn!("provider-core only mode enabled; blocking whatsapp webhook");
return Err(StatusCode::NOT_IMPLEMENTED);
}
let (parts, body) = request.into_parts();
let headers = parts.headers;
let bytes = collect_body(body).await?;
verify_signature(&headers, &bytes)?;
let raw_value: Value = serde_json::from_slice(&bytes).map_err(|_| StatusCode::BAD_REQUEST)?;
let webhook: WhatsappWebhook =
serde_json::from_value(raw_value.clone()).map_err(|_| StatusCode::BAD_REQUEST)?;
let message = webhook
.entry
.iter()
.flat_map(|entry| &entry.changes)
.find_map(|change| {
change
.value
.messages
.as_ref()
.and_then(|msgs| msgs.first().cloned())
});
if message.is_none() {
return Ok(StatusCode::OK);
}
let message = message.unwrap();
if mark_processed(runtime.webhook_cache(), &message.id) {
return Ok(StatusCode::ACCEPTED);
}
let flow = runtime
.engine()
.flow_by_type("messaging")
.ok_or(StatusCode::NOT_FOUND)?;
let provider_ids = ProviderIds {
conversation_id: Some(message.from.clone()),
user_id: Some(message.from.clone()),
message_id: Some(message.id.clone()),
..ProviderIds::default()
};
let session_key = canonical_session_key(&tenant, "whatsapp", &provider_ids);
let timestamp = parse_timestamp(message.timestamp.as_deref())?;
let (text, attachments, buttons, scopes) = map_message_content(&message);
let canonical_payload = build_canonical_payload(
&tenant,
"whatsapp",
&provider_ids,
session_key.clone(),
&scopes,
timestamp,
None,
text,
attachments,
buttons,
empty_entities(),
default_metadata(),
json!({ "type": message.r#type }),
raw_value,
);
let envelope = IngressEnvelope {
tenant,
env: None,
pack_id: Some(flow.pack_id.clone()),
flow_id: flow.id.clone(),
flow_type: Some(flow.flow_type.clone()),
action: Some("messaging".into()),
session_hint: Some(session_key),
provider: Some("whatsapp".into()),
channel: Some(message.from.clone()),
conversation: Some(message.from.clone()),
user: Some(message.from.clone()),
activity_id: Some(message.id.clone()),
timestamp: Some(timestamp.to_rfc3339()),
payload: canonical_payload,
metadata: None,
reply_scope: None,
}
.canonicalize();
runtime
.state_machine()
.handle(envelope)
.await
.map_err(|err| {
tracing::error!(error = %err, "whatsapp flow execution failed");
StatusCode::BAD_GATEWAY
})?;
Ok(StatusCode::ACCEPTED)
}
fn map_message_content(
message: &WhatsappMessage,
) -> (Option<String>, Vec<Value>, Vec<Value>, Vec<String>) {
let mut attachments = Vec::new();
let mut buttons = Vec::new();
let mut scopes = vec!["chat".to_string()];
let mut text = message.text.as_ref().map(|text| text.body.clone());
match message.r#type.as_str() {
"image" => {
if let Some(image) = &message.image {
attachments.push(
CanonicalAttachment {
attachment_type: "image".into(),
name: image.caption.clone(),
mime: None,
size: None,
url: image.link.clone(),
data_inline_b64: None,
}
.into_value(),
);
}
}
"audio" => {
if let Some(audio) = &message.audio {
attachments.push(
CanonicalAttachment {
attachment_type: "audio".into(),
name: None,
mime: None,
size: None,
url: audio.link.clone(),
data_inline_b64: None,
}
.into_value(),
);
}
}
"video" => {
if let Some(video) = &message.video {
attachments.push(
CanonicalAttachment {
attachment_type: "video".into(),
name: video.caption.clone(),
mime: None,
size: None,
url: video.link.clone(),
data_inline_b64: None,
}
.into_value(),
);
}
}
"document" => {
if let Some(doc) = &message.document {
attachments.push(
CanonicalAttachment {
attachment_type: "file".into(),
name: doc.filename.clone(),
mime: None,
size: None,
url: doc.link.clone(),
data_inline_b64: None,
}
.into_value(),
);
}
}
"location" => {
if let Some(location) = &message.location {
text = Some(location.name.clone().unwrap_or_else(|| "location".into()));
}
}
"interactive" => {
if let Some(interactive) = &message.interactive {
if let Some(reply) = &interactive.button_reply {
buttons.push(
CanonicalButton {
id: reply.id.clone(),
title: reply.title.clone(),
payload: reply.id.clone(),
}
.into_value(),
);
text = Some(reply.title.clone());
} else if let Some(list) = &interactive.list_reply {
buttons.push(
CanonicalButton {
id: list.id.clone(),
title: list.title.clone(),
payload: list.id.clone(),
}
.into_value(),
);
text = Some(list.title.clone());
}
}
}
_ => {}
}
if !attachments.is_empty() {
scopes.push("attachments".into());
}
if !buttons.is_empty() {
scopes.push("buttons".into());
}
(text, attachments, buttons, scopes)
}
fn parse_timestamp(raw: Option<&str>) -> Result<DateTime<Utc>, StatusCode> {
if let Some(epoch) = raw.and_then(|value| value.parse::<i64>().ok()) {
return DateTime::from_timestamp(epoch, 0).ok_or(StatusCode::BAD_REQUEST);
}
Ok(Utc::now())
}
fn verify_signature(headers: &HeaderMap, body: &[u8]) -> Result<(), StatusCode> {
if let Ok(secret) = std::env::var("WHATSAPP_APP_SECRET") {
let signature = headers
.get("X-Hub-Signature-256")
.and_then(|value| value.to_str().ok())
.and_then(|value| value.strip_prefix("sha256="))
.ok_or(StatusCode::UNAUTHORIZED)?;
let mut mac =
HmacSha256::new_from_slice(secret.as_bytes()).map_err(|_| StatusCode::UNAUTHORIZED)?;
mac.update(body);
let expected = hex::encode(mac.finalize().into_bytes());
if !subtle_equals(signature, &expected) {
return Err(StatusCode::UNAUTHORIZED);
}
}
Ok(())
}
fn subtle_equals(a: &str, b: &str) -> bool {
if a.len() != b.len() {
return false;
}
let mut diff = 0u8;
for (x, y) in a.as_bytes().iter().zip(b.as_bytes()) {
diff |= x ^ y;
}
diff == 0
}
#[derive(Debug, Deserialize)]
pub struct VerifyQuery {
#[serde(rename = "hub.mode")]
mode: Option<String>,
#[serde(rename = "hub.challenge")]
challenge: Option<String>,
#[serde(rename = "hub.verify_token")]
verify_token: Option<String>,
}
#[derive(Debug, Deserialize)]
struct WhatsappWebhook {
entry: Vec<WhatsappEntry>,
}
#[derive(Debug, Deserialize)]
struct WhatsappEntry {
changes: Vec<WhatsappChange>,
}
#[derive(Debug, Deserialize)]
struct WhatsappChange {
value: WhatsappValue,
}
#[derive(Debug, Deserialize)]
struct WhatsappValue {
#[serde(default)]
messages: Option<Vec<WhatsappMessage>>,
}
#[derive(Debug, Clone, Deserialize)]
struct WhatsappMessage {
id: String,
#[serde(default)]
from: String,
#[serde(rename = "type")]
r#type: String,
#[serde(default)]
timestamp: Option<String>,
#[serde(default)]
text: Option<WhatsappText>,
#[serde(default)]
image: Option<WhatsappMedia>,
#[serde(default)]
audio: Option<WhatsappMedia>,
#[serde(default)]
video: Option<WhatsappMedia>,
#[serde(default)]
document: Option<WhatsappDocument>,
#[serde(default)]
location: Option<WhatsappLocation>,
#[serde(default)]
interactive: Option<WhatsappInteractive>,
}
#[derive(Debug, Clone, Deserialize)]
struct WhatsappText {
body: String,
}
#[derive(Debug, Clone, Deserialize)]
struct WhatsappMedia {
#[serde(default)]
link: Option<String>,
#[serde(default)]
caption: Option<String>,
}
#[derive(Debug, Clone, Deserialize)]
struct WhatsappDocument {
#[serde(default)]
link: Option<String>,
#[serde(default)]
filename: Option<String>,
}
#[derive(Debug, Clone, Deserialize)]
struct WhatsappLocation {
#[serde(default)]
name: Option<String>,
}
#[derive(Debug, Clone, Deserialize)]
struct WhatsappInteractive {
#[serde(default)]
button_reply: Option<WhatsappButtonReply>,
#[serde(default)]
list_reply: Option<WhatsappListReply>,
}
#[derive(Debug, Clone, Deserialize)]
struct WhatsappButtonReply {
id: String,
title: String,
}
#[derive(Debug, Clone, Deserialize)]
struct WhatsappListReply {
id: String,
title: String,
}
#[cfg(test)]
mod tests {
use super::*;
use crate::routing::TenantRuntimeHandle;
use axum::body::Body;
use axum::extract::Query;
use axum::http::Request;
use axum::response::IntoResponse;
use serde_json::json;
#[test]
fn whatsapp_message_maps_to_canonical_payload() {
let raw = json!({
"entry": [{
"changes": [{
"value": {
"messages": [{
"id": "wamid.HBgM",
"from": "447700900123",
"timestamp": "1731315600",
"type": "text",
"text": { "body": "Hi" }
}]
}
}]
}]
});
let webhook: WhatsappWebhook = serde_json::from_value(raw.clone()).unwrap();
let message = webhook.entry[0].changes[0].value.messages.as_ref().unwrap()[0].clone();
let provider_ids = ProviderIds {
conversation_id: Some(message.from.clone()),
user_id: Some(message.from.clone()),
message_id: Some(message.id.clone()),
..ProviderIds::default()
};
let session_key = canonical_session_key("zain-kuwait", "whatsapp", &provider_ids);
assert_eq!(
session_key,
"zain-kuwait:whatsapp:447700900123:447700900123"
);
let timestamp = parse_timestamp(message.timestamp.as_deref()).unwrap();
let (text, attachments, buttons, scopes) = map_message_content(&message);
let canonical = build_canonical_payload(
"zain-kuwait",
"whatsapp",
&provider_ids,
session_key,
&scopes,
timestamp,
None,
text,
attachments,
buttons,
empty_entities(),
default_metadata(),
json!({ "type": message.r#type }),
raw,
);
assert_eq!(canonical["provider"], json!("whatsapp"));
assert_eq!(
canonical["session"]["key"],
json!("zain-kuwait:whatsapp:447700900123:447700900123")
);
assert_eq!(canonical["text"], json!("Hi"));
assert_eq!(canonical["attachments"], json!([]));
assert_eq!(canonical["buttons"], json!([]));
}
#[test]
fn whatsapp_maps_media_location_and_interactive_content() {
let image: WhatsappMessage = serde_json::from_value(json!({
"id": "1",
"from": "447700900123",
"type": "image",
"image": { "link": "https://example.com/image.png", "caption": "photo" }
}))
.unwrap();
let (text, attachments, buttons, scopes) = map_message_content(&image);
assert_eq!(text, None);
assert_eq!(attachments[0]["type"], json!("image"));
assert!(buttons.is_empty());
assert!(scopes.contains(&"attachments".to_string()));
let list_reply: WhatsappMessage = serde_json::from_value(json!({
"id": "2",
"from": "447700900123",
"type": "interactive",
"interactive": {
"list_reply": { "id": "choice-1", "title": "Choice" }
}
}))
.unwrap();
let (text, _attachments, buttons, scopes) = map_message_content(&list_reply);
assert_eq!(text.as_deref(), Some("Choice"));
assert_eq!(buttons[0]["payload"], json!("choice-1"));
assert!(scopes.contains(&"buttons".to_string()));
let location: WhatsappMessage = serde_json::from_value(json!({
"id": "3",
"from": "447700900123",
"type": "location",
"location": { "name": "HQ" }
}))
.unwrap();
let (text, _, _, _) = map_message_content(&location);
assert_eq!(text.as_deref(), Some("HQ"));
}
#[test]
fn whatsapp_timestamp_and_compare_helpers_are_stable() {
assert!(parse_timestamp(Some("not-epoch")).is_ok());
assert!(!subtle_equals("abc", "abd"));
}
#[tokio::test]
async fn whatsapp_verify_accepts_and_rejects_tokens() {
let ok = verify(Query(VerifyQuery {
mode: Some("subscribe".into()),
challenge: Some("12345".into()),
verify_token: Some("token".into()),
}))
.await
.into_response();
assert_eq!(ok.status(), StatusCode::FORBIDDEN);
let bad = verify(Query(VerifyQuery {
mode: Some("other".into()),
challenge: Some("12345".into()),
verify_token: Some("token".into()),
}))
.await
.into_response();
assert_eq!(bad.status(), StatusCode::FORBIDDEN);
}
#[test]
fn whatsapp_maps_document_audio_and_button_reply() {
let document: WhatsappMessage = serde_json::from_value(json!({
"id": "4",
"from": "447700900123",
"type": "document",
"document": { "link": "https://example.com/file.pdf", "filename": "file.pdf" }
}))
.unwrap();
let (_text, attachments, _buttons, scopes) = map_message_content(&document);
assert_eq!(attachments[0]["type"], json!("file"));
assert!(scopes.contains(&"attachments".to_string()));
let button: WhatsappMessage = serde_json::from_value(json!({
"id": "5",
"from": "447700900123",
"type": "interactive",
"interactive": {
"button_reply": { "id": "yes", "title": "Yes" }
}
}))
.unwrap();
let (text, _attachments, buttons, scopes) = map_message_content(&button);
assert_eq!(text.as_deref(), Some("Yes"));
assert_eq!(buttons[0]["id"], json!("yes"));
assert!(scopes.contains(&"buttons".to_string()));
}
#[tokio::test]
async fn whatsapp_webhook_returns_not_implemented_in_provider_core_only_mode() {
let (_workspace, runtime) = crate::test_support::build_test_runtime()
.await
.expect("runtime");
let request = Request::builder().body(Body::empty()).unwrap();
let status = webhook(
TenantRuntimeHandle {
tenant: "demo".into(),
runtime,
},
request,
)
.await
.expect_err("provider-core only should block legacy webhook");
assert_eq!(status, StatusCode::NOT_IMPLEMENTED);
}
}