security: avoid shell fallback in SLURM requeue

src/lightning/pytorch/trainer/connectors/signal_connector.py

@@ -86,18 +86,13 @@ def _slurm_sigusr_handler_fn(self, signum: _SIGNUM, _: FrameType) -> None:
             else:
                 job_id = os.environ["SLURM_JOB_ID"]
 
-            assert re.match("[0-9_-]+", job_id)
+            if re.fullmatch(r"[0-9_-]+", job_id) is None:
+                raise RuntimeError(f"Invalid SLURM job id: {job_id!r}")
             cmd = ["scontrol", "requeue", job_id]
 
             # requeue job
             log.info(f"requeing job {job_id}...")
-            try:
-                result = call(cmd)
-            except FileNotFoundError:
-                # This can occur if a subprocess call to `scontrol` is run outside a shell context
-                # Re-attempt call (now with shell context). If any error is raised, propagate to user.
-                # When running a shell command, it should be passed as a single string.
-                result = call(" ".join(cmd), shell=True)
+            result = call(cmd)
 
             # print result text
             if result == 0:

tests/tests_pytorch/trainer/connectors/test_signal_connector.py

@@ -135,8 +135,23 @@ def test_auto_requeue_invalid_job_id(call_mock):
     call_mock.return_value = 0
     trainer = Trainer(plugins=[SLURMEnvironment()])
     connector = _SignalConnector(trainer)
-    with pytest.raises(AssertionError):
+    with pytest.raises(RuntimeError, match="Invalid SLURM job id"):
         connector._slurm_sigusr_handler_fn(None, None)
+    call_mock.assert_not_called()
+
+
+@RunIf(skip_windows=True)
+@mock.patch("lightning.pytorch.trainer.connectors.signal_connector.call")
+@mock.patch("lightning.pytorch.trainer.Trainer.save_checkpoint", mock.MagicMock())
+@mock.patch.dict(os.environ, {"SLURM_JOB_ID": "123; rm -rf /"})
+def test_auto_requeue_job_id_command_injection(call_mock):
+    """Test that a crafted SLURM_JOB_ID containing shell metacharacters is rejected."""
+    call_mock.return_value = 0
+    trainer = Trainer(plugins=[SLURMEnvironment()])
+    connector = _SignalConnector(trainer)
+    with pytest.raises(RuntimeError, match="Invalid SLURM job id"):
+        connector._slurm_sigusr_handler_fn(None, None)
+    call_mock.assert_not_called()
 
 
 def _registering_signals():