feat: Major security and reliability improvements for v2.0.0

Security fixes:
- Add HTTP request timeout (10s) to prevent hanging
- Add base64 decode size limit (1MB) to prevent DoS
- Validate HTTP methods against whitelist
- Fix expression variable naming to escape brackets

Reliability improvements:
- Add SIGTERM/SIGINT signal handling for graceful shutdown
- Add resource cleanup (MQTT disconnect, InfluxDB close)
- Improve exception handling with specific error types
- Add timestamp fallback logging

Configuration validation:
- Require at least one field per point
- Validate TLS cert/key relationship
- Validate MQTT topic patterns
- Validate cron schedule ranges
- Validate field types

New features:
- Add QoS configuration per point (0-2)
- Add binary payload support ($.payload_binary.hex)
- Safer debug logging (payload size, not content)

Author: hubpav

mqtt2influxdb/config.py

@@ -4,18 +4,18 @@
 import os
 import re
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 
 import jsonpath_ng
 import yaml
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from .expr import parse_expression
 
 # Regex for environment variable substitution: ${VAR} or ${VAR:default}
 ENV_VAR_REGEX = re.compile(r"\$\{([^}:]+)(?::([^}]*))?\}")
 
-# Regex for schedule config entries
+# Regex for schedule config entries (basic syntax validation)
 CRONTAB_REGEX = re.compile(
     r"(?P<minute>\*|[0-5]?\d|\*/\d+|\d+-\d+|\d+(,\d+)*)\s+"
     r"(?P<hour>\*|[01]?\d|2[0-3]|\*/\d+|\d+-\d+|\d+(,\d+)*)\s+"
@@ -24,6 +24,18 @@
     r"(?P<day_of_week>\*|[0-6](-[0-6])?|\*/\d+|\d+(,\d+)*)"
 )
 
+# Valid MQTT topic pattern (no empty segments, valid wildcards)
+MQTT_TOPIC_REGEX = re.compile(
+    r"^(?:[^/#+]+|\+)(?:/(?:[^/#+]+|\+))*(?:/#)?$|"  # Normal topics with + and trailing #
+    r"^#$"  # Just # is valid
+)
+
+# Allowed HTTP methods for forwarding
+ALLOWED_HTTP_METHODS = frozenset({"get", "post", "put", "patch", "delete", "head"})
+
+# Allowed field types for type conversion
+ALLOWED_FIELD_TYPES = frozenset({"float", "int", "str", "bool", "booltoint"})
+
 
 class ConfigError(Exception):
     """Configuration validation error."""
@@ -51,6 +63,22 @@ def validate_file_exists(cls, v: Path | None) -> Path | None:
             raise ValueError(f"File not found: {v}")
         return v
 
+    @model_validator(mode="after")
+    def validate_tls_config(self) -> "MqttConfig":
+        """Validate TLS certificate configuration consistency."""
+        # If client cert is provided, key must also be provided (and vice versa)
+        if self.certfile and not self.keyfile:
+            raise ValueError("certfile requires keyfile to be set")
+        if self.keyfile and not self.certfile:
+            raise ValueError("keyfile requires certfile to be set")
+        # If using client certs, CA file should typically be set
+        if self.certfile and not self.cafile:
+            logging.warning(
+                "certfile/keyfile set without cafile - "
+                "server certificate will not be verified"
+            )
+        return self
+
 
 class InfluxDBConfig(BaseModel):
     """InfluxDB v3 configuration."""
@@ -75,6 +103,24 @@ class HttpConfig(BaseModel):
     username: str | None = Field(default=None, min_length=1)
     password: str | None = Field(default=None, min_length=1)
 
+    @field_validator("action", mode="after")
+    @classmethod
+    def validate_http_method(cls, v: str) -> str:
+        """Validate HTTP method against allowed methods."""
+        if v.lower() not in ALLOWED_HTTP_METHODS:
+            allowed = ", ".join(sorted(ALLOWED_HTTP_METHODS))
+            raise ValueError(f"Invalid HTTP method: {v}. Allowed: {allowed}")
+        return v
+
+    @model_validator(mode="after")
+    def validate_auth_config(self) -> "HttpConfig":
+        """Validate HTTP authentication configuration consistency."""
+        if self.username and not self.password:
+            raise ValueError("username requires password to be set")
+        if self.password and not self.username:
+            raise ValueError("password requires username to be set")
+        return self
+
 
 class Base64DecodeConfig(BaseModel):
     """Base64 decode configuration."""
@@ -93,6 +139,15 @@ class FieldConfig(BaseModel):
     value: str = Field(..., min_length=1)
     type: str | None = Field(default=None, min_length=1)
 
+    @field_validator("type", mode="after")
+    @classmethod
+    def validate_field_type(cls, v: str | None) -> str | None:
+        """Validate field type against allowed types."""
+        if v is not None and v not in ALLOWED_FIELD_TYPES:
+            allowed = ", ".join(sorted(ALLOWED_FIELD_TYPES))
+            raise ValueError(f"Invalid field type: {v}. Allowed: {allowed}")
+        return v
+
 
 class PointConfig(BaseModel):
     """Measurement point configuration."""
