diff --git a/modules/db_migrations.py b/modules/db_migrations.py
index 23a8877..7ef9427 100644
--- a/modules/db_migrations.py
+++ b/modules/db_migrations.py
@@ -25,6 +25,16 @@ from typing import Callable
 
 VALID_IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
 
+# Allowed column definition pattern: type keyword(s) optionally followed by
+# DEFAULT and a literal value.  This prevents SQL injection through the
+# definition parameter of _add_column().
+_VALID_COL_DEF = re.compile(
+    r"^[A-Z]+(?:\s+[A-Z]+)*"                      # type name, e.g. "TEXT", "INTEGER", "BOOLEAN"
+    r"(?:\s+DEFAULT\s+(?:'[^']*'|[0-9.]+|NULL|CURRENT_TIMESTAMP))?"  # optional DEFAULT clause
+    r"$",
+    re.IGNORECASE,
+)
+
 
 def _validate_ident(name: str, kind: str) -> None:
     if not VALID_IDENT.match(name):
@@ -48,12 +58,19 @@ def _column_exists(cursor: sqlite3.Cursor, table: str, column: str) -> bool:
     return any(row[1] == column for row in cursor.fetchall())
 
 
+def _validate_col_definition(definition: str) -> None:
+    """Ensure *definition* matches a safe SQLite column definition pattern."""
+    if not _VALID_COL_DEF.match(definition.strip()):
+        raise ValueError(f"Invalid column definition: {definition!r}")
+
+
 def _add_column(
     cursor: sqlite3.Cursor, table: str, column: str, definition: str
 ) -> None:
     """Add *column* to *table* if it does not already exist."""
     _validate_ident(table, "table")
     _validate_ident(column, "column")
+    _validate_col_definition(definition)
     if not _column_exists(cursor, table, column):
         cursor.execute(f"ALTER TABLE {table} ADD COLUMN {column} {definition}")
 
diff --git a/modules/repeater_manager.py b/modules/repeater_manager.py
index 5e6873d..5ff34d6 100644
--- a/modules/repeater_manager.py
+++ b/modules/repeater_manager.py
@@ -38,9 +38,9 @@ class RepeaterManager:
 
         # Initialize companion purge settings
         self.companion_purge_enabled = bot.config.getboolean('Companion_Purge', 'companion_purge_enabled', fallback=False)
-        self.companion_dm_threshold_days = bot.config.getint('Companion_Purge', 'companion_dm_threshold_days', fallback=30)
-        self.companion_advert_threshold_days = bot.config.getint('Companion_Purge', 'companion_advert_threshold_days', fallback=30)
-        self.companion_min_inactive_days = bot.config.getint('Companion_Purge', 'companion_min_inactive_days', fallback=30)
+        self.companion_dm_threshold_days = max(0, bot.config.getint('Companion_Purge', 'companion_dm_threshold_days', fallback=30))
+        self.companion_advert_threshold_days = max(0, bot.config.getint('Companion_Purge', 'companion_advert_threshold_days', fallback=30))
+        self.companion_min_inactive_days = max(0, bot.config.getint('Companion_Purge', 'companion_min_inactive_days', fallback=30))
 
         # Geocoding cache: packet_hash -> timestamp (to prevent duplicate geocoding within 1 minute)
         self.geocoding_cache = {}
@@ -464,7 +464,7 @@ class RepeaterManager:
         """Update the is_currently_tracked flag on an existing connection (no commit)."""
         is_tracked = False
         if hasattr(self.bot.meshcore, 'contacts'):
-            for contact_key, contact_data in self.bot.meshcore.contacts.items():
+            for contact_key, contact_data in list(self.bot.meshcore.contacts.items()):
                 if contact_data.get('public_key', contact_key) == public_key:
                     is_tracked = True
                     break
@@ -480,7 +480,7 @@ class RepeaterManager:
             # Check if this repeater is currently in the device's contact list
             is_tracked = False
             if hasattr(self.bot.meshcore, 'contacts'):
