dont allow yaml tags in dict

Author: martvanrijthoven

pyamlo/sources.py

@@ -8,72 +8,6 @@
 from pyamlo.security import SecurityPolicy
 
 
-def _process_dict_tags(data: Any, security_policy: SecurityPolicy) -> Any:
-    """Process dictionary data - handle both raw dicts and dicts with YAML tag strings."""
-    
-    def has_yaml_tags(obj):
-        """Check if the data structure contains YAML tag-like strings."""
-        if isinstance(obj, dict):
-            return any(has_yaml_tags(v) for v in obj.values())
-        elif isinstance(obj, list):
-            return any(has_yaml_tags(item) for item in obj)
-        elif isinstance(obj, str):
-            return obj.startswith('!')
-        return False
-    
-    # If the dict contains no YAML tag strings, it's already fully resolved
-    if not has_yaml_tags(data):
-        return data
-    
-    # Otherwise, process YAML tags using the YAML stream approach (simpler than direct parsing)
-    from io import StringIO
-    import yaml
-    
-    def dict_to_yaml_text(obj, indent=0):
-        """Convert dict to YAML text without escaping tags."""
-        def format_value(val):
-            if isinstance(val, str):
-                # Don't quote strings that are YAML tags
-                if val.startswith('!'):
-                    return val
-                # Quote strings to preserve them as strings
-                dumped = yaml.safe_dump(val, default_flow_style=True).strip()
-                # Remove document separators
-                if dumped.endswith('\n...'):
-                    dumped = dumped[:-4]
-                elif dumped.endswith('...'):
-                    dumped = dumped[:-3]
-                return dumped
-            return str(val)
-        
-        if isinstance(obj, dict):
-            lines = []
-            for key, value in obj.items():
-                if isinstance(value, dict):
-                    lines.append("  " * indent + f"{key}:")
-                    lines.append(dict_to_yaml_text(value, indent + 1))
-                elif isinstance(value, list):
-                    lines.append("  " * indent + f"{key}:")
-                    for item in value:
-                        if isinstance(item, dict):
-                            lines.append("  " * (indent + 1) + "-")
-                            lines.append(dict_to_yaml_text(item, indent + 2))
-                        else:
-                            lines.append("  " * (indent + 1) + f"- {format_value(item)}")
-                else:
-                    lines.append("  " * indent + f"{key}: {format_value(value)}")
-            return '\n'.join(lines)
-        return str(obj)
-    
-    # Convert to YAML text and parse with ConfigLoader
-    yaml_text = dict_to_yaml_text(data)
-    loader = ConfigLoader(StringIO(yaml_text), security_policy=security_policy)
-    try:
-        return loader.get_single_data()
-    finally:
-        loader.dispose()
-
-
 def _load_source(
     source: Union[str, Path, IO[str]], security_policy: SecurityPolicy
 ) -> dict[str, Any]:
@@ -93,7 +27,8 @@ def _process_single_source(
     src: Union[str, Path, IO[str], dict], security_policy: SecurityPolicy
 ) -> dict[str, Any]:
     if isinstance(src, dict):
-        raw = _process_dict_tags(src.copy(), security_policy)
+        # Dict sources contain already resolved Python objects, use as-is
+        raw = src.copy()
         src_path = "<dict>"
     else:
         raw = _load_source(src, security_policy=security_policy)

tests/test_dict_sources.py

@@ -110,18 +110,22 @@ def test_dict_with_interpolation():
 
 
 def test_dict_with_object_instantiation():
-    """Test dictionary source with object instantiation."""
+    """Test dictionary source with actual Python objects."""
+    from pathlib import Path
+    
     config_dict = {
         "paths": {
-            "base": "!@pathlib.Path /tmp/test",
-            "data": "!@pathlib.Path /tmp/test/data.txt"
+            "base": Path("/tmp/test"),
+            "data": Path("/tmp/test/data.txt")
         }
     }
 
     result = load_config(config_dict)
 
     assert str(result["paths"]["base"]) == "/tmp/test"
     assert str(result["paths"]["data"]) == "/tmp/test/data.txt"
+    assert isinstance(result["paths"]["base"], Path)
+    assert isinstance(result["paths"]["data"], Path)
 
 
 def test_empty_dict_source():
@@ -131,25 +135,28 @@ def test_empty_dict_source():
 
 
 def test_dict_with_security_policy():
-    """Test dictionary source with restrictive security policy."""
+    """Test dictionary source with actual Python objects and security policy."""
+    from pathlib import Path
+    
     config_dict = {
         "safe": {
             "value": "test"
         },
-        "unsafe": {
-            "path": "!@pathlib.Path /tmp"
+        "paths": {
+            "path": Path("/tmp")
         }
     }
 
     restrictive_policy = SecurityPolicy(restrictive=True)
 
-    # Should work with non-restrictive policy
-    result = load_config(config_dict)
-    assert str(result["unsafe"]["path"]) == "/tmp"
+    # Should work with both policies since dict contains actual objects, not tags
+    result1 = load_config(config_dict)
+    assert str(result1["paths"]["path"]) == "/tmp"
+    assert isinstance(result1["paths"]["path"], Path)
 
-    # Should fail with restrictive policy
-    with pytest.raises(Exception):
-        load_config(config_dict, security_policy=restrictive_policy)
+    result2 = load_config(config_dict, security_policy=restrictive_policy)
+    assert str(result2["paths"]["path"]) == "/tmp"
+    assert isinstance(result2["paths"]["path"], Path)
 
 
 def test_nested_dict_merging():

tests/test_nested_interpolation_bug.py

@@ -95,17 +95,21 @@ def test_deep_nested_interpolation():
 
 def test_object_property_access():
     """Test accessing properties of instantiated objects."""
-    config1 = {
-        "stage": {
-            "step": {
-                "component1": {
-                    "path": "!@pathlib.Path /tmp/test/example.txt",
-                    "version": 1.0,
-                }
-            }
-        }
-    }
+    from io import StringIO
+    
+    # Step 1: Build config with YAML tags from YAML source
+    config1_yaml = """
+stage:
+  step:
+    component1:
+      path: !@pathlib.Path /tmp/test/example.txt
+      version: 1.0
+"""
+    
+    # Load and resolve the YAML config first
+    config1_built = load_config(StringIO(config1_yaml))
 
+    # Step 2: Use the built config with dict containing interpolations
     config2 = {
         "stage": {
             "step": {
@@ -118,7 +122,7 @@ def test_object_property_access():
         }
     }
 
-    result = load_config([config1, config2])
+    result = load_config([config1_built, config2])
 
     # Check that the object was created correctly
     path_obj = result["stage"]["step"]["component1"]["path"]