fix(keras): prevent spoofed built-in registered_name from hiding non-allowlisted modules

CHANGELOG.md

@@ -162,6 +162,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **security:** keep spoofed built-in Keras `registered_name` values from hiding non-allowlisted custom modules in `.keras` ZIP scans
+- **keras:** suppress duplicate custom-object warnings for allowlisted registered objects when module metadata is absent
 - **security:** stop auto-applying local `.modelaudit.toml` and `pyproject.toml` rule config during scans unless a human explicitly trusts that config in an interactive scan; remembered trust is stored securely under the local ModelAudit cache and invalidated when the config changes
 - **telemetry:** preserve secret-scrubbed model references in telemetry payloads while omitting raw credentials, query strings, and local directory paths
 - **cli:** preserve original local files during `--stream` directory scans instead of unlinking them after analysis

modelaudit/scanners/keras_zip_scanner.py

@@ -135,47 +135,60 @@ def _is_allowlisted_keras_module(module_value: Any) -> bool:
             return False
         return module_value.strip().split(".")[0] in _SAFE_KERAS_MODULE_ROOTS
 
-    def _layer_uses_allowlisted_module(self, layer: dict[str, Any]) -> bool:
+    def _iter_layer_module_references(self, layer: dict[str, Any]) -> list[str]:
         layer_config = layer.get("config", {})
         if not isinstance(layer_config, dict):
             layer_config = {}
 
+        module_references: list[str] = []
         for key in ("module", "fn_module"):
-            if self._is_allowlisted_keras_module(layer.get(key)):
-                return True
-            if self._is_allowlisted_keras_module(layer_config.get(key)):
-                return True
-        return False
+            for value in (layer.get(key), layer_config.get(key)):
+                if isinstance(value, str) and value.strip():
+                    module_references.append(value.strip())
+        return module_references
+
+    def _layer_uses_allowlisted_module(self, layer: dict[str, Any]) -> bool:
+        return any(
+            self._is_allowlisted_keras_module(module_value)
+            for module_value in self._iter_layer_module_references(layer)
+        )
+
+    def _layer_uses_non_allowlisted_module(self, layer: dict[str, Any]) -> bool:
+        return any(
+            not self._is_allowlisted_keras_module(module_value)
+            for module_value in self._iter_layer_module_references(layer)
+        )
 
     @staticmethod
     def _is_known_safe_allowlisted_registered_object(identifier: Any) -> bool:
         return isinstance(identifier, str) and identifier.strip().lower() in _SAFE_ALLOWLISTED_REGISTERED_OBJECTS
 
     def _is_known_safe_serialized_layer(self, layer: dict[str, Any]) -> bool:
         layer_class = layer.get("class_name")
-        if is_known_safe_keras_layer_class(layer_class):
-            return True
-
-        return self._layer_uses_allowlisted_module(layer) and self._is_known_safe_allowlisted_registered_object(
+        if is_known_safe_keras_layer_class(layer_class) or self._is_known_safe_allowlisted_registered_object(
             layer_class
-        )
+        ):
+            return not self._layer_uses_non_allowlisted_module(layer)
+
+        return False
 
     def _should_flag_registered_object(self, layer: dict[str, Any]) -> bool:
         registered_name = layer.get("registered_name")
         if not isinstance(registered_name, str) or not registered_name.strip():
             return False
 
-        if is_known_safe_keras_layer_class(registered_name):
-            return False
-
+        normalized_registered_name = registered_name.strip()
+        has_non_allowlisted_module = self._layer_uses_non_allowlisted_module(layer)
         layer_class = layer.get("class_name")
-        if isinstance(layer_class, str) and registered_name.strip() == layer_class.strip():
-            if self._is_known_safe_serialized_layer(layer):
-                return False
-            return not (
-                self._layer_uses_allowlisted_module(layer)
-                and self._is_known_safe_allowlisted_registered_object(layer_class)
-            )
+        if isinstance(layer_class, str) and normalized_registered_name == layer_class.strip():
+            if self._is_known_safe_serialized_layer(layer) or self._is_known_safe_allowlisted_registered_object(
+                layer_class
+            ):
+                return has_non_allowlisted_module
+            return True
+
+        if is_known_safe_keras_layer_class(normalized_registered_name):
+            return has_non_allowlisted_module
 
         return True
 

tests/scanners/test_keras_zip_scanner.py

@@ -875,6 +875,97 @@ def test_registered_builtin_layer_does_not_false_positive(self, tmp_path: Path)
         assert all(check.name != "Custom Layer Class Detection" for check in result.checks)
         assert all(check.name != "Custom Object Detection" for check in result.checks)
 