-                for contact_key, contact_data in self.bot.meshcore.contacts.items():
+                for contact_key, contact_data in list(self.bot.meshcore.contacts.items()):
                     if contact_data.get('public_key', contact_key) == public_key:
                         is_tracked = True
                         break
@@ -670,7 +670,7 @@ class RepeaterManager:
                 self.logger.warning("No repeaters available for auto-purge")
                 # Log some debugging info
                 total_contacts = len(self.bot.meshcore.contacts)
-                repeater_count = sum(1 for contact_data in self.bot.meshcore.contacts.values() if self._is_repeater_device(contact_data))
+                repeater_count = sum(1 for contact_data in list(self.bot.meshcore.contacts.values()) if self._is_repeater_device(contact_data))
                 self.logger.debug(f"Debug: {total_contacts} total contacts, {repeater_count} repeaters found")
                 return False
 
@@ -712,7 +712,7 @@ class RepeaterManager:
                 self.logger.warning("No companions available for auto-purge")
                 # Log some debugging info
                 total_contacts = len(self.bot.meshcore.contacts)
-                companion_count = sum(1 for contact_data in self.bot.meshcore.contacts.values() if self._is_companion_device(contact_data))
+                companion_count = sum(1 for contact_data in list(self.bot.meshcore.contacts.values()) if self._is_companion_device(contact_data))
                 self.logger.debug(f"Debug: {total_contacts} total contacts, {companion_count} companions found")
                 return False
 
@@ -758,7 +758,7 @@ class RepeaterManager:
             # Get repeaters directly from device contacts, not database
             device_repeaters = []
 
-            for contact_key, contact_data in self.bot.meshcore.contacts.items():
+            for contact_key, contact_data in list(self.bot.meshcore.contacts.items()):
                 # Check if this is a repeater device
                 if self._is_repeater_device(contact_data):
                     public_key = contact_data.get('public_key', contact_key)
@@ -854,7 +854,7 @@ class RepeaterManager:
             scored_companions = []
 
             # Get activity data from database for all companions
-            for contact_key, contact_data in self.bot.meshcore.contacts.items():
+            for contact_key, contact_data in list(self.bot.meshcore.contacts.items()):
                 # Check if this is a companion device
                 if not self._is_companion_device(contact_data):
                     continue
@@ -990,9 +990,10 @@ class RepeaterManager:
             ))
 
             # Enhanced debugging
