From 9a56529d5d9b6091ef2a599d190827d78eaa1407 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Iv=C3=A1n=20Ovejero?= <ivov.src@gmail.com>
Date: Mon, 3 Nov 2025 13:16:47 +0100
Subject: [PATCH] fix(core): Protect against name-mangling in native Python
 runner (#21482)

---
 packages/@n8n/task-runner-python/src/constants.py  |  3 ++-
 .../@n8n/task-runner-python/src/task_analyzer.py   |  6 ++++++
 .../tests/unit/test_task_analyzer.py               | 14 ++++++++++++++
 3 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/packages/@n8n/task-runner-python/src/constants.py b/packages/@n8n/task-runner-python/src/constants.py
index ea301c07d0c..e339aa51fab 100644
--- a/packages/@n8n/task-runner-python/src/constants.py
+++ b/packages/@n8n/task-runner-python/src/constants.py
@@ -122,7 +122,7 @@ TASK_REJECTED_REASON_OFFER_EXPIRED = (
 TASK_REJECTED_REASON_AT_CAPACITY = "No open task slots - runner already at capacity"
 
 # Security
-BUILTINS_DENY_DEFAULT = "eval,exec,compile,open,input,breakpoint,getattr,object,type,vars,setattr,delattr,hasattr,dir,memoryview,__build_class__,globals,locals"
+BUILTINS_DENY_DEFAULT = "eval,exec,compile,open,input,breakpoint,getattr,object,type,vars,setattr,delattr,hasattr,dir,memoryview,__build_class__,globals,locals,license,help,credits,copyright"
 BLOCKED_NAMES = {
     "__loader__",
     "__builtins__",
@@ -183,6 +183,7 @@ ERROR_STDLIB_DISALLOWED = "Import of standard library module '{module}' is disal
 ERROR_EXTERNAL_DISALLOWED = "Import of external package '{module}' is disallowed. Allowed external packages: {allowed}"
 ERROR_DANGEROUS_NAME = "Access to name '{name}' is disallowed, because it can be used to bypass security restrictions."
 ERROR_DANGEROUS_ATTRIBUTE = "Access to attribute '{attr}' is disallowed, because it can be used to bypass security restrictions."
+ERROR_NAME_MANGLED_ATTRIBUTE = "Access to name-mangled attributes (pattern: _ClassName__attr) is disallowed for security reasons."
 ERROR_DYNAMIC_IMPORT = (
     "Dynamic __import__() calls are not allowed for security reasons."
 )
diff --git a/packages/@n8n/task-runner-python/src/task_analyzer.py b/packages/@n8n/task-runner-python/src/task_analyzer.py
index 062f777686d..ea2d8281068 100644
--- a/packages/@n8n/task-runner-python/src/task_analyzer.py
+++ b/packages/@n8n/task-runner-python/src/task_analyzer.py
@@ -10,6 +10,7 @@ from src.constants import (
     ERROR_RELATIVE_IMPORT,
     ERROR_DANGEROUS_NAME,
     ERROR_DANGEROUS_ATTRIBUTE,
+    ERROR_NAME_MANGLED_ATTRIBUTE,
     ERROR_DYNAMIC_IMPORT,
     BLOCKED_ATTRIBUTES,
     BLOCKED_NAMES,
@@ -62,6 +63,11 @@ class SecurityValidator(ast.NodeVisitor):
                 node.lineno, ERROR_DANGEROUS_ATTRIBUTE.format(attr=node.attr)
             )
 
+        if node.attr.startswith("_") and "__" in node.attr:
+            parts = node.attr.split("__", 1)
+            if len(parts) == 2 and parts[0].startswith("_"):
+                self._add_violation(node.lineno, ERROR_NAME_MANGLED_ATTRIBUTE)
+
         self.generic_visit(node)
 
     def visit_Call(self, node: ast.Call) -> None:
diff --git a/packages/@n8n/task-runner-python/tests/unit/test_task_analyzer.py b/packages/@n8n/task-runner-python/tests/unit/test_task_analyzer.py
index 920066109d4..26d7afecfba 100644
--- a/packages/@n8n/task-runner-python/tests/unit/test_task_analyzer.py
+++ b/packages/@n8n/task-runner-python/tests/unit/test_task_analyzer.py
@@ -136,6 +136,20 @@ class TestAttributeAccessValidation(TestTaskAnalyzer):
         for code in allowed_attributes:
             analyzer.validate(code)
 
+    def test_name_mangled_attributes_blocked(self, analyzer: TaskAnalyzer) -> None:
+        exploit_attempts = [
+            "license._Printer__filenames",
+            "obj._SomeClass__private_attr",
+            "help._Helper__name",
+            "credits._Printer__data",
+            "instance._MyClass__secret",
+        ]
+
+        for code in exploit_attempts:
+            with pytest.raises(SecurityViolationError) as exc_info:
+                analyzer.validate(code)
+            assert "name-mangled" in exc_info.value.description.lower()
+
 
 class TestDynamicImportDetection(TestTaskAnalyzer):
     def test_various_dynamic_import_patterns(self, analyzer: TaskAnalyzer) -> None: