fix(companion): Validate stop_signal and harden control dispatch

A typo'd companion_stop_signal (e.g. "SIGTRM") passed validate_string
but raised ValueError in _signal_number when the manager later tried to
send it -- propagating past handle_line and killing the run loop.

Validate stop_signal at config-build time so a bad value fails loudly
on load and reread. As defense-in-depth, catch unexpected exceptions in
ControlServer.handle_line so no handler bug can escape and kill the
manager; they now return an error envelope.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

gunicorn/companion/config.py

@@ -4,6 +4,7 @@
 
 import hashlib
 import json
+import signal
 
 
 # Maps each optional companion field to the global setting build_companion_configs
@@ -93,6 +94,23 @@ class CompanionConfig:
 ALLOWED_SPEC_KEYS = {"name", "target"} | set(FIELD_DEFAULTS)
 
 
+def _validate_stop_signal(stop_signal, name):
+    """Reject a stop_signal that does not name a real signal.
+
+    Caught here at build time so a typo like ``"SIGTRM"`` fails loudly when
+    config is loaded or rereaded, rather than crashing the manager later when
+    it tries to send the signal.
+    """
+    try:
+        if isinstance(stop_signal, str):
+            signal.Signals[stop_signal]
+        else:
+            signal.Signals(stop_signal)
+    except (KeyError, ValueError):
+        raise ValueError(
+            "companion %s has unknown stop_signal %r" % (name, stop_signal))
+
+
 def _load_companion_settings(cfg):
     """Return the ``companion_*`` settings from ``companion_config_file``, or
     ``{}`` when no dedicated file is configured."""
@@ -130,6 +148,7 @@ def build_companion_configs(cfg):
                 % (sorted(unknown), spec))
         fields = {field: spec.get(field, setting(global_setting))
                   for field, global_setting in FIELD_DEFAULTS.items()}
+        _validate_stop_signal(fields["stop_signal"], spec["name"])
         configs.append(CompanionConfig(
             name=spec["name"], target=spec["target"],
             restart_delay=setting("companion_restart_delay"), **fields))

gunicorn/companion/control.py

@@ -87,14 +87,20 @@ class ControlServer:
     def handle_line(self, line):
         """Run one request line and return the encoded response bytes.
 
-        Both decoding and dispatch failures are caught and rendered as an
-        error response, so one bad request never breaks the connection or the
-        manager.
+        Decoding and dispatch failures are caught and rendered as an error
+        response, so one bad request never breaks the connection or the
+        manager. CommandError is the expected rejection; any other exception
+        from dispatch is an unexpected handler bug, caught here for the same
+        reason -- it must not escape and kill the manager's run loop.
         """
         try:
             response = self.dispatch(decode_command(line))
         except CommandError as error:
             response = {"ok": False, "error": str(error)}
+        except Exception as error:
+            if self.log is not None:
+                self.log.exception("companion control command failed")
+            response = {"ok": False, "error": "internal error: %s" % error}
         return encode_response(response)
 
     def serve_connection(self, connection):

tests/test_companion_config.py

@@ -50,6 +50,20 @@ def test_build_rejects_unknown_spec_key():
             make_config([{"name": "rq", "target": "pkg:run", "stop_sigal": "X"}]))
 
 
+def test_build_rejects_unknown_stop_signal():
+    with pytest.raises(ValueError):
+        build_companion_configs(
+            make_config([{"name": "rq", "target": "pkg:run",
+                          "stop_signal": "SIGTRM"}]))
+
+
+def test_build_rejects_unknown_global_stop_signal():
+    with pytest.raises(ValueError):
+        build_companion_configs(
+            make_config([{"name": "rq", "target": "pkg:run"}],
+                        companion_stop_signal="NOPE"))
+
+
 def test_build_reads_companion_config_file(tmp_path):
     config_file = tmp_path / "companion.conf.py"
     config_file.write_text(

tests/test_companion_control.py

@@ -73,6 +73,15 @@ def test_handle_line_dispatch_command_error():
     assert response["ok"] is False and response["error"] == "unknown command"
 
 
+def test_handle_line_unexpected_exception_caught():
+    def dispatch(command):
+        raise ValueError("unknown stop signal 'SIGTRM'")
+    server = ControlServer(dispatch=dispatch, path="/tmp/x.sock",
+                           log=mock.Mock())
+    response = json.loads(server.handle_line('{"cmd": "stop", "name": "rq"}'))
+    assert response["ok"] is False and "internal error" in response["error"]
+
+
 def test_create_unlinks_stale_and_chmods():
     server = ControlServer(dispatch=lambda command: {}, path="/tmp/x.sock",
                            mode=0o600)