-            total_companions_checked = sum(1 for contact_data in self.bot.meshcore.contacts.values()
+            contacts_snapshot = list(self.bot.meshcore.contacts.items())
+            total_companions_checked = sum(1 for _, contact_data in contacts_snapshot
                                           if self._is_companion_device(contact_data))
-            acl_skipped = sum(1 for contact_key, contact_data in self.bot.meshcore.contacts.items()
+            acl_skipped = sum(1 for contact_key, contact_data in contacts_snapshot
                             if self._is_companion_device(contact_data) and
                             self._is_in_acl(contact_data.get('public_key', contact_key)))
             recent_skipped = total_companions_checked - acl_skipped - len(scored_companions)
@@ -1676,7 +1677,7 @@ class RepeaterManager:
             if not sender_id:
                 # Try to get sender_id from device contacts
                 if hasattr(self.bot.meshcore, 'contacts'):
-                    for contact_key, contact_data in self.bot.meshcore.contacts.items():
+                    for contact_key, contact_data in list(self.bot.meshcore.contacts.items()):
                         if contact_data.get('public_key', contact_key) == public_key:
                             sender_id = contact_data.get('name', contact_data.get('adv_name', ''))
                             break
@@ -1778,7 +1779,7 @@ class RepeaterManager:
         processed_count = 0
 
         try:
-            for contact_key, contact_data in self.bot.meshcore.contacts.items():
+            for contact_key, contact_data in list(self.bot.meshcore.contacts.items()):
                 processed_count += 1
 
                 # Log progress every 20 contacts
@@ -2274,7 +2275,7 @@ class RepeaterManager:
                 name = repeater['name']
 
                 # Find the contact in meshcore.contacts
-                for contact_key, contact_data in self.bot.meshcore.contacts.items():
+                for contact_key, contact_data in list(self.bot.meshcore.contacts.items()):
                     if contact_data.get('public_key', contact_key) == public_key:
                         # Check the actual last_advert time
                         last_advert = contact_data.get('last_advert')
@@ -2316,7 +2317,7 @@ class RepeaterManager:
                 # Show some recent repeaters to understand the timestamp format
                 self.logger.info("No old repeaters found. Showing recent repeater activity:")
                 recent_count = 0
-                for contact_key, contact_data in self.bot.meshcore.contacts.items():
+                for contact_key, contact_data in list(self.bot.meshcore.contacts.items()):
                     if self._is_repeater_device(contact_data):
                         last_advert = contact_data.get('last_advert', 'No last_advert')
                         name = contact_data.get('adv_name', contact_data.get('name', 'Unknown'))
@@ -2441,7 +2442,7 @@ class RepeaterManager:
             # Count repeaters from actual device contacts (more accurate than database)
             device_repeater_count = 0
             if hasattr(self.bot.meshcore, 'contacts'):
-                for _contact_key, contact_data in self.bot.meshcore.contacts.items():
+                for _contact_key, contact_data in list(self.bot.meshcore.contacts.items()):
                     if self._is_repeater_device(contact_data):
                         device_repeater_count += 1
 
@@ -2484,7 +2485,7 @@ class RepeaterManager:
                 return []
 
             stale_contacts = []
-            for _contact_key, contact_data in self.bot.meshcore.contacts.items():
+            for _contact_key, contact_data in list(self.bot.meshcore.contacts.items()):
                 # Skip repeaters (they're managed separately)
                 if self._is_repeater_device(contact_data):
                     continue
@@ -3276,7 +3277,7 @@ class RepeaterManager:
             test_public_key = None
 
             # Look for a repeater contact to test with
-            for key, contact_data in self.bot.meshcore.contacts.items():
+            for key, contact_data in list(self.bot.meshcore.contacts.items()):
                 if self._is_repeater_device(contact_data):
                     test_contact = contact_data
                     test_public_key = str(contact_data.get('public_key', key))
diff --git a/modules/scheduler.py b/modules/scheduler.py
index ebab976..738c503 100644
--- a/modules/scheduler.py
+++ b/modules/scheduler.py
@@ -54,8 +54,8 @@ class MessageScheduler:
         if self._apscheduler is not None:
             try:
                 self._apscheduler.shutdown(wait=False)
-            except Exception:
-                pass
+            except Exception as e:
+                self.logger.debug("Error shutting down scheduler: %s", e)
         tz, _ = get_config_timezone(self.bot.config, self.logger)
         self._apscheduler = BackgroundScheduler(timezone=tz)
         self.scheduled_messages.clear()
@@ -142,15 +142,12 @@ class MessageScheduler:
             except Exception as e:
                 self.logger.error(f"Error sending scheduled message: {e}")
         else:
-            # Fallback: create new event loop if main loop not available
+            # Fallback: create a temporary event loop and close it when done
+            loop = asyncio.new_event_loop()
             try:
-                loop = asyncio.get_event_loop()
-            except RuntimeError:
-                loop = asyncio.new_event_loop()
-                asyncio.set_event_loop(loop)
-
-            # Run the async function in the event loop
-            loop.run_until_complete(self._send_scheduled_message_async(channel, message))
+                loop.run_until_complete(self._send_scheduled_message_async(channel, message))
+            finally:
+                loop.close()
 
     async def _get_mesh_info(self) -> dict[str, Any]:
         """Get mesh network information for scheduled messages"""
@@ -220,8 +217,8 @@ class MessageScheduler:
                             result = cursor.fetchone()
                             if result:
                                 info['recent_activity_24h'] = result[0]
-                except Exception:
-                    pass
+                except Exception as e:
+                    self.logger.debug("Error querying message_stats: %s", e)
 
             # Calculate new devices in last 7 days (matching web viewer logic)
             # Query devices first heard in the last 7 days, grouped by role
@@ -340,8 +337,8 @@ class MessageScheduler:
         if self._apscheduler is not None:
             try:
                 self._apscheduler.shutdown(wait=False)
-            except Exception:
-                pass
+            except Exception as e:
+                self.logger.debug("Error shutting down scheduler: %s", e)
         if self.scheduler_thread and self.scheduler_thread.is_alive():
             self.scheduler_thread.join(timeout=timeout)
             if self.scheduler_thread.is_alive():
@@ -394,18 +391,15 @@ class MessageScheduler:
                             if not f.cancelled() and f.exception() else None
                         )
                     else:
-                        # Fallback: create new event loop if main loop not available
-                        try:
-                            loop = asyncio.get_event_loop()
-                        except RuntimeError:
-                            loop = asyncio.new_event_loop()
-                            asyncio.set_event_loop(loop)
-
+                        # Fallback: create a temporary event loop and close it when done
+                        loop = asyncio.new_event_loop()
                         try:
                             loop.run_until_complete(self.bot.feed_manager.poll_all_feeds())
                             self.logger.debug("Feed polling cycle completed")
                         except Exception as e:
                             self.logger.error(f"Error in feed polling cycle: {e}")
+                        finally:
+                            loop.close()
                     last_feed_poll_time = time.time()
 
             # Channels are fetched once on launch only - no periodic refresh
diff --git a/modules/service_plugins/webhook_service.py b/modules/service_plugins/webhook_service.py
index 3e55963..23c94c6 100644
--- a/modules/service_plugins/webhook_service.py
+++ b/modules/service_plugins/webhook_service.py
@@ -36,6 +36,8 @@ Response codes:
 """
 
 import secrets
+import time
+from collections import OrderedDict
 from typing import Any, Optional
 
 from .base_service import BaseServicePlugin
@@ -83,7 +85,7 @@ class WebhookService(BaseServicePlugin):
 
         raw_channels = cfg.get("Webhook", "allowed_channels", fallback="").strip()
         self.allowed_channels = (
-            {c.strip().lstrip("#").lower() for c in raw_channels.split(",") if c.strip()}
+            {c.strip().removeprefix("#").lower() for c in raw_channels.split(",") if c.strip()}
             if raw_channels
             else set()
         )
@@ -91,6 +93,14 @@ class WebhookService(BaseServicePlugin):
         self._runner: Optional[Any] = None  # aio_web.AppRunner
         self._site: Optional[Any] = None    # aio_web.TCPSite
 
+        # Per-IP rate limiting
+        self._rate_limit_per_minute: int = cfg.getint(
+            "Webhook", "rate_limit_per_minute", fallback=30
+        )
+        self._rate_window: float = 60.0  # seconds
+        self._request_log: OrderedDict[str, list[float]] = OrderedDict()  # ip -> [timestamps]
+        self._max_tracked_ips: int = 1000
+
     # ------------------------------------------------------------------
     # Lifecycle
     # ------------------------------------------------------------------
@@ -126,8 +136,43 @@ class WebhookService(BaseServicePlugin):
     # Request handler
     # ------------------------------------------------------------------
 
+    def _is_rate_limited(self, remote_ip: str) -> bool:
+        """Return True if *remote_ip* has exceeded the per-minute request limit."""
+        if self._rate_limit_per_minute <= 0:
+            return False  # Rate limiting disabled
+
+        now = time.monotonic()
+        cutoff = now - self._rate_window
+
+        timestamps = self._request_log.get(remote_ip)
+        if timestamps is not None:
+            # Prune expired entries
+            timestamps[:] = [t for t in timestamps if t > cutoff]
+            if len(timestamps) >= self._rate_limit_per_minute:
+                return True
+            timestamps.append(now)
+            self._request_log.move_to_end(remote_ip)
+        else:
+            self._request_log[remote_ip] = [now]
+
+        # Evict oldest IPs to bound memory
+        while len(self._request_log) > self._max_tracked_ips:
+            self._request_log.popitem(last=False)
+
+        return False
+
     async def _handle_webhook(self, request: Any) -> Any:
         """Handle a POST /webhook request."""
+        # --- Rate limiting ---
+        remote_ip = request.remote or "unknown"
+        if self._is_rate_limited(remote_ip):
+            self.logger.warning(f"Webhook: rate limited request from {remote_ip}")
+            return aio_web.Response(
+                status=429,
+                content_type="application/json",
+                text='{"error": "Rate limit exceeded"}',
+            )
+
         # --- Auth ---
         if self.secret_token and not self._verify_token(request):
             self.logger.warning(
@@ -161,7 +206,7 @@ class WebhookService(BaseServicePlugin):
         if len(message_text) > self.max_message_length:
             message_text = message_text[: self.max_message_length]
 
-        channel: str = str(body.get("channel", "")).strip().lstrip("#")
+        channel: str = str(body.get("channel", "")).strip().removeprefix("#")
         dm_to: str = str(body.get("dm_to", "")).strip()
 
         if not channel and not dm_to:
diff --git a/modules/transmission_tracker.py b/modules/transmission_tracker.py
index 0e5b074..a12c28a 100644
--- a/modules/transmission_tracker.py
+++ b/modules/transmission_tracker.py
@@ -4,6 +4,7 @@ Transmission tracker for monitoring message transmission success
 Tracks transmitted message hashes and detects repeats from neighboring repeaters
 """
 
+import threading
 import time
 from contextlib import closing
 from dataclasses import dataclass, field
@@ -49,6 +50,9 @@ class TransmissionTracker:
         self._cleanup_interval = 60  # Run cleanup check every 60 seconds
         self._last_cleanup_time = 0.0
 
+        # Lock protects record mutations (repeat_count, repeater_prefixes, etc.)
+        self._lock = threading.Lock()
+
         # Track our bot's public key prefix (first 2 hex chars) for filtering
         self.bot_prefix: Optional[str] = None
         self._update_bot_prefix()
@@ -161,16 +165,21 @@ class TransmissionTracker:
             record = self.match_packet_hash(packet_hash, time.time())
 
         if record:
-            record.repeat_count += 1
-            if repeater_prefix:
-                record.repeater_prefixes.add(repeater_prefix)
-                # Track count per repeater
-                record.repeater_counts[repeater_prefix] = record.repeater_counts.get(repeater_prefix, 0) + 1
-            else:
-                # No prefix but still a repeat (heard by radio)
-                record.repeater_counts['_unknown'] = record.repeater_counts.get('_unknown', 0) + 1
+            with self._lock:
+                record.repeat_count += 1
+                if repeater_prefix:
+                    record.repeater_prefixes.add(repeater_prefix)
+                    # Track count per repeater
+                    record.repeater_counts[repeater_prefix] = record.repeater_counts.get(repeater_prefix, 0) + 1
+                else:
+                    # No prefix but still a repeat (heard by radio)
+                    record.repeater_counts['_unknown'] = record.repeater_counts.get('_unknown', 0) + 1
 
-            self.logger.info(f"📡 Recorded repeat for hash {packet_hash}: {record.repeat_count} repeats, {len(record.repeater_prefixes)} unique repeaters, prefixes: {sorted(record.repeater_prefixes)}")
+                repeat_count = record.repeat_count
+                unique_repeaters = len(record.repeater_prefixes)
+                prefixes = sorted(record.repeater_prefixes)
+
+            self.logger.info(f"📡 Recorded repeat for hash {packet_hash}: {repeat_count} repeats, {unique_repeaters} unique repeaters, prefixes: {prefixes}")
 
             # Update the database entry if we have a command_id
             if record.command_id and hasattr(self.bot, 'web_viewer_integration'):
diff --git a/modules/web_viewer/app.py b/modules/web_viewer/app.py
index e1cdab5..bb10d82 100644
--- a/modules/web_viewer/app.py
+++ b/modules/web_viewer/app.py
@@ -110,6 +110,9 @@ class BotDataViewer:
         )
         import secrets as _secrets
         self.app.config['SECRET_KEY'] = _secrets.token_hex(32)
