security: fix NTP command handling

Adds basic validation of the NTP server parameter to harden the
usage of the NTP server time setting *AND* also uses shlex.quote
to and removes shell=true from the execution invocation to prevent
any possibliity of injecting a malicious command to be executed.

NOTE(JayF): Rebase to unmaintained/2024.1 no conflicts, but had to add
            imports.

Related-Bug: 2160050
Assisted-By: Claude Opus 4.6
Change-Id: I69e85a1c5040066674d90a6e12acc30cec96ba76
Signed-off-by: Julia Kreger <juliaashleykreger@gmail.com>
Signed-off-by: Jay Faulkner <jay@jvf.cc>

Author: juliakreger

ironic_python_agent/tests/unit/extensions/test_standby.py

@@ -1471,8 +1471,8 @@ def test__sync_clock(self, execute_mock, mock_timemethod):
         self.agent_extension._sync_clock()
 
         calls = [mock.call('chronyc', 'shutdown', check_exit_code=[0, 1]),
-                 mock.call("chronyd -q 'server 192.168.1.1 iburst'",
-                           shell=True),
+                 mock.call('chronyd', '-q',
+                           'server 192.168.1.1 iburst'),
                  mock.call('hwclock', '-v', '--systohc')]
         execute_mock.assert_has_calls(calls)
 

ironic_python_agent/tests/unit/test_utils.py

@@ -846,7 +846,7 @@ def test_sync_clock_chrony(self, mock_time_method, mock_execute):
         utils.sync_clock()
         mock_execute.assert_has_calls([
             mock.call('chronyc', 'shutdown', check_exit_code=[0, 1]),
-            mock.call("chronyd -q 'server 192.168.1.1 iburst'", shell=True),
+            mock.call('chronyd', '-q', 'server 192.168.1.1 iburst'),
         ])
 
     @mock.patch.object(utils, 'determine_time_method', autospec=True)
