fix: close the '*'-only flood_scopes bypass and schedule write race

A flood_scopes of "*" alone leaves scope_keys empty while setting
flood_scope_allow_global, and the loader already logs that as an active
allowlist. The handler gated on scope_keys alone, so that configuration
skipped authorisation entirely and admitted absent, uncorrelated and
TRANSPORT_FLOOD traffic. The gate now fires when either is set, so "*"
means global-only rather than everything.

The scheduled-message duplicate check ran outside the write lock inside
update_ini_values, so two concurrent creates for the same schedule could
both pass and the second silently replace the first instead of getting
the 409. Check and write are now one critical section, and delete is too.
Repaired the delete path's error handling while moving it, so an OSError
during the write is still a 500 rather than escaping.

modules/message_handler.py

@@ -2398,8 +2398,11 @@ class MessageHandler:
             # Allowlist enforcement: when flood_scopes is configured, only reply to
             # messages whose scope matched an entry.  Unscoped FLOOD is allowed only
             # when '*' (or equivalent) is explicitly listed.
-            if scope_keys and reply_scope is None:
-                allow_global = getattr(cmd_mgr, "flood_scope_allow_global", False)
+            allow_global = getattr(cmd_mgr, "flood_scope_allow_global", False)
+            # A '*'-only flood_scopes leaves scope_keys empty but still means an
+            # allowlist is configured (global only). Gating on scope_keys alone let
+            # that configuration skip authorisation entirely.
+            if (scope_keys or allow_global) and reply_scope is None:
                 if (
                     scope_rf_data
                     and scope_rf_is_correlated

modules/web_viewer/app.py

@@ -3987,6 +3987,11 @@ class BotDataViewer:
                 self.logger.exception("Failed to queue config reload")
                 return False
 
+        # The duplicate check and the write have to be one critical section, or two
+        # concurrent creates for the same schedule both pass the check and the second
+        # silently replaces the first instead of getting the promised 409.
+        schedule_write_lock = threading.Lock()
+
         def _existing_schedules():
             return {e['schedule'] for e in read_entries(self.config_path, _schedule_tz())}
 
@@ -4022,6 +4027,10 @@ class BotDataViewer:
 
         def _save_scheduled_message(data, *, replacing=None):
             """Shared create/update: validate, write config.ini, queue a reload."""
+            with schedule_write_lock:
+                return _save_scheduled_message_locked(data, replacing=replacing)
+
+        def _save_scheduled_message_locked(data, *, replacing=None):
             schedule = (data.get('schedule') or '').strip()
             channel = (data.get('channel') or '').strip()
             message = (data.get('message') or '').strip()
@@ -4120,16 +4129,22 @@ class BotDataViewer:
                 schedule = (data.get('schedule') or '').strip()
                 if not schedule:
                     return jsonify({'success': False, 'error': 'schedule is required'}), 400
-                if schedule not in _existing_schedules():
-                    return jsonify({'success': False, 'error': f"No scheduled message for '{schedule}'"}), 404
-                try:
-                    update_ini_values(self.config_path, {}, {SCHEDULED_MESSAGES_SECTION: [schedule]})
-                except OSError as exc:
-                    self.logger.error("Failed to delete scheduled message: %s", exc)
-                    return jsonify({
-                        'success': False,
-                        'error': 'Could not write config.ini — check file permissions',
-                    }), 500
+                with schedule_write_lock:
+                    if schedule not in _existing_schedules():
+                        return jsonify({
+                            'success': False,
+                            'error': f"No scheduled message for '{schedule}'",
+                        }), 404
+                    try:
+                        update_ini_values(
+                            self.config_path, {}, {SCHEDULED_MESSAGES_SECTION: [schedule]}
+                        )
+                    except OSError as exc:
+                        self.logger.error("Failed to delete scheduled message: %s", exc)
+                        return jsonify({
+                            'success': False,
+                            'error': 'Could not write config.ini — check file permissions',
+                        }), 500
                 reloaded = _queue_config_reload()
                 self.logger.info("Scheduled message deleted: %r", schedule)
                 return jsonify({

tests/integration/test_flood_scope_reply.py

@@ -518,3 +518,41 @@ class TestSnocoScopedPingRegression:
         cm.send_channel_message.assert_awaited_once()
         _, kwargs = cm.send_channel_message.call_args
         assert kwargs.get("scope") == "#snoco"
+
+
+class TestWildcardOnlyAllowlist:
+    """flood_scopes = "*" leaves scope_keys empty but still configures an allowlist
+    (global only). Gating on scope_keys alone let that configuration skip
+    authorisation entirely."""
+
+    @staticmethod
+    def _cmd_mgr(raw):
+        from unittest.mock import Mock
+
+        from modules.command_manager import CommandManager
+
+        mgr = object.__new__(CommandManager)
+        mgr.logger = Mock()
+        mgr._flood_scopes_config_raw = lambda: raw
+        mgr.flood_scope_allow_global = False
+        keys = mgr._load_flood_scope_keys()
+        return keys, mgr.flood_scope_allow_global
+
+    def test_wildcard_only_yields_no_keys_but_allows_global(self):
+        keys, allow_global = self._cmd_mgr("*")
+        assert keys == {}
+        assert allow_global is True
+
+    def test_wildcard_only_still_counts_as_a_configured_allowlist(self):
+        """The gate condition the handler uses must be true here, or '*' bypasses it."""
+        keys, allow_global = self._cmd_mgr("*")
+        assert bool(keys or allow_global) is True
+
+    def test_named_scope_plus_wildcard(self):
+        keys, allow_global = self._cmd_mgr("#west, *")
+        assert list(keys) == ["#west"]
+        assert allow_global is True
+
+    def test_unset_config_configures_no_allowlist(self):
+        keys, allow_global = self._cmd_mgr("")
+        assert bool(keys or allow_global) is False