+        self.app.config['SESSION_COOKIE_HTTPONLY'] = True
+        self.app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'
+        self.app.config['PERMANENT_SESSION_LIFETIME'] = 86400  # 24 hours
 
         # Flask-SocketIO configuration following 5.x best practices
         # CORS origins are configured after config is loaded; create without app for now
@@ -2234,8 +2237,20 @@ class BotDataViewer:
 
         @self.app.route('/api/stream_data', methods=['POST'])
         def api_stream_data():
-            """API endpoint for receiving real-time data from bot"""
+            """API endpoint for receiving real-time data from bot.
+
+            Requires a valid X-Stream-Token header matching the token stored
+            in DB metadata by BotIntegration.  This prevents unauthenticated
+            callers from injecting fake stream data when the web viewer is
+            network-accessible.
+            """
             try:
+                if not current_app.config.get('TESTING'):
+                    token = request.headers.get('X-Stream-Token', '')
+                    expected = self.db_manager.get_metadata('internal.stream_token') if self.db_manager else None
+                    if not expected or not token or token != expected:
+                        return jsonify({'error': 'Unauthorized'}), 401
+
                 data = request.get_json()
                 if not data:
                     return jsonify({'error': 'No data provided'}), 400
@@ -2802,14 +2817,16 @@ class BotDataViewer:
                                                                    fallback=False)
                 }
 