@@ -877,6 +877,126 @@ def test_sync_clock_ntp_server_is_none(self, mock_time_method,
         utils.sync_clock()
         self.assertEqual(0, mock_execute.call_count)
 
+    def test_sync_clock_invalid_ntp_server_shell_escape(
+            self, mock_execute):
+        self.config(ntp_server="'; rm -rf /; echo '")
+        self.assertRaisesRegex(
+            errors.CommandExecutionError,
+            'Invalid NTP server address',
+            utils.sync_clock)
+        mock_execute.assert_not_called()
+
+    def test_sync_clock_invalid_ntp_server_command_sub(
+            self, mock_execute):
+        self.config(ntp_server='$(reboot)')
+        self.assertRaisesRegex(
+            errors.CommandExecutionError,
+            'Invalid NTP server address',
+            utils.sync_clock)
+        mock_execute.assert_not_called()
+
+    def test_sync_clock_invalid_ntp_server_backtick(
+            self, mock_execute):
+        self.config(ntp_server='`reboot`')
+        self.assertRaisesRegex(
+            errors.CommandExecutionError,
+            'Invalid NTP server address',
+            utils.sync_clock)
+        mock_execute.assert_not_called()
+
+    def test_sync_clock_invalid_ntp_server_pipe(
+            self, mock_execute):
+        self.config(ntp_server='foo | bar')
+        self.assertRaisesRegex(
+            errors.CommandExecutionError,
+            'Invalid NTP server address',
+            utils.sync_clock)
+        mock_execute.assert_not_called()
+
+    def test_sync_clock_invalid_ntp_server_chain(
+            self, mock_execute):
+        self.config(ntp_server='foo && bar')
+        self.assertRaisesRegex(
+            errors.CommandExecutionError,
+            'Invalid NTP server address',
+            utils.sync_clock)
+        mock_execute.assert_not_called()
+
+    def test_sync_clock_invalid_ntp_server_space(
+            self, mock_execute):
+        self.config(ntp_server='foo bar')
+        self.assertRaisesRegex(
+            errors.CommandExecutionError,
+            'Invalid NTP server address',
+            utils.sync_clock)
+        mock_execute.assert_not_called()
+
+    @mock.patch.object(utils, 'determine_time_method', autospec=True)
+    def test_sync_clock_valid_ipv6(self, mock_time_method,
+                                   mock_execute):
+        self.config(ntp_server='2001:db8::1')
+        mock_time_method.return_value = 'ntpdate'
+        utils.sync_clock()
+        mock_execute.assert_has_calls(
+            [mock.call('ntpdate', '2001:db8::1')])
+
+    @mock.patch.object(utils, 'determine_time_method', autospec=True)
+    def test_sync_clock_valid_hostname(self, mock_time_method,
+                                       mock_execute):
+        self.config(ntp_server='ntp.example.com')
+        mock_time_method.return_value = 'ntpdate'
+        utils.sync_clock()
+        mock_execute.assert_has_calls(
+            [mock.call('ntpdate', 'ntp.example.com')])
+
+    @mock.patch.object(utils, 'determine_time_method', autospec=True)
+    def test_sync_clock_valid_hostname_with_hyphens(
+            self, mock_time_method, mock_execute):
+        self.config(ntp_server='my-ntp_server.example.com')
+        mock_time_method.return_value = 'ntpdate'
+        utils.sync_clock()
+        mock_execute.assert_has_calls(
+            [mock.call('ntpdate',
+                       'my-ntp_server.example.com')])
+
+    def test_validate_ntp_server_valid_ipv4(self, mock_execute):
+        utils._validate_ntp_server('192.168.1.1')
+
+    def test_validate_ntp_server_valid_ipv6(self, mock_execute):
+        utils._validate_ntp_server('2001:db8::1')
+
+    def test_validate_ntp_server_valid_hostname(self,
+                                                mock_execute):
+        utils._validate_ntp_server('ntp.example.com')
+
+    def test_validate_ntp_server_rejects_semicolon(
+            self, mock_execute):
+        self.assertRaises(
+            errors.CommandExecutionError,
+            utils._validate_ntp_server,
+            "'; rm -rf /; echo '")
+
+    def test_validate_ntp_server_rejects_dollar(self,
+                                                mock_execute):
+        self.assertRaises(
+            errors.CommandExecutionError,
+            utils._validate_ntp_server,
+            '$(reboot)')
+
+    def test_validate_ntp_server_rejects_backtick(
+            self, mock_execute):
+        self.assertRaises(
+            errors.CommandExecutionError,
+            utils._validate_ntp_server,
+            '`reboot`')
+
+    def test_validate_ntp_server_rejects_space(self,
+                                               mock_execute):
+        self.assertRaises(
+            errors.CommandExecutionError,
+            utils._validate_ntp_server,
+            'foo bar')
+
 
 @mock.patch.object(utils, '_unmount_any_config_drives', autospec=True)
 @mock.patch.object(utils, '_booted_from_vmedia', autospec=True)

ironic_python_agent/utils.py

@@ -19,9 +19,11 @@
 import errno
 import glob
 import io
+import ipaddress
 import json
 import os
 import re
+import shlex
 import shutil
 import subprocess
 import sys
@@ -760,6 +762,7 @@ def get_partition_table_type_from_specs(node):
 
 
 _LARGE_KEYS = frozenset(['configdrive', 'system_logs'])
+_VALID_NTP_SERVER_RE = re.compile(r'^[a-zA-Z0-9.:_-]+$')
 
 
 def remove_large_keys(var):
@@ -794,6 +797,27 @@ def determine_time_method():
     return None
 
 
+def _validate_ntp_server(ntp_server):
+    """Validate an NTP server address for safety.
+
+    :param ntp_server: The NTP server address string.
+    :raises: CommandExecutionError if the value is not valid.
+    """
+    try:
+        ipaddress.ip_address(ntp_server)
+        return
+    except ValueError:
+        pass
+
+    if (ntp_server
+            and len(ntp_server) <= 253
+            and _VALID_NTP_SERVER_RE.match(ntp_server)):
+        return
+
+    raise errors.CommandExecutionError(
+        'Invalid NTP server address: %s' % ntp_server)
+
+
 def sync_clock(ignore_errors=False):
     """Syncs the software clock of the system.
 
@@ -817,6 +841,8 @@ def sync_clock(ignore_errors=False):
     if not CONF.ntp_server:
         return
 
+    _validate_ntp_server(CONF.ntp_server)
+
     method = determine_time_method()
 
     if method == 'ntpdate':
@@ -834,8 +860,9 @@ def sync_clock(ignore_errors=False):
             # stop chronyd, ignore if it ran before or not
             execute('chronyc', 'shutdown', check_exit_code=[0, 1])
             # force a time sync now
-            query = "server " + CONF.ntp_server + " iburst"
-            execute("chronyd -q \'%s\'" % query, shell=True)
+            query = ("server %s iburst"
+                     % shlex.quote(CONF.ntp_server))
+            execute('chronyd', '-q', query)
             LOG.debug('Set software clock using chrony')
         except (processutils.ProcessExecutionError,
                 errors.CommandExecutionError) as e:

releasenotes/notes/fix-ntp-server-shell-injection-5ccf2972e0ea632e.yaml

@@ -0,0 +1,12 @@
+---
+security:
+  - |
+    Fixes a shell command injection vulnerability in the NTP clock
+    synchronization when using chrony. The ``ntp_server`` configuration
+    value was interpolated into a shell command string, allowing
+    a crafted value to execute arbitrary system commands. The chrony
+    execution path no longer uses a shell, and the ``ntp_server``
+    value is now validated and sanitized before use.
+    See `bug 2160050
+    <https://bugs.launchpad.net/ironic-python-agent/+bug/2160050>`_
+    for details.