Fail closed on arbitrary static extraction calls

This commit is contained in:
2026-07-02 17:47:11 +02:00
parent 79d9921ba6
commit 07822bc3ec
2 changed files with 204 additions and 1 deletions
@@ -1489,6 +1489,81 @@ NODE_CLASS_MAPPINGS = {
self.assertEqual({}, result["nodes"])
self.assertEqual("no_static_nodes", result["pack"]["status"])
def test_return_types_arbitrary_call_skips_node(self):
source = '''
class ArbitraryCallReturnTypesNode:
RETURN_TYPES = ["IMAGE"]
mutate(RETURN_TYPES)
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
},
}
NODE_CLASS_MAPPINGS = {
"ArbitraryCallReturnTypesNode": ArbitraryCallReturnTypesNode,
}
'''
result = self._extract_source(source, "arbitrary-call-return-types-pack")
self.assertEqual({}, result["nodes"])
self.assertEqual("no_static_nodes", result["pack"]["status"])
def test_return_types_alias_arbitrary_call_skips_node(self):
source = '''
class AliasArbitraryCallReturnTypesNode:
RETURN_TYPES = ["IMAGE"]
ALIAS = RETURN_TYPES
mutate(ALIAS)
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
},
}
NODE_CLASS_MAPPINGS = {
"AliasArbitraryCallReturnTypesNode": AliasArbitraryCallReturnTypesNode,
}
'''
result = self._extract_source(source, "alias-arbitrary-call-return-types-pack")
self.assertEqual({}, result["nodes"])
self.assertEqual("no_static_nodes", result["pack"]["status"])
def test_return_types_function_default_arbitrary_call_skips_node(self):
source = '''
class DefaultArbitraryCallReturnTypesNode:
RETURN_TYPES = ["IMAGE"]
def helper(value=mutate(RETURN_TYPES)):
pass
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
},
}
NODE_CLASS_MAPPINGS = {
"DefaultArbitraryCallReturnTypesNode": DefaultArbitraryCallReturnTypesNode,
}
'''
result = self._extract_source(source, "default-arbitrary-call-return-types-pack")
self.assertEqual({}, result["nodes"])
self.assertEqual("no_static_nodes", result["pack"]["status"])
def test_return_names_alias_subscript_assignment_skips_node(self):
source = '''
class AliasSubscriptMutatedReturnNamesNode:
@@ -2727,6 +2802,55 @@ globals().update(NODE_CLASS_MAPPINGS={})
self.assertEqual({}, result["nodes"])
self.assertEqual("no_static_nodes", result["pack"]["status"])
def test_arbitrary_call_invalidates_static_node_mapping(self):
source = '''
class ArbitraryCallMappingNode:
RETURN_TYPES = ("IMAGE",)
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
},
}
NODE_CLASS_MAPPINGS = {
"ArbitraryCallMappingNode": ArbitraryCallMappingNode,
}
mutate(NODE_CLASS_MAPPINGS)
'''
result = self._extract_source(source, "arbitrary-call-mapping-pack")
self.assertEqual({}, result["nodes"])
self.assertEqual("no_static_nodes", result["pack"]["status"])
def test_alias_arbitrary_call_invalidates_static_node_mapping(self):
source = '''
class AliasArbitraryCallMappingNode:
RETURN_TYPES = ("IMAGE",)
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
},
}
NODE_CLASS_MAPPINGS = {
"AliasArbitraryCallMappingNode": AliasArbitraryCallMappingNode,
}
ALIAS = globals()["NODE_CLASS_MAPPINGS"]
mutate(ALIAS)
'''
result = self._extract_source(source, "alias-arbitrary-call-mapping-pack")
self.assertEqual({}, result["nodes"])
self.assertEqual("no_static_nodes", result["pack"]["status"])
def test_globals_subscript_alias_mutation_invalidates_static_node_mapping(self):
source = '''
class GlobalAliasMutatedMappingNode:
+80 -1
View File
@@ -493,8 +493,78 @@ def _mutating_call_target_names(stmt):
return names
def _referenced_names(node):
names = set()
class ReferenceVisitor(ast.NodeVisitor):
def visit_Call(self, child):
name = _namespace_lookup_name(child)
if name is not None:
names.add(name)
self.generic_visit(child)
def visit_Subscript(self, child):
name = _namespace_subscript_name(child)
if name is not None:
names.add(name)
self.generic_visit(child)
def visit_Name(self, child):
names.add(child.id)
ReferenceVisitor().visit(node)
return names
def _arbitrary_call_observed_names(stmt):
names = set()
class ArbitraryCallVisitor(ast.NodeVisitor):
def _visit_function_definition_expressions(self, node):
for decorator in node.decorator_list:
self.visit(decorator)
self.visit(node.args)
if node.returns is not None:
self.visit(node.returns)
for type_param in getattr(node, "type_params", ()):
self.visit(type_param)
def visit_FunctionDef(self, node):
self._visit_function_definition_expressions(node)
def visit_AsyncFunctionDef(self, node):
self._visit_function_definition_expressions(node)
def visit_ClassDef(self, node):
for decorator in node.decorator_list:
self.visit(decorator)
for base in node.bases:
self.visit(base)
for keyword in node.keywords:
self.visit(keyword.value)
for type_param in getattr(node, "type_params", ()):
self.visit(type_param)
for child in node.body:
self.visit(child)
def visit_Lambda(self, node):
self.visit(node.args)
def visit_Call(self, node):
if isinstance(node.func, ast.Attribute):
names.update(_referenced_names(node.func.value))
for arg in node.args:
names.update(_referenced_names(arg))
for keyword in node.keywords:
names.update(_referenced_names(keyword.value))
self.generic_visit(node)
ArbitraryCallVisitor().visit(stmt)
return names
def _assigned_names_in_control_flow(stmt):
names = _mutating_call_target_names(stmt)
names = _mutating_call_target_names(stmt) | _arbitrary_call_observed_names(stmt)
class AssignmentVisitor(ast.NodeVisitor):
def visit_FunctionDef(self, node):
@@ -536,6 +606,7 @@ def _assigned_names_in_control_flow(stmt):
def visit_Expr(self, node):
names.update(_mutating_call_target_names(node))
names.update(_arbitrary_call_observed_names(node))
names.update(_named_expr_target_names(node))
def visit_With(self, node):
@@ -808,10 +879,15 @@ def _class_attr(cls, name, env):
aliases = set()
for stmt in cls.body:
mutating_targets = _mutating_call_target_names(stmt)
observed_targets = _arbitrary_call_observed_names(stmt)
if aliases.intersection(mutating_targets):
value = _INVALID
if name in mutating_targets:
value = _INVALID
if aliases.intersection(observed_targets):
value = _INVALID
if name in observed_targets:
value = _INVALID
if isinstance(stmt, ast.Assign):
target_names = _assignment_target_names(stmt)
if (
@@ -1179,6 +1255,7 @@ def _update_module_dict_alias_from_unpack(target, value, name, aliases):
def _module_dict_alias_invalidated(stmt, aliases):
names = (
_mutating_call_target_names(stmt)
| _arbitrary_call_observed_names(stmt)
| _assignment_target_names(stmt)
| _delete_target_names(stmt)
| _bound_names(stmt)
@@ -1232,6 +1309,8 @@ def _final_module_dict(tree, name, value_converter, value_invalidated_by_names=N
value = _INVALID
if _name_invalidated_by(name, _mutating_call_target_names(stmt)):
value = _INVALID
if _name_invalidated_by(name, _arbitrary_call_observed_names(stmt)):
value = _INVALID
if _module_dict_alias_invalidated(stmt, module_dict_aliases):
value = _INVALID
if isinstance(stmt, ast.Assign):