-                # Generate sample greeting
-                sample_greeting = settings['greeting_message'].format(sender='SampleUser')
+                # Generate sample greeting — use str.replace() instead of .format()
+                # to avoid KeyError / info leaks from user-controlled templates
+                sample_greeting = settings['greeting_message'].replace('{sender}', 'SampleUser')
                 if settings['include_mesh_info']:
-                    sample_mesh_info = settings['mesh_info_format'].format(
-                        total_contacts=100,
-                        repeaters=5,
-                        companions=95,
-                        recent_activity_24h=10
+                    sample_mesh_info = (
+                        settings['mesh_info_format']
+                        .replace('{total_contacts}', '100')
+                        .replace('{repeaters}', '5')
+                        .replace('{companions}', '95')
+                        .replace('{recent_activity_24h}', '10')
                     )
                     sample_greeting += sample_mesh_info
 
diff --git a/modules/web_viewer/integration.py b/modules/web_viewer/integration.py
index fc22821..ed860df 100644
--- a/modules/web_viewer/integration.py
+++ b/modules/web_viewer/integration.py
@@ -7,6 +7,7 @@ Provides integration between the main bot and the web viewer
 import os
 import queue
 import re
+import secrets
 import subprocess
 import sys
 import threading
@@ -40,6 +41,15 @@ class BotIntegration:
         self._drain_thread: Optional[threading.Thread] = None
         # Initialize HTTP session with connection pooling for efficient reuse
         self._init_http_session()
+        # Generate a shared secret for authenticating internal /api/stream_data calls.
+        # Stored in DB metadata so the web viewer can validate it.
+        self._stream_token = secrets.token_hex(32)
+        try:
+            self.bot.db_manager.set_metadata('internal.stream_token', self._stream_token)
+        except Exception as e:
+            self.bot.logger.debug(f"Could not persist stream token: {e}")
+        if getattr(self, 'http_session', None):
+            self.http_session.headers['X-Stream-Token'] = self._stream_token
         # Start background drain thread after table is confirmed to exist
         self._start_drain_thread()
 
@@ -81,9 +91,10 @@ class BotIntegration:
             self.http_session.mount("http://", adapter)
             self.http_session.mount("https://", adapter)
 
-            # Set default headers for keep-alive (though urllib3 handles this automatically)
+            # Set default headers for keep-alive and internal auth
             self.http_session.headers.update({
                 'Connection': 'keep-alive',
+                'X-Requested-With': 'BotIntegration',  # CSRF bypass for internal calls
             })
         except ImportError:
             # Fallback if requests is not available
@@ -423,6 +434,10 @@ class BotIntegration:
             }
 
             # Use session with connection pooling if available, otherwise fallback to requests.post
