From 2c971875c13ef24be65c75be4a5b3889d8cda8e8 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sat, 3 Oct 2026 22:34:11 -0700 Subject: [PATCH] refactor(commands): share the pubkey ACL check and the [Keywords] response format S4. _check_admin_access and announcements' _check_announcements_access ran the same pubkey validation and logging, differing only in wording; both now call BaseCommand._pubkey_in_acl(message, acl, kind, denied), which builds each log line from those two words, so every line keeps its exact text and level. The admin check still runs it inside its try, announcements outside, as before. ping and magic8 had identical get_response_format overrides reading [Keywords] ; a keywords_format_key class attribute on BaseCommand now does that, and commands without one still get None. typing.Optional stays importable from both modules. Old-vs-new over 20,000 randomized configs and senders (missing, malformed, mixed-case and listed keys; missing sections; quoted and empty formats) matched on results and log calls; three mutations were each caught. --- modules/commands/announcements_command.py | 36 +-------- modules/commands/base_command.py | 91 ++++++++++++++--------- modules/commands/magic8_command.py | 14 +--- modules/commands/ping_command.py | 14 +--- 4 files changed, 63 insertions(+), 92 deletions(-) diff --git a/modules/commands/announcements_command.py b/modules/commands/announcements_command.py index b357d6d..27bc561 100644 --- a/modules/commands/announcements_command.py +++ b/modules/commands/announcements_command.py @@ -157,39 +157,9 @@ class AnnouncementsCommand(BaseCommand): self.logger.warning("No announcements ACL configured") return False - # Get sender's public key - NEVER fall back to sender_id - sender_pubkey = getattr(message, 'sender_pubkey', None) - if not sender_pubkey: - self.logger.warning( - f"No sender public key available for {message.sender_id} - " - "announcements access denied (missing pubkey)" - ) - return False - - # Validate sender pubkey format - if not validate_pubkey_format(sender_pubkey, expected_length=64): - self.logger.warning( - f"Invalid sender pubkey format from {message.sender_id}: " - f"{sender_pubkey[:16]}... - announcements access denied" - ) - return False - - # Normalize and compare - sender_pubkey_normalized = sender_pubkey.lower() - has_access = sender_pubkey_normalized in self.announcements_acl - - if not has_access: - self.logger.warning( - f"Announcements access denied for {message.sender_id} " - f"(pubkey: {sender_pubkey[:16]}...) - not in announcements ACL" - ) - else: - self.logger.info( - f"Announcements access granted for {message.sender_id} " - f"(pubkey: {sender_pubkey[:16]}...)" - ) - - return has_access + return self._pubkey_in_acl( + message, self.announcements_acl, "announcements", "Announcements access denied" + ) def can_execute(self, message: MeshMessage, skip_channel_check: bool = False) -> bool: """Check if announcements command can be executed. diff --git a/modules/commands/base_command.py b/modules/commands/base_command.py index 93ccc5d..b6a58da 100644 --- a/modules/commands/base_command.py +++ b/modules/commands/base_command.py @@ -85,6 +85,11 @@ class BaseCommand(ABC): # lists. requires_admin_access() is still the method callers ask. admin_only: bool = False + # The [Keywords] option whose value, when set, is this command's response + # format (quotes stripped). None: get_response_format() returns None unless + # the command overrides it. + keywords_format_key: Optional[str] = None + # Whether can_execute honors skip_channel_check. Commands whose own override # used to drop the argument (it called super().can_execute(message)) set this # to False so they keep enforcing their channel list even when a delegating @@ -1323,8 +1328,16 @@ class BaseCommand(ABC): return response_format def get_response_format(self) -> Optional[str]: - """Get the response format for this command from config""" - # Override in subclasses to provide custom response formats + """Get the response format for this command from config. + + Reads ``[Keywords] `` when the class names one; + override for any other source. + """ + if self.keywords_format_key is None: + return None + if self.bot.config.has_section('Keywords'): + format_str = self.bot.config.get('Keywords', self.keywords_format_key, fallback=None) + return self._strip_quotes_from_config(format_str) if format_str else None return None def requires_admin_access(self) -> bool: @@ -1347,6 +1360,46 @@ class BaseCommand(ABC): self.logger.warning(f"Error checking admin access requirement: {e}") return False + def _pubkey_in_acl(self, message: MeshMessage, acl: Any, kind: str, denied: str) -> bool: + """Whether the sender's validated public key is in ``acl`` (lowercase keys). + + Never falls back to sender_id. ``kind`` names the access in the log + lines ("admin", "announcements"); ``denied`` starts the not-in-ACL line. + """ + # Get sender's public key - NEVER fall back to sender_id + sender_pubkey = getattr(message, 'sender_pubkey', None) + if not sender_pubkey: + self.logger.warning( + f"No sender public key available for {message.sender_id} - " + f"{kind} access denied (missing pubkey)" + ) + return False + + # Validate sender pubkey format + if not validate_pubkey_format(sender_pubkey, expected_length=64): + self.logger.warning( + f"Invalid sender pubkey format from {message.sender_id}: " + f"{sender_pubkey[:16]}... - {kind} access denied" + ) + return False + + # Normalize and compare + sender_pubkey_normalized = sender_pubkey.lower() + has_access = sender_pubkey_normalized in acl + + if not has_access: + self.logger.warning( + f"{denied} for {message.sender_id} " + f"(pubkey: {sender_pubkey[:16]}...) - not in {kind} ACL" + ) + else: + self.logger.info( + f"{kind.capitalize()} access granted for {message.sender_id} " + f"(pubkey: {sender_pubkey[:16]}...)" + ) + + return has_access + def _check_admin_access(self, message: MeshMessage) -> bool: """ Check if the message sender has admin access (security-hardened) @@ -1388,39 +1441,7 @@ class BaseCommand(ABC): self.logger.error("No valid admin pubkeys found in config after validation") return False - # Get sender's public key - NEVER fall back to sender_id - sender_pubkey = getattr(message, 'sender_pubkey', None) - if not sender_pubkey: - self.logger.warning( - f"No sender public key available for {message.sender_id} - " - "admin access denied (missing pubkey)" - ) - return False - - # Validate sender pubkey format - if not validate_pubkey_format(sender_pubkey, expected_length=64): - self.logger.warning( - f"Invalid sender pubkey format from {message.sender_id}: " - f"{sender_pubkey[:16]}... - admin access denied" - ) - return False - - # Normalize and compare - sender_pubkey_normalized = sender_pubkey.lower() - is_admin = sender_pubkey_normalized in admin_pubkey_list - - if not is_admin: - self.logger.warning( - f"Access denied for {message.sender_id} " - f"(pubkey: {sender_pubkey[:16]}...) - not in admin ACL" - ) - else: - self.logger.info( - f"Admin access granted for {message.sender_id} " - f"(pubkey: {sender_pubkey[:16]}...)" - ) - - return is_admin + return self._pubkey_in_acl(message, admin_pubkey_list, "admin", "Access denied") except Exception as e: self.logger.error(f"Error checking admin access: {e}") diff --git a/modules/commands/magic8_command.py b/modules/commands/magic8_command.py index 837ae1e..897a6ff 100644 --- a/modules/commands/magic8_command.py +++ b/modules/commands/magic8_command.py @@ -4,7 +4,7 @@ Magic 8-ball command for the MeshCore Bot Handles the 'magic8' keyword response """ import random -from typing import Optional +from typing import Optional # noqa: F401 importable from this module on dev from ..models import MeshMessage from .base_command import BaseCommand @@ -28,6 +28,7 @@ class Magic8Command(BaseCommand): name = "magic8" honors_skip_channel_check = False enabled_attr = "magic8_enabled" + keywords_format_key = "magic8" keywords = ['magic8'] description = "Emulates the classic Magic 8-ball toy'" category = "games" @@ -54,17 +55,6 @@ class Magic8Command(BaseCommand): """ return self.translate('commands.magic8.description') - def get_response_format(self) -> Optional[str]: - """Get the response format from config. - - Returns: - Optional[str]: The format string for the response, or None if not configured. - """ - if self.bot.config.has_section('Keywords'): - format_str = self.bot.config.get('Keywords', 'magic8', fallback=None) - return self._strip_quotes_from_config(format_str) if format_str else None - return None - async def execute(self, message: MeshMessage) -> bool: """Execute the magic8 command. diff --git a/modules/commands/ping_command.py b/modules/commands/ping_command.py index 0a0bac5..aa9fe86 100644 --- a/modules/commands/ping_command.py +++ b/modules/commands/ping_command.py @@ -4,7 +4,7 @@ Ping command for the MeshCore Bot Handles the 'ping' keyword response """ -from typing import Optional +from typing import Optional # noqa: F401 importable from this module on dev from ..models import MeshMessage from .base_command import BaseCommand @@ -21,6 +21,7 @@ class PingCommand(BaseCommand): name = "ping" honors_skip_channel_check = False enabled_attr = "ping_enabled" + keywords_format_key = "ping" keywords = ['ping'] description = "Responds to 'ping' with 'Pong!'" category = "basic" @@ -47,17 +48,6 @@ class PingCommand(BaseCommand): """ return self.translate('commands.ping.description') - def get_response_format(self) -> Optional[str]: - """Get the response format from config. - - Returns: - Optional[str]: The format string for the response, or None if not configured. - """ - if self.bot.config.has_section('Keywords'): - format_str = self.bot.config.get('Keywords', 'ping', fallback=None) - return self._strip_quotes_from_config(format_str) if format_str else None - return None - async def execute(self, message: MeshMessage) -> bool: """Execute the ping command.