summaryrefslogtreecommitdiff
path: root/cloudinit
diff options
context:
space:
mode:
Diffstat (limited to 'cloudinit')
-rw-r--r--cloudinit/net/dhcp.py5
-rw-r--r--cloudinit/net/tests/test_dhcp.py18
-rw-r--r--cloudinit/temp_utils.py22
-rw-r--r--cloudinit/tests/test_temp_utils.py101
-rw-r--r--cloudinit/util.py2
5 files changed, 132 insertions, 16 deletions
diff --git a/cloudinit/net/dhcp.py b/cloudinit/net/dhcp.py
index c842c839..05350639 100644
--- a/cloudinit/net/dhcp.py
+++ b/cloudinit/net/dhcp.py
@@ -48,8 +48,9 @@ def maybe_perform_dhcp_discovery(nic=None):
if not dhclient_path:
LOG.debug('Skip dhclient configuration: No dhclient command found.')
return {}
- with temp_utils.tempdir(prefix='cloud-init-dhcp-') as tmpdir:
- return dhcp_discovery(dhclient_path, nic, tmpdir)
+ with temp_utils.tempdir(prefix='cloud-init-dhcp-', needs_exe=True) as tdir:
+ # Use /var/tmp because /run/cloud-init/tmp is mounted noexec
+ return dhcp_discovery(dhclient_path, nic, tdir)
def parse_dhcp_lease_file(lease_file):
diff --git a/cloudinit/net/tests/test_dhcp.py b/cloudinit/net/tests/test_dhcp.py
index 4a37e98a..1324c3d0 100644
--- a/cloudinit/net/tests/test_dhcp.py
+++ b/cloudinit/net/tests/test_dhcp.py
@@ -8,7 +8,7 @@ from cloudinit.net.dhcp import (
InvalidDHCPLeaseFileError, maybe_perform_dhcp_discovery,
parse_dhcp_lease_file, dhcp_discovery)
from cloudinit.util import ensure_file, write_file
-from cloudinit.tests.helpers import CiTestCase
+from cloudinit.tests.helpers import CiTestCase, wrap_and_call
class TestParseDHCPLeasesFile(CiTestCase):
@@ -91,21 +91,27 @@ class TestDHCPDiscoveryClean(CiTestCase):
'Skip dhclient configuration: No dhclient command found.',
self.logs.getvalue())
+ @mock.patch('cloudinit.temp_utils.os.getuid')
@mock.patch('cloudinit.net.dhcp.dhcp_discovery')
@mock.patch('cloudinit.net.dhcp.util.which')
@mock.patch('cloudinit.net.dhcp.find_fallback_nic')
- def test_dhclient_run_with_tmpdir(self, m_fallback, m_which, m_dhcp):
+ def test_dhclient_run_with_tmpdir(self, m_fback, m_which, m_dhcp, m_uid):
"""maybe_perform_dhcp_discovery passes tmpdir to dhcp_discovery."""
- m_fallback.return_value = 'eth9'
+ m_uid.return_value = 0 # Fake root user for tmpdir
+ m_fback.return_value = 'eth9'
m_which.return_value = '/sbin/dhclient'
m_dhcp.return_value = {'address': '192.168.2.2'}
- self.assertEqual(
- {'address': '192.168.2.2'}, maybe_perform_dhcp_discovery())
+ retval = wrap_and_call(
+ 'cloudinit.temp_utils',
+ {'_TMPDIR': {'new': None},
+ 'os.getuid': 0},
+ maybe_perform_dhcp_discovery)
+ self.assertEqual({'address': '192.168.2.2'}, retval)
m_dhcp.assert_called_once()
call = m_dhcp.call_args_list[0]
self.assertEqual('/sbin/dhclient', call[0][0])
self.assertEqual('eth9', call[0][1])
- self.assertIn('/tmp/cloud-init-dhcp-', call[0][2])
+ self.assertIn('/var/tmp/cloud-init/cloud-init-dhcp-', call[0][2])
@mock.patch('cloudinit.net.dhcp.util.subp')
def test_dhcp_discovery_run_in_sandbox(self, m_subp):
diff --git a/cloudinit/temp_utils.py b/cloudinit/temp_utils.py
index 0355f19d..5d7adf70 100644
--- a/cloudinit/temp_utils.py
+++ b/cloudinit/temp_utils.py
@@ -8,9 +8,10 @@ import tempfile
_TMPDIR = None
_ROOT_TMPDIR = "/run/cloud-init/tmp"
+_EXE_ROOT_TMPDIR = "/var/tmp/cloud-init"
-def _tempfile_dir_arg(odir=None):
+def _tempfile_dir_arg(odir=None, needs_exe=False):
"""Return the proper 'dir' argument for tempfile functions.
When root, cloud-init will use /run/cloud-init/tmp to avoid
@@ -20,8 +21,10 @@ def _tempfile_dir_arg(odir=None):
If the caller of this function (mkdtemp or mkstemp) was provided
with a 'dir' argument, then that is respected.
- @param odir: original 'dir' arg to 'mkdtemp' or other."""
-
+ @param odir: original 'dir' arg to 'mkdtemp' or other.
+ @param needs_exe: Boolean specifying whether or not exe permissions are
+ needed for tempdir. This is needed because /run is mounted noexec.
+ """
if odir is not None:
return odir
@@ -29,7 +32,9 @@ def _tempfile_dir_arg(odir=None):
if _TMPDIR:
return _TMPDIR
- if os.getuid() == 0:
+ if needs_exe:
+ tdir = _EXE_ROOT_TMPDIR
+ elif os.getuid() == 0:
tdir = _ROOT_TMPDIR
else:
tdir = os.environ.get('TMPDIR', '/tmp')
@@ -42,7 +47,8 @@ def _tempfile_dir_arg(odir=None):
def ExtendedTemporaryFile(**kwargs):
- kwargs['dir'] = _tempfile_dir_arg(kwargs.pop('dir', None))
+ kwargs['dir'] = _tempfile_dir_arg(
+ kwargs.pop('dir', None), kwargs.pop('needs_exe', False))
fh = tempfile.NamedTemporaryFile(**kwargs)
# Replace its unlink with a quiet version
# that does not raise errors when the
@@ -82,12 +88,14 @@ def tempdir(**kwargs):
def mkdtemp(**kwargs):
- kwargs['dir'] = _tempfile_dir_arg(kwargs.pop('dir', None))
+ kwargs['dir'] = _tempfile_dir_arg(
+ kwargs.pop('dir', None), kwargs.pop('needs_exe', False))
return tempfile.mkdtemp(**kwargs)
def mkstemp(**kwargs):
- kwargs['dir'] = _tempfile_dir_arg(kwargs.pop('dir', None))
+ kwargs['dir'] = _tempfile_dir_arg(
+ kwargs.pop('dir', None), kwargs.pop('needs_exe', False))
return tempfile.mkstemp(**kwargs)
# vi: ts=4 expandtab
diff --git a/cloudinit/tests/test_temp_utils.py b/cloudinit/tests/test_temp_utils.py
new file mode 100644
index 00000000..ffbb92cd
--- /dev/null
+++ b/cloudinit/tests/test_temp_utils.py
@@ -0,0 +1,101 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+"""Tests for cloudinit.temp_utils"""
+
+from cloudinit.temp_utils import mkdtemp, mkstemp
+from cloudinit.tests.helpers import CiTestCase, wrap_and_call
+
+
+class TestTempUtils(CiTestCase):
+
+ def test_mkdtemp_default_non_root(self):
+ """mkdtemp creates a dir under /tmp for the unprivileged."""
+ calls = []
+
+ def fake_mkdtemp(*args, **kwargs):
+ calls.append(kwargs)
+ return '/fake/return/path'
+
+ retval = wrap_and_call(
+ 'cloudinit.temp_utils',
+ {'os.getuid': 1000,
+ 'tempfile.mkdtemp': {'side_effect': fake_mkdtemp},
+ '_TMPDIR': {'new': None},
+ 'os.path.isdir': True},
+ mkdtemp)
+ self.assertEqual('/fake/return/path', retval)
+ self.assertEqual([{'dir': '/tmp'}], calls)
+
+ def test_mkdtemp_default_non_root_needs_exe(self):
+ """mkdtemp creates a dir under /var/tmp/cloud-init when needs_exe."""
+ calls = []
+
+ def fake_mkdtemp(*args, **kwargs):
+ calls.append(kwargs)
+ return '/fake/return/path'
+
+ retval = wrap_and_call(
+ 'cloudinit.temp_utils',
+ {'os.getuid': 1000,
+ 'tempfile.mkdtemp': {'side_effect': fake_mkdtemp},
+ '_TMPDIR': {'new': None},
+ 'os.path.isdir': True},
+ mkdtemp, needs_exe=True)
+ self.assertEqual('/fake/return/path', retval)
+ self.assertEqual([{'dir': '/var/tmp/cloud-init'}], calls)
+
+ def test_mkdtemp_default_root(self):
+ """mkdtemp creates a dir under /run/cloud-init for the privileged."""
+ calls = []
+
+ def fake_mkdtemp(*args, **kwargs):
+ calls.append(kwargs)
+ return '/fake/return/path'
+
+ retval = wrap_and_call(
+ 'cloudinit.temp_utils',
+ {'os.getuid': 0,
+ 'tempfile.mkdtemp': {'side_effect': fake_mkdtemp},
+ '_TMPDIR': {'new': None},
+ 'os.path.isdir': True},
+ mkdtemp)
+ self.assertEqual('/fake/return/path', retval)
+ self.assertEqual([{'dir': '/run/cloud-init/tmp'}], calls)
+
+ def test_mkstemp_default_non_root(self):
+ """mkstemp creates secure tempfile under /tmp for the unprivileged."""
+ calls = []
+
+ def fake_mkstemp(*args, **kwargs):
+ calls.append(kwargs)
+ return '/fake/return/path'
+
+ retval = wrap_and_call(
+ 'cloudinit.temp_utils',
+ {'os.getuid': 1000,
+ 'tempfile.mkstemp': {'side_effect': fake_mkstemp},
+ '_TMPDIR': {'new': None},
+ 'os.path.isdir': True},
+ mkstemp)
+ self.assertEqual('/fake/return/path', retval)
+ self.assertEqual([{'dir': '/tmp'}], calls)
+
+ def test_mkstemp_default_root(self):
+ """mkstemp creates a secure tempfile in /run/cloud-init for root."""
+ calls = []
+
+ def fake_mkstemp(*args, **kwargs):
+ calls.append(kwargs)
+ return '/fake/return/path'
+
+ retval = wrap_and_call(
+ 'cloudinit.temp_utils',
+ {'os.getuid': 0,
+ 'tempfile.mkstemp': {'side_effect': fake_mkstemp},
+ '_TMPDIR': {'new': None},
+ 'os.path.isdir': True},
+ mkstemp)
+ self.assertEqual('/fake/return/path', retval)
+ self.assertEqual([{'dir': '/run/cloud-init/tmp'}], calls)
+
+# vi: ts=4 expandtab
diff --git a/cloudinit/util.py b/cloudinit/util.py
index 7e9d94fc..4c01f449 100644
--- a/cloudinit/util.py
+++ b/cloudinit/util.py
@@ -1755,7 +1755,7 @@ def subp_blob_in_tempfile(blob, *args, **kwargs):
args = [tuple()]
# Use tmpdir over tmpfile to avoid 'text file busy' on execute
- with temp_utils.tempdir() as tmpd:
+ with temp_utils.tempdir(needs_exe=True) as tmpd:
tmpf = os.path.join(tmpd, basename)
if 'args' in kwargs:
kwargs['args'] = [tmpf] + list(kwargs['args'])
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-13 12:52:15 +0000