|
|
|
@ -5,6 +5,8 @@ import logging |
|
|
|
import os |
|
|
|
import tempfile |
|
|
|
import subprocess |
|
|
|
import base64 |
|
|
|
import jwt |
|
|
|
from typing import Dict, Any |
|
|
|
|
|
|
|
from django.conf import settings |
|
|
|
@ -113,16 +115,38 @@ class PlugNMeetWebhookAPIView(APIView): |
|
|
|
return False |
|
|
|
|
|
|
|
body_bytes = request.data |
|
|
|
|
|
|
|
# 1. Try JWT verification (PlugNMeet production standard) |
|
|
|
try: |
|
|
|
# A JWT typically consists of three segments separated by dots |
|
|
|
if len(hash_token.split('.')) == 3: |
|
|
|
decoded = jwt.decode(hash_token, api_secret, algorithms=['HS256']) |
|
|
|
token_sha256 = decoded.get('sha256') |
|
|
|
if token_sha256: |
|
|
|
hasher = hashlib.sha256() |
|
|
|
hasher.update(body_bytes) |
|
|
|
computed_sha256 = base64.b64encode(hasher.digest()).decode('utf-8') |
|
|
|
if hmac.compare_digest(token_sha256, computed_sha256): |
|
|
|
return True |
|
|
|
else: |
|
|
|
logger.error(f"❌ [PlugNMeet Webhook] SHA256 mismatch! Token claim: {token_sha256}, Computed: {computed_sha256}") |
|
|
|
else: |
|
|
|
logger.error("❌ [PlugNMeet Webhook] JWT missing 'sha256' claim") |
|
|
|
except jwt.PyJWTError as e: |
|
|
|
logger.warning(f"⚠️ [PlugNMeet Webhook] JWT decoding failed: {e}. Falling back to HMAC-SHA256 check.") |
|
|
|
|
|
|
|
# 2. Fallback: HMAC-SHA256 verification (for backward compatibility with test_webhook.py) |
|
|
|
expected_signature = hmac.new( |
|
|
|
api_secret.encode('utf-8'), |
|
|
|
body_bytes, |
|
|
|
hashlib.sha256 |
|
|
|
).hexdigest() |
|
|
|
|
|
|
|
if not hmac.compare_digest(hash_token, expected_signature): |
|
|
|
logger.error(f"❌ [PlugNMeet Webhook] Signature mismatch! \nReceived: {hash_token[:10]}...\nExpected: {expected_signature[:10]}...") |
|
|
|
return False |
|
|
|
return True |
|
|
|
if hmac.compare_digest(hash_token, expected_signature): |
|
|
|
return True |
|
|
|
|
|
|
|
logger.error(f"❌ [PlugNMeet Webhook] Signature mismatch! \nReceived: {hash_token[:10]}...\nExpected HMAC: {expected_signature[:10]}...") |
|
|
|
return False |
|
|
|
|
|
|
|
def _handle_room_finished(self, payload: Dict[str, Any]) -> Dict[str, Any]: |
|
|
|
room_data = payload.get('room', {}) |
|
|
|
|