security: SSRF hardening, log injection sanitization, and allow_local_smtp

Add SSRF host validation to maintenance.py send_nightly_email and
scheduler.py send_zombie_alert_email using validate_external_url().
New allow_local_smtp config key permits private-IP SMTP for local
relay setups.

Add sanitize_name() to security_utils and apply it to all log calls
in message_handler, repeater_manager, path_command, solarforecast_command,
command_manager, and discord_bridge_service to prevent log injection.

Move nightly email logic from duplicate scheduler._send_nightly_email()
into the canonical maintenance.py implementation, removing the duplicate.
Update tests to call maintenance.send_nightly_email() directly.

Add validate_external_url allow_private parameter with support for
loopback, RFC1918, CGN, and link-local address ranges.
This commit is contained in:
Stacy Olivas authored and agessaman committed 2026-04-14 10:02:36 -07:00
1 parent c7fa0ba3d2
commit 54aeb28bf0
18 files changed
+707 -80

No files matched your search

+56 -5
View File
@@ -16,6 +16,9 @@ from urllib.parse import urlparse
logger = logging.getLogger('MeshCoreBot.Security')
# CGN (Carrier-Grade NAT) network 100.64.0.0/10 - RFC 6598
_CGN_NETWORK = ipaddress.ip_network("100.64.0.0/10")
def _is_nix_environment() -> bool:
"""
@@ -42,13 +45,19 @@ def _is_nix_environment() -> bool:
return bool(any(var in os.environ for var in nix_env_vars))
def validate_external_url(url: str, allow_localhost: bool = False, timeout: float = 2.0) -> bool:
def validate_external_url(
url: str,
allow_private: bool = False,
allow_loopback: bool | None = None, # Deprecated: use allow_private=True instead
timeout: float = 2.0
) -> bool:
"""
Validate that URL points to safe external resource (SSRF protection)
Args:
url: URL to validate
allow_localhost: Whether to allow localhost/private IPs (default: False)
allow_private: Whether to allow private/internal IPs (default: False)
allow_loopback: Deprecated alias for allow_private
timeout: DNS resolution timeout in seconds (default: 2.0)
Returns:
@@ -56,6 +65,10 @@ def validate_external_url(url: str, allow_localhost: bool = False, timeout: floa
Raises:
ValueError: If URL is invalid or unsafe
Note:
- allow_loopback=True only permits loopback addresses (127.0.0.1, ::1)
- allow_private=True permits all internal ranges (loopback, RFC1918, CGN, link-local)
"""
try:
parsed = urlparse(url)
@@ -84,17 +97,30 @@ def validate_external_url(url: str, allow_localhost: bool = False, timeout: floa
ip_obj = ipaddress.ip_address(ip)
# If localhost is not allowed, reject private/internal IPs
if not allow_localhost:
# Reject private/internal IPs
# If loopback is not allowed, reject loopback addresses
if allow_loopback is True:
# Only allow loopback, reject everything else
if not ip_obj.is_loopback:
logger.warning(f"URL resolves to non-loopback IP with allow_loopback: {ip}")
return False
elif allow_loopback is False or not allow_private:
# Reject private/internal IPs (RFC1918, CGN, link-local)
if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_link_local:
logger.warning(f"URL resolves to private/internal IP: {ip}")
return False
# Reject CGN (Carrier-Grade NAT) - RFC 6598
if ip_obj in _CGN_NETWORK:
logger.warning(f"URL resolves to CGN IP: {ip}")
return False
# Reject reserved ranges
if ip_obj.is_reserved or ip_obj.is_multicast:
logger.warning(f"URL resolves to reserved/multicast IP: {ip}")
return False
else:
# allow_private=True: allow all internal ranges
pass
except socket.gaierror as e:
logger.warning(f"Failed to resolve hostname {parsed.hostname}: {e}")
@@ -243,6 +269,31 @@ def sanitize_input(content: str, max_length: Optional[int] = 500, strip_controls
return content.strip()
def sanitize_name(name: object, max_length: int = 64) -> str:
"""
Sanitize a short identifier (node name, channel name, etc.) for safe logging and storage.
Strips all control characters including newlines, carriage returns, tabs,
null bytes, and ANSI escape sequences. Truncates to max_length.
Args:
name: Value to sanitize (coerced to str if not already).
max_length: Maximum character length (default: 64).
Returns:
Sanitized string.
Raises:
ValueError: If max_length is negative.
"""
if max_length < 0:
raise ValueError(f"max_length must be non-negative, got {max_length}")
text = str(name) if not isinstance(name, str) else name
# Strip all control characters (including \n \r \t \x00 and ANSI escapes)
text = re.sub(r'[\x00-\x1f\x7f]|\x1b\[[0-9;]*[A-Za-z]', '', text)
return text[:max_length]
# Valid SQLite journal modes for PRAGMA journal_mode validation
VALID_JOURNAL_MODES = {"DELETE", "TRUNCATE", "PERSIST", "MEMORY", "WAL", "OFF"}