Release Candidate v3.10.1 (#197)

Author: milosmns

.version

@@ -1 +1 @@
-3.10.0
+3.10.1

src/api/auth.py

@@ -1,3 +1,5 @@
+import hashlib
+import hmac
 from datetime import datetime, timedelta, timezone
 from typing import Any, Dict
 
@@ -13,7 +15,6 @@
 
 api_key_header = APIKeyHeader(name = "X-API-Key", auto_error = True)
 telegram_auth_key_header = APIKeyHeader(name = "X-Telegram-Bot-Api-Secret-Token", auto_error = False)
-whatsapp_auth_key_header = APIKeyHeader(name = "X-WhatsApp-Bot-Api-Secret-Token", auto_error = False)
 jwt_header = HTTPBearer(bearerFormat = "JWT", auto_error = True)
 
 
@@ -29,10 +30,35 @@ def verify_telegram_auth_key(auth_key: str = Security(telegram_auth_key_header))
     return auth_key
 
 
-def verify_whatsapp_auth_key(auth_key: str = Security(whatsapp_auth_key_header)) -> str:
-    if config.whatsapp_must_auth and auth_key != config.whatsapp_auth_key.get_secret_value():
-        raise HTTPException(status_code = HTTP_403_FORBIDDEN, detail = "Could not validate the WhatsApp auth token")
-    return auth_key
+def verify_whatsapp_webhook_challenge(mode: str, challenge: str, verify_token: str) -> str:
+    if not config.whatsapp_must_auth:
+        log.i("WhatsApp webhook verification skipped (auth disabled)")
+        return challenge
+    token_valid = verify_token == config.whatsapp_auth_key.get_secret_value()
+    if mode == "subscribe" and token_valid:
+        log.i("WhatsApp webhook verified successfully")
+        return challenge
+    else:
+        log.w(f"WhatsApp webhook verification failed: mode={mode}, token_match={token_valid}")
+        raise HTTPException(status_code = HTTP_403_FORBIDDEN, detail = "Webhook verification failed")
+
+
+def verify_whatsapp_signature(payload: bytes, signature_header: str | None) -> None:
+    if not config.whatsapp_must_auth:
+        log.i("WhatsApp signature verification skipped (auth disabled)")
+        return
+    if not signature_header:
+        log.w("WhatsApp signature verification failed: missing X-Hub-Signature-256 header")
+        raise HTTPException(status_code = HTTP_403_FORBIDDEN, detail = "Missing signature header")
+    if not signature_header.startswith("sha256="):
+        log.w(f"WhatsApp signature verification failed: invalid header format: {signature_header}")
+        raise HTTPException(status_code = HTTP_403_FORBIDDEN, detail = "Invalid signature format")
+    received_signature = signature_header[7:]
+    expected_signature = hmac.new(config.whatsapp_app_secret.get_secret_value().encode(), payload, hashlib.sha256).hexdigest()
+    if not hmac.compare_digest(expected_signature, received_signature):
+        log.w("WhatsApp signature verification failed: signature mismatch")
+        raise HTTPException(status_code = HTTP_403_FORBIDDEN, detail = "Invalid signature")
+    log.i("WhatsApp signature verified successfully")
 
 
 def verify_jwt_credentials(authorization: HTTPAuthorizationCredentials = Security(jwt_header)) -> Dict[str, Any]:

src/main.py

@@ -9,7 +9,7 @@
 from uuid import UUID
 
 import uvicorn
-from fastapi import BackgroundTasks, Depends, FastAPI, HTTPException
+from fastapi import BackgroundTasks, Depends, FastAPI, Header, HTTPException, Query, Request
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import SecretStr
 from starlette.responses import RedirectResponse
@@ -20,7 +20,8 @@
     verify_api_key,
     verify_jwt_credentials,
     verify_telegram_auth_key,
-    verify_whatsapp_auth_key,
+    verify_whatsapp_signature,
+    verify_whatsapp_webhook_challenge,
 )
 from api.model.chat_settings_payload import ChatSettingsPayload
 from api.model.release_output_payload import ReleaseOutputPayload
