Update workflows to test on the current Python version

.github/workflows/lint.yaml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.9"]
+        python-version: ["3.13"]
     steps:
       - uses: actions/checkout@v4
       - name: Install Task

.github/workflows/tests.yaml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.9"]
+        python-version: ["3.13"]
     steps:
       - uses: actions/checkout@v5
       - name: Install Task

README.md

@@ -12,27 +12,55 @@ The bot is designed to run within a Kubernetes environment, but can be operated
 
 ## Configuration
 
-All configuration is provided via environment variables
+SMSBot can be configured using either a configuration file or environment variables. Environment variables will override any values set in the configuration file.
 
-| Variable                   | Required? | Description                                                                 |
-| -------------------------- | --------- | --------------------------------------------------------------------------- |
-| SMSBOT_DEFAULT_SUBSCRIBERS | No        | A list of IDs, seperated by commas, to add to the subscribers list on start |
-| SMSBOT_LISTEN_HOST         | No        | The host for the webhooks to listen on, defaults to `0.0.0.0`               |
-| SMSBOT_LISTEN_PORT         | No        | The port to listen to, defaults to `80`                                     |
-| SMSBOT_OWNER_ID            | No        | ID of the owner of this bot                                                 |
-| SMSBOT_TELEGRAM_BOT_TOKEN  | Yes       | Your Bot Token for Telegram                                                 |
-| SMSBOT_TWILIO_AUTH_TOKEN   | No        | Twilio auth token, used to validate any incoming webhook calls              |
+### Configuration File
+
+Create a configuration file (e.g., `config.ini`) based on the provided `config-example.ini`:
+
+```ini
+[logging]
+level = INFO
+
+[webhook]
+host = 127.0.0.1
+port = 80
+
+[telegram]
+owner_id = OWNER_USER_ID
+bot_token = BOT_TOKEN
+
+[twilio]
+account_sid = TWILIO_ACCOUNT_SID
+auth_token = TWILIO_AUTH_TOKEN
+```
+
+### Environment Variables
+
+All configuration options can be overridden using environment variables:
+
+| Environment Variable        | Config Section | Config Key  | Required? | Description                                                                 |
+| --------------------------- | -------------- | ----------- | --------- | --------------------------------------------------------------------------- |
+| SMSBOT_LOGGING_LEVEL        | logging        | level       | No        | The log level to output to the console, defaults to `INFO`                  |
+| SMSBOT_TELEGRAM_BOT_TOKEN   | telegram       | bot_token   | Yes       | Your Bot Token for Telegram                                                 |
+| SMSBOT_TELEGRAM_OWNER_ID    | telegram       | owner_id    | No        | ID of the owner of this bot                                                 |
+| SMSBOT_TELEGRAM_SUBSCRIBERS | telegram       | subscribers | No        | A list of IDs, separated by commas, to add to the subscribers list on start |
+| SMSBOT_TWILIO_ACCOUNT_SID   | twilio         | account_sid | No        | Twilio account SID                                                          |
+| SMSBOT_TWILIO_AUTH_TOKEN    | twilio         | auth_token  | No        | Twilio auth token, used to validate any incoming webhook calls              |
+| SMSBOT_WEBHOOK_HOST         | webhook        | host        | No        | The host for the webhooks to listen on, defaults to `127.0.0.1`             |
+| SMSBOT_WEBHOOK_PORT         | webhook        | port        | No        | The port to listen to, defaults to `80`                                     |
 
 ## Setup
 
 To configure SMSBot, you'll need a Twilio account, either paid or trial is fine.
 
-* Setup a number in the location you want.
-* Under Phone Numbers -> Manage -> Active Numbers, click the number you want to setup.
-* In the "Voice & Fax" section, update the "A Call Comes In" to the URL of your SMSBot instance, with the endpoint being `/call`, e.g. `http://mymachine.test.com/call`
-* In the "Messaging" section, update the "A Message Comes In" to the URL of your SMSBot instance, with the endpoint being `/message`, e.g. `http://mymachine.test.com/message`
+1. Copy `config-example.ini` to `config.ini` and update the values, or set the appropriate environment variables.
+2. Setup a number in the location you want.
+3. Under Phone Numbers -> Manage -> Active Numbers, click the number you want to setup.
+4. In the "Voice & Fax" section, update the "A Call Comes In" to the URL of your SMSBot instance, with the endpoint being `/call`, e.g. `http://mymachine.test.com/call`
+5. In the "Messaging" section, update the "A Message Comes In" to the URL of your SMSBot instance, with the endpoint being `/message`, e.g. `http://mymachine.test.com/message`
 
