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] <name>; 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

@@ -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.

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] <keywords_format_key>`` 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}")

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.
 

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.