"""
SMS + Call webhook routes — Twilio callback endpoints.
These routes receive delivery status updates and inbound SMS from Twilio.
No authentication required (webhook-level signature verification only).
"""
import hmac
import hashlib
import logging
from datetime import datetime, timezone
from flask import Blueprint, request, jsonify
from app.models import Connector, Company
from app.connectors import build_connector, ConnectorError
logger = logging.getLogger(__name__)
sms_webhooks_bp = Blueprint('sms_webhooks', __name__)
# ---------------------------------------------------------------------------
# Helper: resolve connector from webhook URL
# ---------------------------------------------------------------------------
def _resolve_connector_from_url() -> tuple:
"""
Resolve the connector and company from the webhook request URL.
Expected URL pattern: /connectors/sms/webhook/<connector_id>/sms
/connectors/sms/webhook/<connector_id>/call
Returns: (connector, company_id) or raises 404
"""
# Extract connector_id from URL path
# Path segments: ['', 'connectors', 'sms', 'webhook', '<connector_id>', 'sms'|'call']
parts = request.path.strip('/').split('/')
if len(parts) < 5:
logger.warning(f'Webhook URL too short: {request.path}')
return None, None
connector_id = parts[3] # /connectors/sms/webhook/<connector_id>/...
connector = Connector.query.filter_by(
service='sms',
id=connector_id,
active=True,
).first()
if not connector:
logger.warning(f'No active SMS connector found for ID: {connector_id}')
return None, None
return connector, connector.company_id
def _resolve_connector_from_headers() -> tuple:
"""
Fallback: resolve connector from X-Connector-ID header.
Returns: (connector, company_id) or raises 404
"""
connector_id = request.headers.get('X-Connector-ID', '')
if not connector_id:
return None, None
connector = Connector.query.filter_by(
service='sms',
id=connector_id,
active=True,
).first()
if not connector:
return None, None
return connector, connector.company_id
# ---------------------------------------------------------------------------
# Webhook signature verification
# ---------------------------------------------------------------------------
def verify_twilio_signature(connector, sms_conn) -> bool:
"""
Verify Twilio webhook signature (X-Twilio-Signature header).
Verification is MANDATORY — there is no configuration option to disable
it. Requests without a valid signature are rejected.
Returns True if valid, False otherwise.
"""
signature = request.headers.get('X-Twilio-Signature', '')
if not signature:
logger.warning('Missing X-Twilio-Signature header')
return False
if not sms_conn or not hasattr(sms_conn, 'provider'):
logger.error(f'No SMS connector instance for {connector.id}')
return False
# Build the URL used for signature verification
webhook_url = request.url
body = request.get_data(as_text=True)
try:
return bool(sms_conn.provider.verify_webhook_signature(signature, body, webhook_url))
except Exception:
# Fail closed on any provider error (e.g. provider without verification support)
logger.exception('Webhook signature verification raised — rejecting request')
return False
# ---------------------------------------------------------------------------
# SMS Webhook Endpoint
# ---------------------------------------------------------------------------
@sms_webhooks_bp.route('/connectors/sms/webhook/<connector_id>/sms', methods=['POST'])
def sms_webhook(connector_id: str):
"""
Twilio SMS webhook — handles inbound replies and delivery status.
Expected JSON or form data from Twilio:
- MessageSid, From, To, Body, MessageStatus, Direction, ErrorCode, ErrorText, NumMedia
"""
try:
# Resolve connector
connector = Connector.query.filter_by(
service='sms', id=connector_id, active=True
).first()
if not connector:
logger.warning(f'No active SMS connector: {connector_id}')
return jsonify({'error': 'Connector not found'}), 404
# Build connector instance
sms_conn = build_connector('sms', connector.company_id, connector.config, connector.id)
# Verify webhook signature
if not verify_twilio_signature(connector, sms_conn):
logger.warning(f'Invalid webhook signature for connector {connector_id}')
return jsonify({'error': 'Invalid signature'}), 401
# Parse request data (Twilio sends as form-encoded or JSON)
data = request.get_json(silent=True)
if not data:
# Fall back to form data
data = dict(request.form)
# Process webhook
result = sms_conn.handle_sms_webhook(connector.company_id, data)
# Log result
logger.info(
f'SMS webhook processed: connector={connector_id}, '
f'company={connector.company_id}, action={result.get("action", "unknown")}'
)
return jsonify({
'status': 'processed',
'action': result.get('action', ''),
'details': result,
}), 200
except ConnectorError as e:
logger.error(f'Connector error in SMS webhook: {e}')
return jsonify({'error': 'Provider error'}), 400
except Exception as e:
logger.exception(f'Unhandled error in SMS webhook: {e}')
return jsonify({'error': 'Internal server error'}), 500
# ---------------------------------------------------------------------------
# Call Webhook Endpoint
# ---------------------------------------------------------------------------
@sms_webhooks_bp.route('/connectors/sms/webhook/<connector_id>/call', methods=['POST'])
def call_webhook(connector_id: str):
"""
Twilio call status webhook — handles call disposition updates.
Expected data from Twilio:
- CallSid, CallStatus, Duration, From, To, RecordingUrl, CallDuration
"""
try:
# Resolve connector
connector = Connector.query.filter_by(
service='sms', id=connector_id, active=True
).first()
if not connector:
logger.warning(f'No active SMS connector: {connector_id}')
return jsonify({'error': 'Connector not found'}), 404
# Build connector instance
sms_conn = build_connector('sms', connector.company_id, connector.config, connector.id)
# Verify webhook signature
if not verify_twilio_signature(connector, sms_conn):
logger.warning(f'Invalid webhook signature for call webhook')
return jsonify({'error': 'Invalid signature'}), 401
# Parse request data
data = request.get_json(silent=True)
if not data:
data = dict(request.form)
# Process webhook
result = sms_conn.handle_call_webhook(connector.company_id, data)
logger.info(
f'Call webhook processed: connector={connector_id}, '
f'company={connector.company_id}, disposition={result.get("disposition", "unknown")}'
)
return jsonify({
'status': 'processed',
'action': result.get('action', ''),
'details': result,
}), 200
except ConnectorError as e:
logger.error(f'Connector error in call webhook: {e}')
return jsonify({'error': 'Provider error'}), 400
except Exception as e:
logger.exception(f'Unhandled error in call webhook: {e}')
return jsonify({'error': 'Internal server error'}), 500
# ---------------------------------------------------------------------------
# Health check
# ---------------------------------------------------------------------------
@sms_webhooks_bp.route('/connectors/sms/webhook/health', methods=['GET'])
def webhook_health():
"""Health check endpoint for webhook routing."""
return jsonify({'status': 'ok', 'service': 'sms_webhooks'}), 200