-Your bot should now receive messages, on Telegram you need to start a chat or invite it into any channels you want, then update the `SMSBOT_DEFAULT_SUBSCRIBERS` values with their IDs.
+Your bot should now receive messages, on Telegram you need to start a chat or invite it into any channels you want, then update the `SMSBOT_TELEGRAM_SUBSCRIBERS` values with their IDs.
 
 **Note**: You cannot send test messages from your Twilio account to your Twilio numbers, they'll be silently dropped or fail with an "Invalid Number" error.
 

config-example.ini

@@ -7,6 +7,7 @@ port = 80
 
 [telegram]
 owner_id = OWNER_USER_ID
+subscribers = 1111,2222,3333
 bot_token = BOT_TOKEN
 
 [twilio]

pyproject.toml

@@ -30,3 +30,6 @@ dev = [
     "ruff>=0.12.9",
 ]
 github = ["pytest-github-actions-annotate-failures>=0.3.0"]
+
+[tool.ruff]
+line-length = 120

smsbot/cli.py

@@ -24,13 +24,14 @@ def main():
         help="Path to the config file",
     )
     parser.add_argument("--debug", action="store_true", help="Enable debug mode")
+    parser.add_argument("--log-file", type=argparse.FileType("a"), help="Path to the log file", default=sys.stdout)
     args = parser.parse_args()
 
     if args.debug:
         level = logging.DEBUG
     else:
         level = logging.INFO
-    logging.basicConfig(level=level, stream=sys.stdout)
+    logging.basicConfig(level=level, stream=args.log_file)
     logging.info("smsbot v%s", get_smsbot_version())
     logging.debug("Arguments: %s", args)
 
@@ -49,7 +50,7 @@ def main():
 
     # Validate configuration
     if not config.has_section("telegram") or not config.get("telegram", "bot_token"):
-        logging.error("Telegram bot token is required")
+        logging.error("Telegram bot token is required, define a token either in the config file or as an environment variable.")
         return
 
     # Now the config is loaded, set the logger level
@@ -66,8 +67,8 @@ def main():
         logging.warning("No Owner ID is set, which is not a good idea...")
 
     # Add default subscribers
-    if config.has_option("telegram", "default_subscribers"):
-        for chat_id in config.get("telegram", "default_subscribers").split(","):
+    if config.has_option("telegram", "subscribers"):
+        for chat_id in config.get("telegram", "subscribers").split(","):
             telegram_bot.subscribers.append(int(chat_id.strip()))
 
     # Init the webhook handler

smsbot/telegram.py