+            headers = {
+                'X-Stream-Token': self._stream_token,
+                'X-Requested-With': 'BotIntegration',
+            }
             if self.http_session:
                 try:
                     self.http_session.post(url, json=payload, timeout=1.0)
@@ -432,7 +447,7 @@ class BotIntegration:
             else:
                 import requests
                 try:
-                    requests.post(url, json=payload, timeout=1.0)
+                    requests.post(url, json=payload, timeout=1.0, headers=headers)
                     self._record_web_viewer_result(True)
                 except Exception:
                     self._record_web_viewer_result(False)
@@ -455,8 +470,12 @@ class BotIntegration:
                 'data': node_data
             }
 
+            headers = {
+                'X-Stream-Token': self._stream_token,
+                'X-Requested-With': 'BotIntegration',
+            }
             try:
-                requests.post(url, json=payload, timeout=0.5)
+                requests.post(url, json=payload, timeout=0.5, headers=headers)
                 self._record_web_viewer_result(True)
             except Exception:
                 self._record_web_viewer_result(False)
diff --git a/tests/test_scheduler_logic.py b/tests/test_scheduler_logic.py
index 61a9986..afc3c72 100644
--- a/tests/test_scheduler_logic.py
+++ b/tests/test_scheduler_logic.py
@@ -1020,12 +1020,14 @@ class TestSendScheduledMessageWrapper:
                 asyncio.run(coro)
 
         mock_loop.run_until_complete = Mock(side_effect=_run_until_complete)
