fix(security): reject imported and setattr Gunicorn workers with memory://

config.py

@@ -383,6 +383,29 @@ def _call_sets_workers_via_setitem(call):
     return isinstance(key, ast.Constant) and key.value == "workers"
 
 
+def _call_sets_workers_attribute(call):
+    """Return True for setattr/object.__setattr__ calls that bind ``workers``."""
+    if not isinstance(call, ast.Call) or len(call.args) < 2:
+        return False
+    key = call.args[1]
+    if not isinstance(key, ast.Constant) or key.value != "workers":
+        return False
+    func = call.func
+    if isinstance(func, ast.Name) and func.id == "setattr":
+        return True
+    return isinstance(func, ast.Attribute) and func.attr == "__setattr__"
+
+
+def _import_from_binds_workers(node):
+    """Return True when an import statement may define ``workers`` indirectly."""
+    if not isinstance(node, ast.ImportFrom):
+        return False
+    return any(
+        alias.name == "*" or (alias.asname or alias.name) == "workers"
+        for alias in node.names
+    )
+
+
 def _expression_mutates_workers(expr):
     """Return True when an expression statement mutates ``workers`` indirectly."""
     if isinstance(expr, ast.Call):
@@ -431,13 +454,7 @@ def _is_dynamic_workers_mutation(node):
             call = node.value
             if isinstance(call.func, ast.Name) and call.func.id in {"exec", "eval"}:
                 return True
-            if (
-                isinstance(call.func, ast.Name)
-                and call.func.id == "setattr"
-                and len(call.args) >= 2
-                and isinstance(call.args[1], ast.Constant)
-                and call.args[1].value == "workers"
-            ):
+            if _call_sets_workers_attribute(call):
                 return True
 
     if isinstance(node, ast.Assign):
@@ -548,6 +565,9 @@ def _walk_gunicorn_workers_statements(
         if _is_dynamic_workers_mutation(node):
             state.dynamic = True
 
+        if _import_from_binds_workers(node):
+            state.dynamic = True
+
         if isinstance(node, ast.Expr) and isinstance(node.value, ast.NamedExpr):
             walrus = node.value
             if _target_assigns_workers(walrus.target):

tests/test_app.py

@@ -462,6 +462,29 @@ def test_config_rejects_memory_ratelimit_with_gunicorn_configure_hook(monkeypatc
     )
 
 
+def test_config_rejects_memory_ratelimit_with_imported_gunicorn_workers(monkeypatch):
+    settings_file = _PROJECT_ROOT / "tests" / "_gunicorn_worker_settings.py"
+    settings_file.write_text("workers = 4\n", encoding="utf-8")
+    try:
+        _assert_config_import_with_gunicorn_file(
+            monkeypatch,
+            config_content="from _gunicorn_worker_settings import workers\n",
+        )
+    finally:
+        settings_file.unlink(missing_ok=True)
+
+
+def test_config_rejects_memory_ratelimit_with_object_setattr_gunicorn_workers(monkeypatch):
+    _assert_config_import_with_gunicorn_file(
+        monkeypatch,
+        config_content=(
+            "workers = 1\n"
+            "import sys\n"
+            'object.__setattr__(sys.modules[__name__], "workers", 8)\n'
+        ),
+    )
+
+
 def test_config_accepts_dynamic_gunicorn_config_with_shared_ratelimit_backend(monkeypatch):
     _assert_config_import_with_gunicorn_file(
         monkeypatch,

tests/test_codex_security_review.py

@@ -112,10 +112,28 @@ def test_gunicorn_config_on_starting_hook_is_dynamic(tmp_path):
         "workers = 1\nmatch 1:\n    case 1:\n        workers = 4\n",
         "m = __import__('sys').modules[__name__].__dict__\nm['workers'] = 4\n",
         "workers = 1\n[globals().__setitem__('workers', 4)]\n",
+        "from worker_settings import workers\n",
+        "from worker_settings import worker_count as workers\n",
+        "from settings import *\n",
+        "workers = 1\nimport sys\nobject.__setattr__(sys.modules[__name__], 'workers', 8)\n",
     ],
 )
 def test_gunicorn_config_alternate_workers_assignments_are_dynamic(tmp_path, config_content):
     config_file = tmp_path / "gunicorn.conf.py"
+    (tmp_path / "worker_settings.py").write_text(
+        "workers = 4\nworker_count = 4\n", encoding="utf-8"
+    )
+    (tmp_path / "settings.py").write_text("workers = 4\n", encoding="utf-8")
     config_file.write_text(config_content, encoding="utf-8")
 
     assert config._workers_from_gunicorn_config_path(str(config_file)) == (1, True)
+
+
+def test_gunicorn_config_import_aliased_away_from_workers_is_not_dynamic(tmp_path):
+    config_file = tmp_path / "gunicorn.conf.py"
+    config_file.write_text(
+        "from worker_settings import workers as default_workers\n",
+        encoding="utf-8",
+    )
+
+    assert config._workers_from_gunicorn_config_path(str(config_file)) == (1, False)