@@ -12,9 +12,7 @@ from telegram.ext import (
 
 from smsbot.utils import get_smsbot_version
 
-REQUEST_TIME = Summary(
-    "telegram_request_processing_seconds", "Time spent processing request"
-)
+REQUEST_TIME = Summary("telegram_request_processing_seconds", "Time spent processing request")
 COMMAND_COUNT = Counter("telegram_command_count", "Total number of commands processed")
 
 
@@ -35,21 +33,18 @@ class TelegramSmsBot:
 
     async def callback(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
         """Handle the update"""
-        if update.effective_user.id in self.owners:
-            self.logger.info(
-                f"{update.effective_user.username} sent {update.message.text}"
-            )
-            COMMAND_COUNT.inc()
-        else:
-            self.logger.debug(f"Ignoring message from user {update.effective_user.id}")
-            raise ApplicationHandlerStop
+        if update.effective_user and update.message:
+            if update.effective_user.id in self.owners:
+                self.logger.info(f"{update.effective_user.username} sent {update.message.text}")
+                COMMAND_COUNT.inc()
+            else:
+                self.logger.debug(f"Ignoring message from user {update.effective_user.username}")
+                raise ApplicationHandlerStop
 
     async def send_message(self, chat_id: int, text: str):
         """Send a message to a specific chat"""
         self.logger.info(f"Sending message to chat {chat_id}: {text}")
-        await self.app.bot.send_message(
-            chat_id=chat_id, text=text, parse_mode="MarkdownV2"
-        )
+        await self.app.bot.send_message(chat_id=chat_id, text=text, parse_mode="MarkdownV2")
 
     async def send_subscribers(self, text: str):
         """Send a message to all subscribers"""
@@ -64,48 +59,41 @@ class TelegramSmsBot:
             await self.send_message(owner, text)
 
     @REQUEST_TIME.time()
-    async def handler_help(self, update, context):
+    async def handler_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
         """Send a message when the command /help is issued."""
-        self.logger.info("/help command received in chat: %s", update.message.chat)
+        if update.message:
+            self.logger.info("/help command received in chat: %s", update.message.chat)
 
-        commands = []
-        for command in self.app.handlers[0]:
-            if isinstance(command, CommandHandler):
-                commands.extend(["/{0}".format(cmd) for cmd in command.commands])
+            commands = []
+            for command in self.app.handlers[0]:
+                if isinstance(command, CommandHandler):
+                    commands.extend(["/{0}".format(cmd) for cmd in command.commands])
 
-        await update.message.reply_markdown(
-            "Smsbot v{0}\n\n{1}".format(get_smsbot_version(), "\n".join(commands))
-        )
-        COMMAND_COUNT.inc()
+            await update.message.reply_markdown("Smsbot v{0}\n\n{1}".format(get_smsbot_version(), "\n".join(commands)))
+            COMMAND_COUNT.inc()
 
     @REQUEST_TIME.time()
-    async def handler_subscribe(
-        self, update: Update, context: ContextTypes.DEFAULT_TYPE
-    ):
+    async def handler_subscribe(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
         """Handle subscription requests"""
-        user_id = update.effective_user.id
-        if user_id not in self.subscribers:
-            self.subscribers.append(user_id)
-            self.logger.info(f"User {user_id} subscribed.")
-            self.logger.info(f"Current subscribers: {self.subscribers}")
-            await update.message.reply_markdown(
-                "You have successfully subscribed to updates."
-            )
-        else:
-            self.logger.info(f"User {user_id} is already subscribed.")
+        if update.effective_user and update.message:
+            user_id = update.effective_user.id
+            if user_id not in self.subscribers:
+                self.subscribers.append(user_id)
+                self.logger.info(f"User {user_id} subscribed.")
+                self.logger.info(f"Current subscribers: {self.subscribers}")
+                await update.message.reply_markdown("You have successfully subscribed to updates.")
+            else:
+                self.logger.info(f"User {user_id} is already subscribed.")
 
     @REQUEST_TIME.time()
-    async def handler_unsubscribe(
-        self, update: Update, context: ContextTypes.DEFAULT_TYPE
-    ):
+    async def handler_unsubscribe(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
         """Handle unsubscription requests"""
-        user_id = update.effective_user.id
-        if user_id in self.subscribers:
-            self.subscribers.remove(user_id)
-            self.logger.info(f"User {user_id} unsubscribed.")
-            self.logger.info(f"Current subscribers: {self.subscribers}")
-            await update.message.reply_markdown(
-                "You have successfully unsubscribed from updates."
-            )
-        else:
-            self.logger.info(f"User {user_id} is not subscribed.")
+        if update.effective_user and update.message:
+            user_id = update.effective_user.id
+            if user_id in self.subscribers:
+                self.subscribers.remove(user_id)
+                self.logger.info(f"User {user_id} unsubscribed.")
+                self.logger.info(f"Current subscribers: {self.subscribers}")
+                await update.message.reply_markdown("You have successfully unsubscribed from updates.")
+            else:
+                self.logger.info(f"User {user_id} is not subscribed.")

smsbot/utils.py

@@ -1,13 +1,13 @@
 from importlib.metadata import version
 
 
-def get_smsbot_version():
+def get_smsbot_version() -> str:
     return version("smsbot")
 
 
 class TwilioWebhookPayload:
     @staticmethod
-    def parse(data: dict):
+    def parse(data: dict[str, str]) -> "TwilioCall | TwilioMessage | None":
         """Return the correct class for the incoming Twilio webhook payload"""
         if "SmsMessageSid" in data:
             return TwilioMessage(data)
@@ -62,11 +62,7 @@ class TwilioMessage(TwilioWebhookPayload):
         return msg
 
     def to_markdownv2(self):
-        media_str = (
-            "\n".join([f"{self._escape(url)}" for url in self.media])
-            if self.media
-            else ""
-        )
+        media_str = "\n".join([f"{self._escape(url)}" for url in self.media]) if self.media else ""
         msg = f"**From**: {self._escape(self.from_number)}\n**To**: {self._escape(self.to_number)}\n\n{self._escape(self.body)}\n\n{media_str}"
         return msg
 
@@ -85,6 +81,6 @@ class TwilioCall(TwilioWebhookPayload):
         msg = f"Call from {self.from_number}, rejected."
         return msg
 
-    def to_markdownv2(self):
+    def to_markdownv2(self) -> str:
         msg = f"Call from {self._escape(self.from_number)}, rejected\\."
         return msg

smsbot/webhook.py

@@ -8,18 +8,16 @@ from werkzeug.middleware.dispatcher import DispatcherMiddleware
 
 from smsbot.utils import TwilioWebhookPayload, get_smsbot_version
 
-REQUEST_TIME = Summary(
-    "webhook_request_processing_seconds", "Time spent processing request"
-)
+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")
 
 
-
 class TwilioWebhookHandler(object):
     """
     A wrapped Flask app handling webhooks received from Twilio
     """
+
     def __init__(self, account_sid: str | None = None, auth_token: str | None = None):
         self.app = Flask(self.__class__.__name__)
         self.app.add_url_rule("/", "index", self.index, methods=["GET"])
@@ -50,9 +48,7 @@ class TwilioWebhookHandler(object):
         async def decorated_function(*args, **kwargs):
             # Create an instance of the RequestValidator class
             if not self.auth_token:
-                current_app.logger.warning(
-                    "Twilio request validation skipped due to Twilio Auth Token missing"
-                )
+                current_app.logger.warning("Twilio request validation skipped due to Twilio Auth Token missing")
                 return await func(*args, **kwargs)
             validator = RequestValidator(self.auth_token)
 
@@ -75,10 +71,10 @@ class TwilioWebhookHandler(object):
     def set_bot(self, bot):
         self.bot = bot
 
-    async def index(self):
+    async def index(self) -> str:
         return f'smsbot v{get_smsbot_version()} - <a href="https://github.com/nikdoof/smsbot">GitHub</a>'
 
-    async def health(self):
+    async def health(self) -> dict[str, str | int]:
         """Return basic health information"""
         return {
             "version": get_smsbot_version(),
@@ -87,29 +83,24 @@ class TwilioWebhookHandler(object):
         }
 
     @time(REQUEST_TIME)
-    async def message(self):
+    async def message(self) -> str:
         """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(
-            TwilioWebhookPayload.parse(request.values.to_dict()).to_markdownv2()
-        )
+        current_app.logger.info("Received SMS from {From}: {Body}".format(**request.values.to_dict()))
+        hook_data = TwilioWebhookPayload.parse(request.values.to_dict())
+        if hook_data:
+            await self.bot.send_subscribers(hook_data.to_markdownv2())
 
         # Return a blank response
         MESSAGE_COUNT.inc()
         return '<?xml version="1.0" encoding="UTF-8"?><Response></Response>'
 
     @time(REQUEST_TIME)
-    async def call(self):
+    async def call(self) -> str:
         """Handle incoming calls from Twilio"""
-        current_app.logger.info(
-            "Received Call from {From}".format(**request.values.to_dict())
-        )
-        await self.bot.send_subscribers(
-            TwilioWebhookPayload.parse(request.values.to_dict()).to_markdownv2()
-        )
+        current_app.logger.info("Received Call from {From}".format(**request.values.to_dict()))
+        hook_data = TwilioWebhookPayload.parse(request.values.to_dict())
+        if hook_data:
+            await self.bot.send_subscribers(hook_data.to_markdownv2())
 
         # Always reject calls
         CALL_COUNT.inc()