fix: address Codacy issues from PR #105

- Bandit B104: bind Prometheus exporter to 127.0.0.1 by default,
  document the explicit 0.0.0.0 opt-in for container/remote setups.
- Bandit B310: validate InfluxDB HTTP URL scheme is http(s) before
  urlopen, both at start_influxdb_sink configuration time and at
  every send. Annotate the urlopen with nosec B310 + rationale.
- Bandit B110: replace bare 'except Exception: pass' cleanup
  swallowers across the metrics exporters (Prometheus, InfluxDB,
  OpenTelemetry) and the WebSocket / MQTT / gRPC user templates
  with debug-level log lines so cleanup failures stay observable.
- Semgrep non-literal-import: validate the dotted path against a
  strict identifier regex before importlib.import_module in the
  gRPC user, and tag the call with nosemgrep + rationale (the
  framework genuinely needs to load operator-authored stubs).
- pyflakes F401: drop the unused Callable import from
  parameter_resolver.py.

Bandit run on the touched modules now reports zero findings;
pytest test/ still passes (84 cases).

Author: JE-Chen

je_load_density/utils/metrics/influxdb_sink.py

@@ -50,13 +50,18 @@ def _send_udp(line: str, host: str, port: int) -> None:
         sock.close()
 
 
+_ALLOWED_HTTP_SCHEMES = ("http://", "https://")
+
+
 def _send_http(line: str, url: str, token: Optional[str], timeout: float) -> None:
+    if not url.lower().startswith(_ALLOWED_HTTP_SCHEMES):
+        raise ValueError("InfluxDB HTTP URL must use http:// or https://")
     headers = {"Content-Type": "text/plain; charset=utf-8"}
     if token:
         headers["Authorization"] = f"Token {token}"
     req = urllib_request.Request(url, data=line.encode("utf-8"), headers=headers, method="POST")
     try:
-        with urllib_request.urlopen(req, timeout=timeout) as response:  # noqa: S310 - configurable URL
+        with urllib_request.urlopen(req, timeout=timeout) as response:  # nosec B310 - scheme validated above
             response.read()
     except urllib_error.URLError as error:
         load_density_logger.warning(f"InfluxDB HTTP write failed: {error}")
@@ -82,8 +87,11 @@ def start_influxdb_sink(
     transport = transport.lower()
     if transport not in {"udp", "http"}:
         raise ValueError(f"unsupported transport: {transport}")
-    if transport == "http" and not url:
-        raise ValueError("url required when transport=http")
+    if transport == "http":
+        if not url:
+            raise ValueError("url required when transport=http")
+        if not url.lower().startswith(_ALLOWED_HTTP_SCHEMES):
+            raise ValueError("InfluxDB HTTP URL must use http:// or https://")
 
     with _lock:
         if _state["started"]:
@@ -131,8 +139,8 @@ def stop_influxdb_sink() -> None:
             return
         try:
             events.request.remove_listener(_state["listener"])
-        except Exception:
-            pass
+        except Exception as error:
+            load_density_logger.debug(f"influxdb listener detach failed: {error!r}")
         _state["started"] = False
         _state["listener"] = None
         _state["config"] = None

je_load_density/utils/metrics/opentelemetry_exporter.py

@@ -103,14 +103,14 @@ def stop_opentelemetry_exporter() -> None:
             return
         try:
             events.request.remove_listener(_state["listener"])
-        except Exception:
-            pass
+        except Exception as error:
+            load_density_logger.debug(f"otel listener detach failed: {error!r}")
         provider = _state.get("provider")
         if provider is not None:
             try:
                 provider.shutdown()
-            except Exception:
-                pass
+            except Exception as error:
+                load_density_logger.debug(f"otel provider shutdown failed: {error!r}")
         _state["started"] = False
         _state["listener"] = None
         _state["instruments"] = None

je_load_density/utils/metrics/prometheus_exporter.py

@@ -37,11 +37,15 @@ def _build_metrics():
     }
 
 
