python-fire: (1) PAGER+shell=True RCE in console_io.More (2) bash $(...) runs when sourcing --completion from maliciously named .py file. PoC verified.

fire/completion.py

@@ -21,11 +21,20 @@
 import collections
 import copy
 import inspect
+import re
 
 from fire import inspectutils
 
+# Characters that are safe to embed unquoted in the generated Bash/Fish
+# completion scripts. Anything else (e.g. '$', '`', '(', ')', ';', '|',
+# whitespace) is stripped so that a maliciously named component (for
+# example a `.py` file whose basename contains shell metacharacters)
+# cannot inject commands that run when the script is sourced.
+_UNSAFE_NAME_CHARS_RE = re.compile(r'[^\w./,@:%+=-]')
+
 
 def Script(name, component, default_options=None, shell='bash'):
+  name = _UNSAFE_NAME_CHARS_RE.sub('', name)
   if shell == 'fish':
     return _FishScript(name, _Commands(component), default_options)
   return _BashScript(name, _Commands(component), default_options)

fire/completion_test.py

@@ -36,6 +36,15 @@ def testCompletionBashScript(self):
     for last_command in ['command', 'halt']:
       self.assertIn(f'{last_command})', script)
 
+  def testCompletionBashScriptSanitizesUnsafeName(self):
+    # Regression test: a maliciously crafted name (e.g. derived from a
+    # `.py` filename containing shell metacharacters) must not be able to
+    # inject commands into the generated script.
+    script = completion.Script('$(touch /tmp/pwned)', {'run': 0})
+    self.assertNotIn('$(touch', script)
+    self.assertNotIn('`touch', script)
+    self.assertIn('touch/tmp/pwned', script)
+
   def testCompletionFishScript(self):
     # A sanity check test to make sure the fish completion script satisfies
     # some basic assumptions.

fire/console/console_io.py

@@ -16,6 +16,7 @@
 """General console printing utilities used by the Cloud SDK."""
 
 import os
+import shlex
 import signal
 import subprocess
 import sys
@@ -97,7 +98,8 @@ def More(contents, out, prompt=None, check_pager=True):
       # Ignore SIGINT while the pager is running.
       # We don't want to terminate the parent while the child is still alive.
       signal.signal(signal.SIGINT, signal.SIG_IGN)
-      p = subprocess.Popen(pager, stdin=subprocess.PIPE, shell=True)
+      p = subprocess.Popen(
+          shlex.split(pager), stdin=subprocess.PIPE, shell=False)
       enc = console_attr.GetConsoleAttr().GetEncoding()
       p.communicate(input=contents.encode(enc))
       p.wait()

fire/console/console_io_test.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*- #
+# Copyright 2024 Google LLC. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests for fire.console.console_io."""
+
+import io
+from unittest import mock
+
+from fire import testutils
+from fire.console import console_io
+
+
+class MoreTest(testutils.BaseTestCase):
+
+  @mock.patch.object(console_io, 'IsInteractive', return_value=True)
+  @mock.patch.object(console_io.signal, 'signal')
+  @mock.patch.object(console_io.encoding, 'GetEncodedValue')
+  @mock.patch.object(console_io.subprocess, 'Popen')
+  def testMoreDoesNotUseShellForPager(
+      self, mock_popen, mock_get_encoded_value, unused_mock_signal,
+      unused_mock_is_interactive):
+    # Regression test: PAGER must never be passed through a shell, since a
+    # malicious or attacker-controlled PAGER value could otherwise inject
+    # arbitrary commands (CWE-78).
+    mock_get_encoded_value.side_effect = (
+        lambda env, name, default=None: {
+            'PAGER': 'less -R; touch /tmp/pwned',
+        }.get(name, default)
+    )
+    mock_process = mock.Mock()
+    mock_process.communicate.return_value = (b'', b'')
+    mock_popen.return_value = mock_process
+
+    console_io.More('some contents', io.StringIO())
+
+    pager_calls = [
+        call for call in mock_popen.call_args_list
+        if call.args and call.args[0][:1] == ['less']
+    ]
+    self.assertEqual(len(pager_calls), 1)
+    args, kwargs = pager_calls[0]
+    self.assertFalse(kwargs.get('shell'))
+    self.assertEqual(args[0], ['less', '-R;', 'touch', '/tmp/pwned'])
+
+
+if __name__ == '__main__':
+  testutils.main()