Fix Trivy input handling

socket_basics/core/config.py

@@ -1006,6 +1006,23 @@ def merge_json_and_env_config(json_config: Dict[str, Any] | None = None) -> Dict
     """
     # Start with environment defaults (lowest priority)
     config = load_config_from_env()
+
+    def _filter_empty_string_overrides(incoming: Dict[str, Any], source_name: str) -> Dict[str, Any]:
+        """Prevent empty string values from erasing existing non-empty config values."""
+        logger = logging.getLogger(__name__)
+        filtered: Dict[str, Any] = {}
+        for k, v in incoming.items():
+            if isinstance(v, str) and v == '':
+                current = config.get(k)
+                if current not in (None, ''):
+                    logger.debug(
+                        "Skipping empty %s override for '%s' (keeping existing value)",
+                        source_name,
+                        k,
+                    )
+                    continue
+            filtered[k] = v
+        return filtered
 
     # Override with Socket Basics API config if no explicit JSON config provided
     # API config takes precedence over environment defaults
@@ -1017,15 +1034,7 @@ def merge_json_and_env_config(json_config: Dict[str, Any] | None = None) -> Dict
         if socket_basics_config:
             # Normalize camelCase API keys to snake_case internal format
             normalized_config = normalize_api_config(socket_basics_config)
-            # Filter out empty strings for rule configs - treat them as unset to use code defaults
-            # Only filter string values ending with _enabled_rules or _disabled_rules
-            filtered_config = {}
-            for k, v in normalized_config.items():
-                if isinstance(v, str) and v == '' and (k.endswith('_enabled_rules') or k.endswith('_disabled_rules')):
-                    # Skip empty rule config strings - they'll fall back to defaults
-                    logger.debug(f"Filtering out empty rule config: {k}")
-                    continue
-                filtered_config[k] = v
+            filtered_config = _filter_empty_string_overrides(normalized_config, 'API config')
             config.update(filtered_config)
             logging.getLogger(__name__).info("Loaded Socket Basics API configuration (overrides environment defaults)")
         else:
@@ -1036,14 +1045,7 @@ def merge_json_and_env_config(json_config: Dict[str, Any] | None = None) -> Dict
     if json_config:
         # Also normalize JSON config in case it comes from API
         normalized_json = normalize_api_config(json_config)
-        # Filter out empty strings for rule configs - treat them as unset to use code defaults
-        filtered_json = {}
-        for k, v in normalized_json.items():
-            if isinstance(v, str) and v == '' and (k.endswith('_enabled_rules') or k.endswith('_disabled_rules')):
-                # Skip empty rule config strings - they'll fall back to defaults
-                logger.debug(f"Filtering out empty rule config: {k}")
-                continue
-            filtered_json[k] = v
+        filtered_json = _filter_empty_string_overrides(normalized_json, 'JSON config')
         config.update(filtered_json)
         logging.getLogger(__name__).info("Loaded JSON configuration (overrides environment defaults)")
 

socket_basics/core/connector/trivy/trivy.py

@@ -15,6 +15,18 @@
 # Import shared formatters
 from ...formatters import get_all_formatters
 
+
+def _to_bool(value: Any) -> bool:
+    """Normalize config values that may arrive as bool, string, or number."""
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (int, float)):
+        return value != 0
+    if isinstance(value, str):
+        return value.strip().lower() in {'true', '1', 'yes', 'on'}
+    return bool(value)
+
+
 class TrivyScanner(BaseConnector):
     """Trivy container scanner implementation"""
 
@@ -27,13 +39,13 @@ def is_enabled(self) -> bool:
         Returns True if either Dockerfile, container image, or vulnerability scanning is enabled.
         This method supports both the new parameter names and legacy ones.
         """