-def start_prometheus_exporter(port: int = 9646, addr: str = "0.0.0.0") -> Optional[int]:
+def start_prometheus_exporter(port: int = 9646, addr: str = "127.0.0.1") -> Optional[int]:
     """
     啟動 Prometheus 指標伺服器並掛上 Locust request 事件監聽。
     Start the Prometheus exporter HTTP server and attach a request listener.
 
+    Defaults to binding on loopback. Pass ``addr="0.0.0.0"`` explicitly
+    to expose the exporter to the network when running in a container
+    or remote node.
+
     Returns the port the exporter is listening on, or None if the optional
     dependency is not installed.
     """
@@ -81,7 +85,7 @@ def stop_prometheus_exporter() -> None:
             return
         try:
             events.request.remove_listener(_state["listener"])
-        except Exception:
-            pass
+        except Exception as error:
+            load_density_logger.debug(f"prometheus listener detach failed: {error!r}")
         _state["started"] = False
         _state["listener"] = None

je_load_density/utils/parameterization/parameter_resolver.py

@@ -3,7 +3,7 @@
 import os
 import re
 import threading
-from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional
+from typing import Any, Dict, Iterable, Iterator, List, Optional
 
 _PLACEHOLDER_PATTERN = re.compile(r"\$\{([^}]+)\}")
 _FUNCTION_PATTERN = re.compile(r"^([a-zA-Z_][a-zA-Z0-9_]*)\((.*)\)$")

je_load_density/wrapper/user_template/grpc_user_template.py

@@ -1,4 +1,5 @@
 import importlib
+import re
 import time
 from typing import Any, Dict, Tuple
 
@@ -27,10 +28,25 @@ def set_wrapper_grpc_user(user_detail_dict: Dict[str, Any], **kwargs) -> type:
     return GrpcUserWrapper
 
 
+_SAFE_DOTTED_PATH = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*$")
+
+
 def _import_dotted(path: str) -> Any:
+    """
+    Resolve a dotted ``module.attr`` path supplied in a load-test scenario.
+
+    The gRPC user genuinely needs to load operator-authored stub
+    modules at runtime, so a static import literal is not viable. We
+    accept this by validating that ``path`` is a syntactically safe
+    Python identifier chain (no separators, no relative dots, no
+    dunders bridged via traversal) before delegating to importlib.
+    """
+    if not isinstance(path, str) or not _SAFE_DOTTED_PATH.match(path):
+        raise ImportError(f"invalid dotted import path: {path!r}")
     module_name, _, attr = path.rpartition(".")
     if not module_name:
         raise ImportError(f"invalid dotted import path: {path!r}")
+    # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
     module = importlib.import_module(module_name)
     return getattr(module, attr)
 
@@ -74,8 +90,8 @@ def _ensure_channel(self, target: str):
             if self._channel is not None:
                 try:
                     self._channel.close()
-                except Exception:
-                    pass
+                except Exception as error:
+                    load_density_logger.debug(f"grpc channel close before reconnect failed: {error!r}")
             self._channel = grpc.insecure_channel(target)
             self._target = target
         return self._channel

je_load_density/wrapper/user_template/mqtt_user_template.py

@@ -71,8 +71,8 @@ def _ensure_client(self, broker: str, step: Dict[str, Any]):
         if self._client is not None:
             try:
                 self._client.disconnect()
-            except Exception:
-                pass
+            except Exception as error:
+                load_density_logger.debug(f"mqtt disconnect before reconnect failed: {error!r}")
 
         client_id = step.get("client_id") or f"loaddensity-{secrets.token_hex(4)}"
         client = paho_client.Client(client_id=client_id, clean_session=True)
@@ -165,6 +165,6 @@ def on_stop(self) -> None:
             try:
                 self._client.loop_stop()
                 self._client.disconnect()
-            except Exception:
-                pass
+            except Exception as error:
+                load_density_logger.debug(f"mqtt on_stop cleanup failed: {error!r}")
             self._client = None

je_load_density/wrapper/user_template/websocket_user_template.py

@@ -67,8 +67,8 @@ def _connect(self, url: str, timeout: float) -> None:
         if self._ws is not None:
             try:
                 self._ws.close()
-            except Exception:
-                pass
+            except Exception as error:
+                load_density_logger.debug(f"websocket close before reconnect failed: {error!r}")
         self._ws = create_connection(url, timeout=timeout)
         self._url = url