+    def test_builtin_registered_name_with_non_allowlisted_module_is_flagged(self, tmp_path: Path) -> None:
+        """Spoofed built-in registered names must not hide custom modules."""
+        scanner = KerasZipScanner()
+        config = {
+            "class_name": "Functional",
+            "config": {
+                "layers": [
+                    {
+                        "class_name": "InputLayer",
+                        "name": "input_1",
+                        "config": {"batch_shape": [None, 4]},
+                    },
+                    {
+                        "class_name": "Add",
+                        "name": "spoofed_add",
+                        "module": "evil.module",
+                        "registered_name": "Add",
+                        "config": {},
+                    },
+                ]
+            },
+        }
+
+        result = scanner.scan(
+            str(create_configured_keras_zip(tmp_path, config, file_name="spoofed_builtin_registered.keras"))
+        )
+
+        custom_object_checks = [check for check in result.checks if check.name == "Custom Object Detection"]
+        assert len(custom_object_checks) == 1
+        assert custom_object_checks[0].details["registered_name"] == "Add"
+
+    def test_builtin_registered_name_with_mixed_module_refs_is_flagged(self, tmp_path: Path) -> None:
+        """A safe module reference must not suppress a separate unsafe one."""
+        scanner = KerasZipScanner()
+        config = {
+            "class_name": "Functional",
+            "config": {
+                "layers": [
+                    {
+                        "class_name": "InputLayer",
+                        "name": "input_1",
+                        "config": {"batch_shape": [None, 4]},
+                    },
+                    {
+                        "class_name": "Add",
+                        "name": "spoofed_add",
+                        "module": "keras.src.ops.numpy",
+                        "registered_name": "Add",
+                        "config": {"fn_module": "evil.module"},
+                    },
+                ]
+            },
+        }
+
+        result = scanner.scan(
+            str(create_configured_keras_zip(tmp_path, config, file_name="mixed_builtin_registered.keras"))
+        )
+
+        custom_object_checks = [check for check in result.checks if check.name == "Custom Object Detection"]
+        assert len(custom_object_checks) == 1
+        assert custom_object_checks[0].details["registered_name"] == "Add"
+
+    def test_builtin_class_with_non_allowlisted_module_and_no_registered_name_is_flagged(self, tmp_path: Path) -> None:
+        """Built-in class names must still be flagged when module metadata points outside the allowlist."""
+        scanner = KerasZipScanner()
+        config = {
+            "class_name": "Functional",
+            "config": {
+                "layers": [
+                    {
+                        "class_name": "InputLayer",
+                        "name": "input_1",
+                        "config": {"batch_shape": [None, 4]},
+                    },
+                    {
+                        "class_name": "Add",
+                        "name": "spoofed_add",
+                        "module": "evil.module",
+                        "config": {},
+                    },
+                ]
+            },
+        }
+
+        result = scanner.scan(
+            str(create_configured_keras_zip(tmp_path, config, file_name="builtin_class_evil_module.keras"))
+        )
+
+        assert any(check.name == "Custom Layer Class Detection" for check in result.checks)
+        assert any(issue.severity in (IssueSeverity.WARNING, IssueSeverity.CRITICAL) for issue in result.issues)
+
     def test_allowlisted_module_layer_does_not_false_positive(self, tmp_path: Path) -> None:
         """Layers from allowlisted Keras modules should not be treated as custom objects."""
         scanner = KerasZipScanner()
@@ -903,6 +994,37 @@ def test_allowlisted_module_layer_does_not_false_positive(self, tmp_path: Path)
         assert all(check.name != "Custom Layer Class Detection" for check in result.checks)
         assert all(check.name != "Custom Object Detection" for check in result.checks)
 
+    def test_allowlisted_registered_object_without_module_does_not_false_positive_custom_object(
+        self, tmp_path: Path
+    ) -> None:
+        """Allowlisted registered objects should not need module metadata to suppress custom-object warnings."""
+        scanner = KerasZipScanner()
+        config = {
+            "class_name": "Functional",
+            "config": {
+                "layers": [
+                    {
+                        "class_name": "InputLayer",
+                        "name": "input_1",
+                        "config": {"batch_shape": [None, 4]},
+                    },
+                    {
+                        "class_name": "NotEqual",
+                        "name": "not_equal",
+                        "registered_name": "NotEqual",
+                        "config": {},
+                    },
+                ]
+            },
+        }
+
+        result = scanner.scan(
+            str(create_configured_keras_zip(tmp_path, config, file_name="allowlisted_registered_without_module.keras"))
+        )
+
+        assert all(check.name != "Custom Layer Class Detection" for check in result.checks)
+        assert all(check.name != "Custom Object Detection" for check in result.checks)
+
     def test_allowlisted_module_does_not_suppress_unknown_custom_layer(self, tmp_path: Path) -> None:
         """Unknown classes must still be flagged even if module metadata looks Keras-owned."""
         scanner = KerasZipScanner()