dandri/meshcore-bot
1
0
Fork 0
mirror of https://github.com/agessaman/meshcore-bot.git synced 2026-08-28 05:14:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Enhance database migration safety and improve repeater management

Browse Source 
- Added a regex pattern to validate SQLite column definitions, preventing SQL injection in the _add_column() function.
- Introduced a new validation function to ensure safe column definitions are used.
- Updated repeater_manager.py to use list() for iterating over contacts, ensuring compatibility with potential changes in the underlying data structure.
- Enhanced error handling in the MessageScheduler for better debugging during scheduler shutdown and message sending.

These changes improve the security and reliability of database operations and enhance the robustness of the repeater management system.
This commit is contained in:
agessaman
2026-03-29 20:00:44 -07:00
parent 4685ea734c
commit 507c7ad31e
9 changed files with 182 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/db_migrations.py
+17
View File
@@ -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}")
modules/repeater_manager.py
+19 -18
View File
@@ -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))
modules/scheduler.py
+15 -21
View File
@@ -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
modules/service_plugins/webhook_service.py
+47 -2
View File
@@ -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:
modules/transmission_tracker.py
+18 -9
View File
@@ -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'):
modules/web_viewer/app.py
+25 -8
View File
@@ -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
modules/web_viewer/integration.py
+22 -3
View File
@@ -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)
tests/test_scheduler_logic.py
+3 -1
View File
@@ -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")
tests/test_web_viewer.py
+16
View File
@@ -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