From f0a759f26c9ef8baabff2f4e7a51aa31e766db40 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 17 Sep 2026 02:06:46 +0000 Subject: [PATCH] fix(security): close six Gunicorn worker AST scan bypasses Detect walrus-if and match-guard namespace updates, __init_subclass__ subclass definitions, dataclass __post_init__ instantiation, metaclass __new__ hooks, and descriptor __get__ attribute access that mutate workers at import time without being flagged as dynamic. Startup with RATELIMIT_STORAGE_URI=memory:// would otherwise allow multi-worker Gunicorn to bypass per-worker login rate limits and session stores while the panel believes it runs single-worker. Co-authored-by: Alexander Wagner --- config.py | 90 +++++++++++++++++++++++++---- tests/test_security_review_sep17.py | 45 +++++++++++++++ 2 files changed, 123 insertions(+), 12 deletions(-) create mode 100644 tests/test_security_review_sep17.py diff --git a/config.py b/config.py index 1879bb3..31a8340 100644 --- a/config.py +++ b/config.py @@ -3535,7 +3535,7 @@ def _function_is_property_method(func_node, operator_bindings=None): def _metaclass_init_mutates_workers(class_node, operator_bindings): - """Return True when a ``type`` subclass ``__init__`` mutates ``workers``.""" + """Return True when a ``type`` subclass hook mutates ``workers``.""" if not any( isinstance(base, ast.Name) and base.id == "type" for base in class_node.bases @@ -3544,7 +3544,7 @@ def _metaclass_init_mutates_workers(class_node, operator_bindings): for stmt in class_node.body: if ( isinstance(stmt, ast.FunctionDef) - and stmt.name == "__init__" + and stmt.name in ("__init__", "__new__") and _function_mutates_workers(stmt, operator_bindings) ): return True @@ -3587,16 +3587,21 @@ def _record_class_side_effect_target( return False if not _function_mutates_workers(stmt, operator_bindings): return False - if stmt.name == '__init__': + if stmt.name in ("__init__", "__post_init__"): constructors.add(class_name) return True target = (class_name, stmt.name) - if _function_is_static_or_class_method(stmt, operator_bindings): + if stmt.name == "__init_subclass__" or _function_is_static_or_class_method( + stmt, operator_bindings + ): methods.add(target) return True if _function_is_property_method(stmt, operator_bindings): properties.add(target) return True + if stmt.name == "__get__": + methods.add(target) + return True return False @@ -3624,6 +3629,25 @@ def _class_has_worker_side_effect_target( return risky +def _descriptor_class_names(methods): + """Return classes whose ``__get__`` hook may mutate ``workers``.""" + return {class_name for class_name, method_name in methods if method_name == "__get__"} + + +def _record_descriptor_field_assignments(class_node, descriptor_classes, descriptor_fields): + """Record class attributes instantiated from descriptor classes.""" + for stmt in class_node.body: + if not isinstance(stmt, ast.Assign): + continue + if not isinstance(stmt.value, ast.Call) or not isinstance(stmt.value.func, ast.Name): + continue + if stmt.value.func.id not in descriptor_classes: + continue + for target in stmt.targets: + if isinstance(target, ast.Name): + descriptor_fields.add((class_node.name, target.id)) + + def _record_class_side_effect_binding( class_node, binding_events, @@ -3691,6 +3715,7 @@ def _collect_class_side_effect_targets(tree, operator_bindings): constructors = set() methods = set() properties = set() + descriptor_fields = set() metaclass_definitions = _collect_metaclass_definition_mutators( tree, operator_bindings, @@ -3704,7 +3729,22 @@ def _collect_class_side_effect_targets(tree, operator_bindings): properties, binding_events, ) - return constructors, methods, properties, metaclass_definitions, binding_events + descriptor_classes = _descriptor_class_names(methods) + for node in tree.body: + if isinstance(node, ast.ClassDef): + _record_descriptor_field_assignments( + node, + descriptor_classes, + descriptor_fields, + ) + return ( + constructors, + methods, + properties, + metaclass_definitions, + binding_events, + descriptor_fields, + ) @@ -3722,6 +3762,7 @@ def _class_binding_is_active(class_targets, class_name, reference_line): def _expression_triggers_class_workers_side_effect(expr, class_targets): """Return True when attribute access or construction runs a mutating class hook.""" constructors, methods, properties = class_targets[:3] + descriptor_fields = class_targets[5] if len(class_targets) > 5 else set() reference_line = getattr(expr, 'lineno', 0) if isinstance(expr, ast.Call) and isinstance(expr.func, ast.Name): return expr.func.id in constructors and _class_binding_is_active( @@ -3745,9 +3786,14 @@ def _expression_triggers_class_workers_side_effect(expr, class_targets): and isinstance(expr.value.func, ast.Name) ): class_name = expr.value.func.id - return ( + if ( (class_name, expr.attr) in properties and _class_binding_is_active(class_targets, class_name, reference_line) + ): + return True + return ( + (class_name, expr.attr) in descriptor_fields + and _class_binding_is_active(class_targets, class_name, reference_line) ) return False @@ -3755,8 +3801,14 @@ def _expression_triggers_class_workers_side_effect(expr, class_targets): def _classdef_has_import_time_workers_side_effect(class_node, class_targets): """Return True when defining the class itself mutates ``workers``.""" + methods = class_targets[1] metaclass_definitions = class_targets[3] - return class_node.name in metaclass_definitions + if class_node.name in metaclass_definitions: + return True + return any( + isinstance(base, ast.Name) and (base.id, "__init_subclass__") in methods + for base in class_node.bases + ) @@ -4360,10 +4412,13 @@ def _is_dynamic_workers_mutation(node, operator_bindings, dict_subclass_names=No return True if any( isinstance(child, ast.expr) - and _expression_mutates_workers( - child, - operator_bindings, - dict_subclass_names, + and ( + _expression_mutates_workers( + child, + operator_bindings, + dict_subclass_names, + ) + or _expression_has_risky_instance_update(child) ) for child in ast.iter_child_nodes(node) ): @@ -4375,6 +4430,17 @@ def _is_dynamic_workers_mutation(node, operator_bindings, dict_subclass_names=No ) if isinstance(node, (ast.AnnAssign, ast.AugAssign)): return _indirect_workers_assignment_target(node.target) + if isinstance(node, ast.Match): + for case in node.cases: + guard = case.guard + if guard is None: + continue + if _expression_mutates_workers( + guard, + operator_bindings, + dict_subclass_names, + ) or _expression_has_risky_instance_update(guard): + return True return False @@ -4944,7 +5010,7 @@ def _worker_scan_defaults( return ( set() if global_workers_mutators is None else global_workers_mutators, (set(), set()) if operator_bindings is None else operator_bindings, - (set(), set(), set(), set(), {}) if class_targets is None else class_targets, + (set(), set(), set(), set(), {}, set()) if class_targets is None else class_targets, {} if dict_subclass_names is None else dict_subclass_names, ) diff --git a/tests/test_security_review_sep17.py b/tests/test_security_review_sep17.py new file mode 100644 index 0000000..bd9e2c0 --- /dev/null +++ b/tests/test_security_review_sep17.py @@ -0,0 +1,45 @@ +import pytest + +import config + + +@pytest.mark.parametrize( + "config_content", + [ + "workers = 1\nif (ns := globals()):\n ns.update({'workers': 4})\n", + "workers = 1\nmatch globals():\n case ns if ns.update({'workers': 4}) is None: pass\n", + "workers = 1\nclass X:\n def __init_subclass__(cls):\n globals().update({'workers': 4})\nclass Y(X): pass\n", + ( + "workers = 1\n" + "from dataclasses import dataclass\n" + "@dataclass\n" + "class D:\n" + " x: int = 1\n" + " def __post_init__(self):\n" + " globals().update({'workers': 4})\n" + "D()\n" + ), + ( + "workers = 1\n" + "class Meta(type):\n" + " def __new__(mcls, name, bases, ns):\n" + " globals().update({'workers': 4})\n" + " return super().__new__(mcls, name, bases, ns)\n" + "class X(metaclass=Meta): pass\n" + ), + ( + "workers = 1\n" + "class D:\n" + " def __get__(self, obj, owner=None):\n" + " globals().update({'workers': 4})\n" + "class X:\n" + " d = D()\n" + "X().d\n" + ), + ], +) +def test_security_review_worker_scan_gaps_are_dynamic(tmp_path, config_content): + config_file = tmp_path / "gunicorn.conf.py" + config_file.write_text(config_content, encoding="utf-8") + + assert config._workers_from_gunicorn_config_path(str(config_file)) == (1, True)