mirror of
https://github.com/nikdoof/smsbot.git
synced 2025-12-13 10:02:15 +00:00
112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
import os
|
|
from functools import wraps
|
|
|
|
from flask import Flask, abort, current_app, request
|
|
from prometheus_async.aio import time
|
|
from prometheus_client import Counter, Summary, make_wsgi_app
|
|
from twilio.request_validator import RequestValidator
|
|
from werkzeug.middleware.dispatcher import DispatcherMiddleware
|
|
|
|
from smsbot.utils import TwilioMessage, get_smsbot_version
|
|
|
|
REQUEST_TIME = Summary(
|
|
"webhook_request_processing_seconds", "Time spent processing request"
|
|
)
|
|
MESSAGE_COUNT = Counter("webhook_message_count", "Total number of messages processed")
|
|
CALL_COUNT = Counter("webhook_call_count", "Total number of calls processed")
|
|
|
|
|
|
def validate_twilio_request(func):
|
|
"""Validates that incoming requests genuinely originated from Twilio"""
|
|
|
|
@wraps(func)
|
|
async def decorated_function(*args, **kwargs):
|
|
# Create an instance of the RequestValidator class
|
|
twilio_token = os.environ.get("SMSBOT_TWILIO_AUTH_TOKEN")
|
|
|
|
if not twilio_token:
|
|
current_app.logger.warning(
|
|
"Twilio request validation skipped due to SMSBOT_TWILIO_AUTH_TOKEN missing"
|
|
)
|
|
return await func(*args, **kwargs)
|
|
|
|
validator = RequestValidator(twilio_token)
|
|
|
|
# Validate the request using its URL, POST data,
|
|
# and X-TWILIO-SIGNATURE header
|
|
request_valid = validator.validate(
|
|
request.url,
|
|
request.form,
|
|
request.headers.get("X-TWILIO-SIGNATURE", ""),
|
|
)
|
|
|
|
# Continue processing the request if it's valid, return a 403 error if
|
|
# it's not
|
|
if request_valid or current_app.debug:
|
|
return func(*args, **kwargs)
|
|
return abort(403)
|
|
|
|
return decorated_function
|
|
|
|
|
|
class TwilioWebhookHandler(object):
|
|
def __init__(self):
|
|
self.app = Flask(self.__class__.__name__)
|
|
self.app.add_url_rule("/", "index", self.index, methods=["GET"])
|
|
self.app.add_url_rule("/health", "health", self.health, methods=["GET"])
|
|
self.app.add_url_rule("/message", "message", self.message, methods=["POST"])
|
|
self.app.add_url_rule("/call", "call", self.call, methods=["POST"])
|
|
|
|
# Add prometheus wsgi middleware to route /metrics requests
|
|
self.app.wsgi_app = DispatcherMiddleware(
|
|
self.app.wsgi_app,
|
|
{
|
|
"/metrics": make_wsgi_app(),
|
|
},
|
|
)
|
|
|
|
def set_bot(self, bot):
|
|
self.bot = bot
|
|
|
|
async def index(self):
|
|
return f'smsbot v{get_smsbot_version()} - <a href="https://github.com/nikdoof/smsbot">GitHub</a>'
|
|
|
|
async def health(self):
|
|
"""Return basic health information"""
|
|
return {
|
|
"version": get_smsbot_version(),
|
|
"owners": self.bot.owners,
|
|
"subscribers": self.bot.subscribers,
|
|
}
|
|
|
|
@time(REQUEST_TIME)
|
|
@validate_twilio_request
|
|
async def message(self):
|
|
"""Handle incoming SMS messages from Twilio"""
|
|
current_app.logger.info(
|
|
"Received SMS from {From}: {Body}".format(**request.values.to_dict())
|
|
)
|
|
|
|
await self.bot.send_subscribers(
|
|
TwilioMessage(request.values.to_dict()).to_markdownv2()
|
|
)
|
|
|
|
# Return a blank response
|
|
MESSAGE_COUNT.inc()
|
|
return '<?xml version="1.0" encoding="UTF-8"?><Response></Response>'
|
|
|
|
@time(REQUEST_TIME)
|
|
@validate_twilio_request
|
|
async def call(self):
|
|
"""Handle incoming calls from Twilio"""
|
|
current_app.logger.info(
|
|
"Received Call from {From}".format(**request.values.to_dict())
|
|
)
|
|
await self.bot.send_subscribers(
|
|
"Received Call from {From}, rejecting.".format(**request.values.to_dict())
|
|
)
|
|
|
|
# Always reject calls
|
|
CALL_COUNT.inc()
|
|
return '<?xml version="1.0" encoding="UTF-8"?><Response><Reject/></Response>'
|