diff --git a/config.py b/config.py index 8f2e1a6..9ed50a8 100644 --- a/config.py +++ b/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): diff --git a/tests/test_app.py b/tests/test_app.py index 3f075c0..c1beb2f 100644 --- a/tests/test_app.py +++ b/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, diff --git a/tests/test_codex_security_review.py b/tests/test_codex_security_review.py index 3e7d99a..f3674dc 100644 --- a/tests/test_codex_security_review.py +++ b/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)