@@ -103,19 +158,76 @@ class PointConfig(BaseModel):
     topic: str = Field(..., min_length=1)
     bucket: str | None = Field(default=None, min_length=1)
     schedule: str | None = Field(default=None, min_length=1)
-    fields: dict[str, str | FieldConfig] = Field(default_factory=dict)
+    qos: int = Field(default=0, ge=0, le=2, description="MQTT QoS level (0, 1, or 2)")
+    fields: dict[str, str | FieldConfig] = Field(..., min_length=1)
     tags: dict[str, str] = Field(default_factory=dict)
     httpcontent: dict[str, str] = Field(default_factory=dict)
 
+    @field_validator("topic", mode="after")
+    @classmethod
+    def validate_topic_pattern(cls, v: str) -> str:
+        """Validate MQTT topic pattern syntax."""
+        # Check for empty segments
+        if "//" in v:
+            raise ValueError(f"Invalid topic pattern (empty segment): {v}")
+        # Check for invalid wildcard usage
+        segments = v.split("/")
+        for i, segment in enumerate(segments):
+            # + must be alone in its segment
+            if "+" in segment and segment != "+":
+                raise ValueError(
+                    f"Invalid topic pattern (+ must be alone in segment): {v}"
+                )
+            # # must be the last segment and alone
+            if "#" in segment:
+                if segment != "#" or i != len(segments) - 1:
+                    raise ValueError(
+                        f"Invalid topic pattern (# must be last and alone): {v}"
+                    )
+        logging.debug("Validated MQTT topic pattern: '%s'", v)
+        return v
+
     @field_validator("schedule", mode="after")
     @classmethod
     def validate_schedule(cls, v: str | None) -> str | None:
+        """Validate cron schedule syntax."""
         if v is not None:
             if not CRONTAB_REGEX.match(v):
                 raise ValueError(f"Invalid cron format: {v}")
+            # Additional semantic validation
+            parts = v.split()
+            if len(parts) == 5:
+                minute, hour, day, month, dow = parts
+                # Validate ranges make sense
+                cls._validate_cron_range(day, 1, 31, "day")
+                cls._validate_cron_range(month, 1, 12, "month")
+                cls._validate_cron_range(dow, 0, 6, "day_of_week")
             logging.debug("Validated crontab entry: '%s'", v)
         return v
 
+    @staticmethod
+    def _validate_cron_range(value: str, min_val: int, max_val: int, name: str) -> None:
+        """Validate individual cron field ranges."""
+        if value == "*" or value.startswith("*/"):
+            return
+        # Handle ranges like 1-15
+        if "-" in value and "," not in value:
+            try:
+                start, end = map(int, value.split("-"))
+                if start > end:
+                    raise ValueError(
+                        f"Invalid cron {name} range: {start}-{end} (start > end)"
+                    )
+                if start < min_val or end > max_val:
+                    raise ValueError(
+                        f"Invalid cron {name} range: {value} (must be {min_val}-{max_val})"
+                    )
+            except ValueError as e:
+                if "Invalid cron" in str(e):
+                    raise
+                # Not a simple range, skip validation
+                pass
+
     @field_validator("measurement", mode="after")
     @classmethod
     def validate_measurement(cls, v: str) -> str:

mqtt2influxdb/expr.py

@@ -1,5 +1,7 @@
 """Expression parsing utilities for mqtt2influxdb."""
 
+import re
+
 import py_expression_eval
 
 
@@ -9,29 +11,46 @@ class ExpressionError(ValueError):
     pass
 
 
+# Regex to match array bracket notation like [0], [1], etc.
+BRACKET_REGEX = re.compile(r"\[(\d+)\]")
+
+
 def jsonpath_to_variable(path: str) -> str:
     """Convert JSONPath ($.) to valid expression variable (JSON__).
 
+    Handles array bracket notation by converting [n] to _n_.
+    Example: $.topic[1] -> JSON__topic_1_
+
     Args:
         path: JSONPath expression starting with $.
 
     Returns:
         Valid expression variable name.
     """
-    return path.replace("$", "JSON_").replace(".", "_")
+    # First convert brackets [n] to _n_ to avoid parser confusion
+    result = BRACKET_REGEX.sub(r"_\1_", path)
+    # Then do the standard replacements
+    return result.replace("$", "JSON_").replace(".", "_")
 
 
 def variable_to_jsonpath(var) -> str:
     """Convert expression variable back to JSONPath.
 
+    Reverses the conversion done by jsonpath_to_variable.
+    Example: JSON__topic_1_ -> $.topic[1]
+
     Args:
         var: Expression variable (either string or object with .var attribute).
 
     Returns:
         JSONPath expression starting with $.
     """
     name = var.var if hasattr(var, "var") else str(var)
-    return name.replace("JSON_", "$").replace("_", ".")
+    # First do standard replacements
+    result = name.replace("JSON_", "$").replace("_", ".")
+    # Then convert back _n_ patterns to [n] (now they are .n.)
+    result = re.sub(r"\.(\d+)\.", r"[\1]", result)
+    return result
 
 
 def parse_expression(text: str) -> py_expression_eval.Expression: