fix: Restore broken stdlib calls in Python Code node (#29776)

packages/@n8n/task-runner-python/tests/integration/test_execution.py

@@ -185,6 +185,99 @@ async def test_per_item_with_continue_on_fail(broker, manager):
     assert "division by zero" in done_msg["data"]["result"][0]["json"]["error"]
 
 
+@pytest.mark.asyncio
+async def test_per_item_datetime_strptime_works(broker, manager_with_stdlib_wildcard):
+    task_id = nanoid()
+    items = [{"json": {"value": "2026-04-23T16:04:28+0000"}}]
+    code = textwrap.dedent("""
+        from datetime import datetime
+        parsed = datetime.strptime(_item['json']['value'], "%Y-%m-%dT%H:%M:%S%z")
+        return {'parsed': parsed.isoformat()}
+    """)
+    task_settings = create_task_settings(code=code, node_mode="per_item", items=items)
+    await broker.send_task(task_id=task_id, task_settings=task_settings)
+
+    done_msg = await wait_for_task_done(broker, task_id)
+
+    assert done_msg["data"]["result"] == [
+        {"json": {"parsed": "2026-04-23T16:04:28+00:00"}, "pairedItem": {"item": 0}}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_per_item_datetime_strftime_works(broker, manager_with_stdlib_wildcard):
+    task_id = nanoid()
+    items = [{"json": {}}]
+    code = textwrap.dedent("""
+        from datetime import datetime
+        return {'formatted': datetime(2026, 4, 23).strftime('%Y-%m-%d')}
+    """)
+    task_settings = create_task_settings(code=code, node_mode="per_item", items=items)
+    await broker.send_task(task_id=task_id, task_settings=task_settings)
+
+    done_msg = await wait_for_task_done(broker, task_id)
+
+    assert done_msg["data"]["result"] == [
+        {"json": {"formatted": "2026-04-23"}, "pairedItem": {"item": 0}}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_per_item_time_strptime_works(broker, manager_with_stdlib_wildcard):
+    task_id = nanoid()
+    items = [{"json": {"value": "2026-04-23"}}]
+    code = textwrap.dedent("""
+        import time
+        parsed = time.strptime(_item['json']['value'], "%Y-%m-%d")
+        return {'year': parsed.tm_year, 'month': parsed.tm_mon, 'day': parsed.tm_mday}
+    """)
+    task_settings = create_task_settings(code=code, node_mode="per_item", items=items)
+    await broker.send_task(task_id=task_id, task_settings=task_settings)
+
+    done_msg = await wait_for_task_done(broker, task_id)
+
+    assert done_msg["data"]["result"] == [
+        {"json": {"year": 2026, "month": 4, "day": 23}, "pairedItem": {"item": 0}}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_per_item_warnings_warn_works(broker, manager_with_stdlib_wildcard):
+    task_id = nanoid()
+    items = [{"json": {}}]
+    code = textwrap.dedent("""
+        import warnings
+        warnings.warn('sample', stacklevel=1)
+        return {'ok': True}
+    """)
+    task_settings = create_task_settings(code=code, node_mode="per_item", items=items)
+    await broker.send_task(task_id=task_id, task_settings=task_settings)
+
+    done_msg = await wait_for_task_done(broker, task_id)
+
+    assert done_msg["data"]["result"] == [
+        {"json": {"ok": True}, "pairedItem": {"item": 0}}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_all_items_datetime_strptime_works(broker, manager_with_stdlib_wildcard):
+    task_id = nanoid()
+    code = textwrap.dedent("""
+        from datetime import datetime
+        parsed = datetime.strptime("2026-04-23T16:04:28+0000", "%Y-%m-%dT%H:%M:%S%z")
+        return [{'json': {'parsed': parsed.isoformat()}}]
+    """)
+    task_settings = create_task_settings(code=code, node_mode="all_items")
+    await broker.send_task(task_id=task_id, task_settings=task_settings)
+
+    done_msg = await wait_for_task_done(broker, task_id)
+
+    assert done_msg["data"]["result"] == [
+        {"json": {"parsed": "2026-04-23T16:04:28+00:00"}}
+    ]
+
+
 # ========== Security ===========
 
 

packages/@n8n/task-runner-python/tests/unit/test_task_executor.py

@@ -1,6 +1,5 @@
 import pytest
 import json
-import types
 from unittest.mock import MagicMock, patch
 
 from src.task_executor import TaskExecutor
@@ -209,7 +208,70 @@ class TestFilterBuiltins:
             runner_env_deny=False,
         )
 
-    def test_returns_mapping_proxy(self):
+    def test_supports_item_access(self):
         result = TaskExecutor._filter_builtins(self._make_security_config())
 
-        assert isinstance(result, types.MappingProxyType)
+        assert result["__import__"] is not None
+        assert result["len"] is len
+        assert "len" in result
+
+    def test_supports_attribute_access(self):
+        result = TaskExecutor._filter_builtins(self._make_security_config())
+
+        assert result.__import__ is not None
+        assert result.len is len
+
+    def test_is_not_a_dict(self):
+        result = TaskExecutor._filter_builtins(self._make_security_config())
+
+        assert not isinstance(result, dict)
+
+    @pytest.mark.parametrize(
+        "name,mutate,expected",
+        [
+            (
+                "item_assignment",
+                lambda r: r.__setitem__("len", None),
+                (TypeError, AttributeError),
+            ),
+            (
+                "attribute_assignment",
+                lambda r: setattr(r, "__import__", None),
+                AttributeError,
+            ),
+            (
+                "dict_class_setitem",
+                lambda r: dict.__setitem__(r, "len", None),
+                TypeError,
+            ),
+            ("dict_class_init", lambda r: dict.__init__(r, {"pwned": True}), TypeError),
+            ("dict_class_update", lambda r: dict.update(r, {"len": None}), TypeError),
+            ("dict_class_clear", lambda r: dict.clear(r), TypeError),
+            ("class_swap", lambda r: setattr(r, "__class__", dict), AttributeError),
+            ("vars_injection", lambda r: vars(r), TypeError),
+            (
+                "object_setattr",
+                lambda r: object.__setattr__(r, "_x", {"pwned": True}),
+                AttributeError,
+            ),
+        ],
+    )
+    def test_rejects_mutation(self, name, mutate, expected):
+        result = TaskExecutor._filter_builtins(self._make_security_config())
+
+        with pytest.raises(expected):
+            mutate(result)
+
+    def test_applies_builtins_deny(self):
+        config = SecurityConfig(
+            stdlib_allow=set(),
+            external_allow=set(),
+            builtins_deny={"open", "eval"},
+            runner_env_deny=False,
+        )
+        result = TaskExecutor._filter_builtins(config)
+
+        assert "open" not in result
+        assert "eval" not in result
+        assert "len" in result
+        assert result["__import__"] is not None