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.