-        dockerfile_flag = bool(
+        dockerfile_flag = _to_bool(
             self.config.get('dockerfile_scanning_enabled', False) or self.config.get('dockerfile_enabled', False))
-        image_flag = bool(
+        image_flag = _to_bool(
             self.config.get('container_image_scanning_enabled', False) or self.config.get('image_enabled', False))
 
         # Check if Trivy vulnerability scanning is enabled
-        vuln_flag = bool(self.config.get('trivy_vuln_enabled', False))
+        vuln_flag = _to_bool(self.config.get('trivy_vuln_enabled', False))
 
         return dockerfile_flag or image_flag or vuln_flag
 
@@ -125,8 +137,9 @@ def scan(self) -> Dict[str, Any]:
     def scan_dockerfiles(self) -> Dict[str, Any]:
         """Run Trivy Dockerfile scanning"""
         # Consider both new and legacy dockerfile flags
-        dockerfile_enabled = self.config.get('dockerfile_scanning_enabled', False) or self.config.get(
-            'dockerfile_enabled', False)
+        dockerfile_enabled = _to_bool(
+            self.config.get('dockerfile_scanning_enabled', False) or self.config.get('dockerfile_enabled', False)
+        )
         if not dockerfile_enabled:
             logger.info("Dockerfile scanning disabled, skipping Trivy Dockerfile")
             return {}
@@ -260,7 +273,11 @@ def scan_images(self) -> Dict[str, Any]:
             return {}
 
         # Consider both new and legacy image flags (auto-enabled if images provided)
-        image_enabled = self.config.get('container_image_scanning_enabled', False) or self.config.get('image_enabled', False) or bool(images)
+        image_enabled = (
+            _to_bool(self.config.get('container_image_scanning_enabled', False))
+            or _to_bool(self.config.get('image_enabled', False))
+            or bool(images)
+        )
         if not image_enabled:
             logger.info("Image scanning disabled, skipping Trivy Image")
             return {}
@@ -308,7 +325,7 @@ def scan_images(self) -> Dict[str, Any]:
 
     def scan_vulnerabilities(self) -> Dict[str, Any]:
         """Run Trivy filesystem scanning for vulnerabilities"""
-        vuln_enabled = self.config.get('trivy_vuln_enabled', False)
+        vuln_enabled = _to_bool(self.config.get('trivy_vuln_enabled', False))
 
         if not vuln_enabled:
             logger.info("Trivy vulnerability scanning disabled, skipping")
@@ -1096,4 +1113,4 @@ def generate_notifications(self, components: List[Dict[str, Any]], item_name: st
 
     def get_name(self) -> str:
         """Return the display name for this connector"""
-        return "Trivy"
+        return "Trivy"

tests/test_config_trivy_regressions.py

@@ -0,0 +1,46 @@
+from pathlib import Path
+
+from socket_basics.core import config as config_mod
+from socket_basics.core.connector.trivy.trivy import TrivyScanner
+
+
+class _DummyConfig:
+    def __init__(self, values, workspace):
+        self._config = values
+        self.workspace = workspace
+
+    def get(self, key, default=None):
+        return self._config.get(key, default)
+
+
+def test_api_empty_dockerfiles_does_not_override_env(monkeypatch):
+    monkeypatch.setattr(config_mod, "load_config_from_env", lambda: {"dockerfiles": "Dockerfile"})
+    monkeypatch.setattr(config_mod, "load_socket_basics_config", lambda: {"dockerfiles": ""})
+
+    merged = config_mod.merge_json_and_env_config()
+
+    assert merged["dockerfiles"] == "Dockerfile"
+
+
+def test_api_non_empty_dockerfiles_still_overrides_env(monkeypatch):
+    monkeypatch.setattr(config_mod, "load_config_from_env", lambda: {"dockerfiles": "Dockerfile"})
+    monkeypatch.setattr(config_mod, "load_socket_basics_config", lambda: {"dockerfiles": "infra/Dockerfile"})
+
+    merged = config_mod.merge_json_and_env_config()
+
+    assert merged["dockerfiles"] == "infra/Dockerfile"
+
+
+def test_trivy_string_false_disables_vuln_scan(tmp_path):
+    cfg = _DummyConfig({"trivy_vuln_enabled": "false"}, Path(tmp_path))
+    scanner = TrivyScanner(cfg)
+
+    assert scanner.is_enabled() is False
+    assert scanner.scan_vulnerabilities() == {}
+
+
+def test_trivy_string_true_enables_vuln_scan_flag(tmp_path):
+    cfg = _DummyConfig({"trivy_vuln_enabled": "true"}, Path(tmp_path))
+    scanner = TrivyScanner(cfg)
+
+    assert scanner.is_enabled() is True