+        mock_loop.close = Mock()
 
-        with patch("asyncio.get_event_loop", return_value=mock_loop):
+        with patch("asyncio.new_event_loop", return_value=mock_loop):
             with patch.object(scheduler, "_send_scheduled_message_async", side_effect=_fake_send) as mock_send:
                 scheduler.send_scheduled_message("general", "test message")
 
         mock_loop.run_until_complete.assert_called_once()
+        mock_loop.close.assert_called_once()
         mock_send.assert_called_once_with("general", "test message")
 
 
diff --git a/tests/test_web_viewer.py b/tests/test_web_viewer.py
index e8874da..61373a2 100644
--- a/tests/test_web_viewer.py
+++ b/tests/test_web_viewer.py
@@ -779,6 +779,22 @@ class TestStreamRoutes:
         )
         assert resp.status_code == 200
 
+    def test_api_stream_data_rejects_without_token_in_production(self, viewer):
+        """When TESTING is False, requests without a valid stream token are rejected."""
+        viewer.db_manager.set_metadata('internal.stream_token', 'secret-token')
+        viewer.app.config['TESTING'] = False
+        try:
+            with viewer.app.test_client() as c:
+                resp = c.post(
+                    "/api/stream_data",
+                    json={"type": "command", "data": {"cmd": "ping"}},
+                    content_type="application/json",
+                    headers={"X-Requested-With": "XMLHttpRequest"},
+                )
+            assert resp.status_code == 401
+        finally:
+            viewer.app.config['TESTING'] = True
+
 
 # ===========================================================================
 # Greeter routes