fix(trl): disable inline GRPO reward code by default (RCE, #11015) (#11068)

fix(trl): disable inline GRPO reward code by default (RCE)

POST /api/fine-tuning/jobs accepts reward_functions[].code, an inline Python
body, and compile_inline_reward() execs it against a restricted-builtins
allowlist (_SAFE_BUILTINS). That allowlist is not a security boundary:
().__class__.__bases__[0].__subclasses__() reaches os._wrap_close and thus
os.system, giving arbitrary code execution. The fine-tuning endpoint is
unauthenticated by default, so any caller could run code on the host.

Hardening the allowlist is a losing game against CPython introspection, so
inline reward code is now refused unless the operator explicitly opts in with
LOCALAI_TRL_ALLOW_INLINE_REWARD=true on the backend. Builtin reward functions
are unaffected. The gate lives in build_reward_functions(), the single point
all inline specs flow through. Docs updated to stop describing the allowlist as
a sandbox and to document the opt-in.

Fixes #11015

Signed-off-by: Isabel Wu <231155141+wuisabel-gif@users.noreply.github.com>
Co-authored-by: Isabel Wu <231155141+wuisabel-gif@users.noreply.github.com>

backend/python/trl/reward_functions.py

@@ -5,12 +5,26 @@ All reward functions follow TRL's signature: (completions, **kwargs) -> list[flo
 """
 
 import json
+import os
 import re
 import math
 import string
 import functools
 
 
+# Opt-in flag for inline reward code. Compiling a caller-supplied Python body
+# is arbitrary code execution: the _SAFE_BUILTINS allowlist below is NOT a
+# security boundary — expressions like ().__class__.__bases__[0].__subclasses__()
+# escape it trivially to reach os.system. Since the fine-tuning endpoint is
+# unauthenticated by default, inline reward functions are disabled unless the
+# operator explicitly opts in by setting this env var to a truthy value.
+ALLOW_INLINE_ENV = "LOCALAI_TRL_ALLOW_INLINE_REWARD"
+
+
+def _inline_rewards_allowed():
+    return os.environ.get(ALLOW_INLINE_ENV, "").strip().lower() in ("1", "true", "yes", "on")
+
+
 # ---------------------------------------------------------------------------
 # Built-in reward functions
 # ---------------------------------------------------------------------------
@@ -224,6 +238,14 @@ def build_reward_functions(specs_json):
             reward_funcs.append(func)
 
         elif spec_type == "inline":
+            if not _inline_rewards_allowed():
+                raise ValueError(
+                    f"Inline reward function '{name}' rejected: inline reward code "
+                    f"executes arbitrary Python and is disabled by default. Set "
+                    f"{ALLOW_INLINE_ENV}=true on the backend to enable it (only on a "
+                    f"trusted, access-controlled instance), or use a builtin reward "
+                    f"function instead."
+                )
             code = spec.get("code", "")
             if not code.strip():
                 raise ValueError(f"Inline reward function '{name}' has no code")

backend/python/trl/test_reward_functions.py

@@ -0,0 +1,48 @@
+"""
+Tests for reward_functions.py — no gRPC / model deps, pure stdlib.
+
+Covers the inline-code opt-in gate (issue #11015): inline reward code is
+arbitrary code execution and must stay disabled unless the operator opts in.
+"""
+import os
+import unittest
+
+import reward_functions as rf
+
+INLINE_SPEC = [{"type": "inline", "name": "ok", "code": "return [1.0 for _ in completions]"}]
+
+
+class TestInlineRewardGate(unittest.TestCase):
+    def setUp(self):
+        self._saved = os.environ.pop(rf.ALLOW_INLINE_ENV, None)
+
+    def tearDown(self):
+        if self._saved is None:
+            os.environ.pop(rf.ALLOW_INLINE_ENV, None)
+        else:
+            os.environ[rf.ALLOW_INLINE_ENV] = self._saved
+
+    def test_inline_refused_by_default(self):
+        with self.assertRaises(ValueError) as cm:
+            rf.build_reward_functions(INLINE_SPEC)
+        self.assertIn("disabled by default", str(cm.exception))
+
+    def test_builtin_works_without_optin(self):
+        fns = rf.build_reward_functions([{"type": "builtin", "name": "format_reward"}])
+        self.assertEqual(len(fns), 1)
+        self.assertTrue(callable(fns[0]))
+
+    def test_inline_enabled_with_optin(self):
+        os.environ[rf.ALLOW_INLINE_ENV] = "true"
+        fns = rf.build_reward_functions(INLINE_SPEC)
+        self.assertEqual(fns[0](["x"]), [1.0])
+
+    def test_optin_falsey_values_stay_disabled(self):
+        for val in ("", "0", "false", "no", "off"):
+            os.environ[rf.ALLOW_INLINE_ENV] = val
+            with self.assertRaises(ValueError):
+                rf.build_reward_functions(INLINE_SPEC)
+
+
+if __name__ == "__main__":
+    unittest.main()