@@ -90,12 +91,24 @@ async def telegram_chat_update(
     return {"status": "ok"}
 
 
+@app.get("/whatsapp/chat-update")
+async def whatsapp_webhook_verification(
+    hub_mode: str = Query(alias = "hub.mode"),
+    hub_challenge: str = Query(alias = "hub.challenge"),
+    hub_verify_token: str = Query(alias = "hub.verify_token"),
+) -> str:
+    return verify_whatsapp_webhook_challenge(hub_mode, hub_challenge, hub_verify_token)
+
+
 @app.post("/whatsapp/chat-update")
 async def whatsapp_chat_update(
-    update: dict,
-    _ = Depends(verify_whatsapp_auth_key),
+    request: Request,
+    x_hub_signature_256: str | None = Header(default = None),
 ) -> dict:
-    log.d("Received WhatsApp update", update)
+    body = await request.body()
+    verify_whatsapp_signature(body, x_hub_signature_256)
+    update = await request.json()
+    log.d(f"Received WhatsApp update: {update}")
     return {"status": "ok"}
 
 

src/util/config.py

@@ -47,6 +47,7 @@ class Config(metaclass = Singleton):
     telegram_auth_key: SecretStr
     telegram_bot_token: SecretStr
     whatsapp_auth_key: SecretStr
+    whatsapp_app_secret: SecretStr
     jwt_secret_key: SecretStr
     github_issues_token: SecretStr
     rapid_api_twitter_token: SecretStr
@@ -60,6 +61,7 @@ def all_secrets(self) -> list[SecretStr]:
             self.telegram_auth_key,
             self.telegram_bot_token,
             self.whatsapp_auth_key,
+            self.whatsapp_app_secret,
             self.jwt_secret_key,
             self.github_issues_token,
             self.rapid_api_twitter_token,
@@ -107,6 +109,7 @@ def __init__(
         def_telegram_auth_key: SecretStr = SecretStr("it_is_really_telegram"),
         def_telegram_bot_token: SecretStr = SecretStr("invalid"),
         def_whatsapp_auth_key: SecretStr = SecretStr("it_is_really_whatsapp"),
+        def_whatsapp_app_secret: SecretStr = SecretStr("invalid"),
         def_jwt_secret_key: SecretStr = SecretStr("default"),
         def_github_issues_token: SecretStr = SecretStr("invalid"),
         def_rapid_api_twitter_token: SecretStr = SecretStr("invalid"),
@@ -149,6 +152,7 @@ def __init__(
         self.telegram_auth_key = self.__senv("TELEGRAM_API_UPDATE_AUTH_TOKEN", lambda: def_telegram_auth_key)
         self.telegram_bot_token = self.__senv("TELEGRAM_BOT_TOKEN", lambda: def_telegram_bot_token)
         self.whatsapp_auth_key = self.__senv("WHATSAPP_API_UPDATE_AUTH_TOKEN", lambda: def_whatsapp_auth_key)
+        self.whatsapp_app_secret = self.__senv("WHATSAPP_APP_SECRET", lambda: def_whatsapp_app_secret)
         self.jwt_secret_key = self.__senv("JWT_SECRET_KEY", lambda: def_jwt_secret_key)
         self.github_issues_token = self.__senv("THE_AGENT_ISSUES_TOKEN", lambda: def_github_issues_token)
         self.rapid_api_twitter_token = self.__senv("RAPID_API_TWITTER_TOKEN", lambda: def_rapid_api_twitter_token)

test/api/test_auth.py

@@ -14,6 +14,8 @@
     verify_api_key,
     verify_jwt_credentials,
     verify_telegram_auth_key,
+    verify_whatsapp_signature,
+    verify_whatsapp_webhook_challenge,
 )
 from util.config import config
 
@@ -138,3 +140,74 @@ def test_get_chat_type_from_jwt_missing_platform(self):
         claims = {"sub": "user-123"}
         chat_type = get_chat_type_from_jwt(claims)
         self.assertIsNone(chat_type)
+
+    @patch("api.auth.config")
+    def test_whatsapp_webhook_challenge_success(self, mock_config: MagicMock):
+        mock_config.whatsapp_must_auth = True
+        mock_config.whatsapp_auth_key = SecretStr("test-token")
+        challenge = verify_whatsapp_webhook_challenge("subscribe", "test-challenge", "test-token")
+        self.assertEqual(challenge, "test-challenge")
+
+    @patch("api.auth.config")
+    def test_whatsapp_webhook_challenge_invalid_token(self, mock_config: MagicMock):
+        mock_config.whatsapp_must_auth = True
+        mock_config.whatsapp_auth_key = SecretStr("correct-token")
+        with self.assertRaises(HTTPException) as context:
+            verify_whatsapp_webhook_challenge("subscribe", "test-challenge", "wrong-token")
+        self.assertEqual(context.exception.status_code, HTTP_403_FORBIDDEN)
+        self.assertEqual(context.exception.detail, "Webhook verification failed")
+
+    @patch("api.auth.config")
+    def test_whatsapp_webhook_challenge_invalid_mode(self, mock_config: MagicMock):
+        mock_config.whatsapp_must_auth = True
+        mock_config.whatsapp_auth_key = SecretStr("test-token")
+        with self.assertRaises(HTTPException) as context:
+            verify_whatsapp_webhook_challenge("unsubscribe", "test-challenge", "test-token")
+        self.assertEqual(context.exception.status_code, HTTP_403_FORBIDDEN)
+        self.assertEqual(context.exception.detail, "Webhook verification failed")
+
+    def test_whatsapp_webhook_challenge_auth_disabled(self):
+        challenge = verify_whatsapp_webhook_challenge("subscribe", "test-challenge", "any-token")
+        self.assertEqual(challenge, "test-challenge")
+
+    @patch("api.auth.config")
+    def test_whatsapp_signature_verification_success(self, mock_config: MagicMock):
+        import hashlib
+        import hmac
+        mock_config.whatsapp_must_auth = True
+        mock_config.whatsapp_app_secret = SecretStr("test-secret")
+        payload = b'{"test": "data"}'
+        signature = hmac.new(b"test-secret", payload, hashlib.sha256).hexdigest()
+        verify_whatsapp_signature(payload, f"sha256={signature}")
+
+    @patch("api.auth.config")
+    def test_whatsapp_signature_verification_invalid_signature(self, mock_config: MagicMock):
+        mock_config.whatsapp_must_auth = True
+        mock_config.whatsapp_app_secret = SecretStr("test-secret")
+        payload = b'{"test": "data"}'
+        with self.assertRaises(HTTPException) as context:
+            verify_whatsapp_signature(payload, "sha256=wrong-signature")
+        self.assertEqual(context.exception.status_code, HTTP_403_FORBIDDEN)
+        self.assertEqual(context.exception.detail, "Invalid signature")
+
+    @patch("api.auth.config")
+    def test_whatsapp_signature_verification_missing_header(self, mock_config: MagicMock):
+        mock_config.whatsapp_must_auth = True
+        payload = b'{"test": "data"}'
+        with self.assertRaises(HTTPException) as context:
+            verify_whatsapp_signature(payload, None)
+        self.assertEqual(context.exception.status_code, HTTP_403_FORBIDDEN)
+        self.assertEqual(context.exception.detail, "Missing signature header")
+
+    @patch("api.auth.config")
+    def test_whatsapp_signature_verification_invalid_format(self, mock_config: MagicMock):
+        mock_config.whatsapp_must_auth = True
+        payload = b'{"test": "data"}'
+        with self.assertRaises(HTTPException) as context:
+            verify_whatsapp_signature(payload, "invalid-format")
+        self.assertEqual(context.exception.status_code, HTTP_403_FORBIDDEN)
+        self.assertEqual(context.exception.detail, "Invalid signature format")
+
+    def test_whatsapp_signature_verification_auth_disabled(self):
+        payload = b'{"test": "data"}'
+        verify_whatsapp_signature(payload, None)

test/util/test_config.py

@@ -57,6 +57,7 @@ def test_default_config(self):
         self.assertEqual(config.telegram_auth_key.get_secret_value(), "it_is_really_telegram")
         self.assertEqual(config.telegram_bot_token.get_secret_value(), "invalid")
         self.assertEqual(config.whatsapp_auth_key.get_secret_value(), "it_is_really_whatsapp")
+        self.assertEqual(config.whatsapp_app_secret.get_secret_value(), "invalid")
         self.assertEqual(config.jwt_secret_key.get_secret_value(), "default")
         self.assertEqual(config.github_issues_token.get_secret_value(), "invalid")
         self.assertEqual(config.rapid_api_twitter_token.get_secret_value(), "invalid")
@@ -102,6 +103,7 @@ def test_custom_config(self):
         os.environ["TELEGRAM_API_UPDATE_AUTH_TOKEN"] = "abcd1234"
         os.environ["TELEGRAM_BOT_TOKEN"] = "id:sha"
         os.environ["WHATSAPP_API_UPDATE_AUTH_TOKEN"] = "efgh5678"
+        os.environ["WHATSAPP_APP_SECRET"] = "ijkl9012"
         os.environ["JWT_SECRET_KEY"] = "custom"
         os.environ["THE_AGENT_ISSUES_TOKEN"] = "sk-gi-valid"
         os.environ["RAPID_API_TWITTER_TOKEN"] = "sk-rt-valid"
@@ -145,6 +147,7 @@ def test_custom_config(self):
         self.assertEqual(config.telegram_auth_key.get_secret_value(), "abcd1234")
         self.assertEqual(config.telegram_bot_token.get_secret_value(), "id:sha")
         self.assertEqual(config.whatsapp_auth_key.get_secret_value(), "efgh5678")
+        self.assertEqual(config.whatsapp_app_secret.get_secret_value(), "ijkl9012")
         self.assertEqual(config.jwt_secret_key.get_secret_value(), "custom")
         self.assertEqual(config.github_issues_token.get_secret_value(), "sk-gi-valid")
         self.assertEqual(config.rapid_api_twitter_token.get_secret_value(), "sk-rt-valid")