# This file is part of cloud-init. See LICENSE file for license information.
from cloudinit import cloud
from cloudinit import distros
from cloudinit.config import cc_ca_certs
from cloudinit import helpers
from cloudinit import subp
from cloudinit import util
from cloudinit.tests.helpers import TestCase
import logging
import shutil
import tempfile
import unittest
from contextlib import ExitStack
from unittest import mock
class TestNoConfig(unittest.TestCase):
def setUp(self):
super(TestNoConfig, self).setUp()
self.name = "ca-certs"
self.cloud_init = None
self.log = logging.getLogger("TestNoConfig")
self.args = []
def test_no_config(self):
"""
Test that nothing is done if no ca-certs configuration is provided.
"""
config = util.get_builtin_cfg()
with ExitStack() as mocks:
util_mock = mocks.enter_context(
mock.patch.object(util, 'write_file'))
certs_mock = mocks.enter_context(
mock.patch.object(cc_ca_certs, 'update_ca_certs'))
cc_ca_certs.handle(self.name, config, self.cloud_init, self.log,
self.args)
self.assertEqual(util_mock.call_count, 0)
self.assertEqual(certs_mock.call_count, 0)
class TestConfig(TestCase):
def setUp(self):
super(TestConfig, self).setUp()
self.name = "ca-certs"
distro = self._fetch_distro('ubuntu')
self.paths = None
self.cloud = cloud.Cloud(None, self.paths, None, distro, None)
self.log = logging.getLogger("TestNoConfig")
self.args = []
self.mocks = ExitStack()
self.addCleanup(self.mocks.close)
# Mock out the functions that actually modify the system
self.mock_add = self.mocks.enter_context(
mock.patch.object(cc_ca_certs, 'add_ca_certs'))
self.mock_update = self.mocks.enter_context(
mock.patch.object(cc_ca_certs, 'update_ca_certs'))
self.mock_remove = self.mocks.enter_context(
mock.patch.object(cc_ca_certs, 'remove_default_ca_certs'))
def _fetch_distro(self, kind):
cls = distros.fetch(kind)
paths = helpers.Paths({})
return cls(kind, {}, paths)
def test_no_trusted_list(self):
"""
Test that no certificates are written if the 'trusted' key is not
present.
"""
config = {"ca-certs": {}}
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 0)
def test_empty_trusted_list(self):
"""Test that no certificate are written if 'trusted' list is empty."""
config = {"ca-certs": {"trusted": []}}
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 0)
def test_single_trusted(self):
"""Test that a single cert gets passed to add_ca_certs."""
config = {"ca-certs": {"trusted": ["CERT1"]}}
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.mock_add.assert_called_once_with(['CERT1'])
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 0)
def test_multiple_trusted(self):
"""Test that multiple certs get passed to add_ca_certs."""
config = {"ca-certs": {"trusted": ["CERT1", "CERT2"]}}
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.mock_add.assert_called_once_with(['CERT1', 'CERT2'])
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 0)
def test_remove_default_ca_certs(self):
"""Test remove_defaults works as expected."""
config = {"ca-certs": {"remove-defaults": True}}
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 1)
def test_no_remove_defaults_if_false(self):
"""Test remove_defaults is not called when config value is False."""
config = {"ca-certs": {"remove-defaults": False}}
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 0)
def test_correct_order_for_remove_then_add(self):
"""Test remove_defaults is not called when config value is False."""
config = {"ca-certs": {"remove-defaults": True, "trusted": ["CERT1"]}}
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.mock_add.assert_called_once_with(['CERT1'])
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 1)
class TestAddCaCerts(TestCase):
def setUp(self):
super(TestAddCaCerts, self).setUp()
tmpdir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, tmpdir)
self.paths = helpers.Paths({
'cloud_dir': tmpdir,
})
def test_no_certs_in_list(self):
"""Test that no certificate are written if not provided."""
with mock.patch.object(util, 'write_file') as mockobj:
cc_ca_certs.add_ca_certs([])
self.assertEqual(mockobj.call_count, 0)
def test_single_cert_trailing_cr(self):
"""Test adding a single certificate to the trusted CAs
when existing ca-certificates has trailing newline"""
cert = "CERT1\nLINE2\nLINE3"
ca_certs_content = "line1\nline2\ncloud-init-ca-certs.crt\nline3\n"
expected = "line1\nline2\nline3\ncloud-init-ca-certs.crt\n"
with ExitStack() as mocks:
mock_write = mocks.enter_context(
mock.patch.object(util, 'write_file'))
mock_load = mocks.enter_context(
mock.patch.object(util, 'load_file',
return_value=ca_certs_content))
cc_ca_certs.add_ca_certs([cert])
mock_write.assert_has_calls([
mock.call("/usr/share/ca-certificates/cloud-init-ca-certs.crt",
cert, mode=0o644),
mock.call("/etc/ca-certificates.conf", expected, omode="wb")])
mock_load.assert_called_once_with("/etc/ca-certificates.conf")
def test_single_cert_no_trailing_cr(self):
"""Test adding a single certificate to the trusted CAs
when existing ca-certificates has no trailing newline"""
cert = "CERT1\nLINE2\nLINE3"
ca_certs_content = "line1\nline2\nline3"
with ExitStack() as mocks:
mock_write = mocks.enter_context(
mock.patch.object(util, 'write_file'))
mock_load = mocks.enter_context(
mock.patch.object(util, 'load_file',
return_value=ca_certs_content))
cc_ca_certs.add_ca_certs([cert])
mock_write.assert_has_calls([
mock.call("/usr/share/ca-certificates/cloud-init-ca-certs.crt",
cert, mode=0o644),
mock.call("/etc/ca-certificates.conf",
"%s\n%s\n" % (ca_certs_content,
"cloud-init-ca-certs.crt"),
omode="wb")])
mock_load.assert_called_once_with("/etc/ca-certificates.conf")
def test_single_cert_to_empty_existing_ca_file(self):
"""Test adding a single certificate to the trusted CAs
when existing ca-certificates.conf is empty"""
cert = "CERT1\nLINE2\nLINE3"
expected = "cloud-init-ca-certs.crt\n"
with ExitStack() as mocks:
mock_write = mocks.enter_context(
mock.patch.object(util, 'write_file', autospec=True))
mock_stat = mocks.enter_context(
mock.patch("cloudinit.config.cc_ca_certs.os.stat")
)
mock_stat.return_value.st_size = 0
cc_ca_certs.add_ca_certs([cert])
mock_write.assert_has_calls([
mock.call("/usr/share/ca-certificates/cloud-init-ca-certs.crt",
cert, mode=0o644),
mock.call("/etc/ca-certificates.conf", expected, omode="wb")])
def test_multiple_certs(self):
"""Test adding multiple certificates to the trusted CAs."""
certs = ["CERT1\nLINE2\nLINE3", "CERT2\nLINE2\nLINE3"]
expected_cert_file = "\n".join(certs)
ca_certs_content = "line1\nline2\nline3"
with ExitStack() as mocks:
mock_write = mocks.enter_context(
mock.patch.object(util, 'write_file'))
mock_load = mocks.enter_context(
mock.patch.object(util, 'load_file',
return_value=ca_certs_content))
cc_ca_certs.add_ca_certs(certs)
mock_write.assert_has_calls([
mock.call("/usr/share/ca-certificates/cloud-init-ca-certs.crt",
expected_cert_file, mode=0o644),
mock.call("/etc/ca-certificates.conf",
"%s\n%s\n" % (ca_certs_content,
"cloud-init-ca-certs.crt"),
omode='wb')])
mock_load.assert_called_once_with("/etc/ca-certificates.conf")
class TestUpdateCaCerts(unittest.TestCase):
def test_commands(self):
with mock.patch.object(subp, 'subp') as mockobj:
cc_ca_certs.update_ca_certs()
mockobj.assert_called_once_with(
["update-ca-certificates"], capture=False)
class TestRemoveDefaultCaCerts(TestCase):
def setUp(self):
super(TestRemoveDefaultCaCerts, self).setUp()
tmpdir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, tmpdir)
self.paths = helpers.Paths({
'cloud_dir': tmpdir,
})
def test_commands(self):
with ExitStack() as mocks:
mock_delete = mocks.enter_context(
mock.patch.object(util, 'delete_dir_contents'))
mock_write = mocks.enter_context(
mock.patch.object(util, 'write_file'))
mock_subp = mocks.enter_context(mock.patch.object(subp, 'subp'))
cc_ca_certs.remove_default_ca_certs('ubuntu')
mock_delete.assert_has_calls([
mock.call("/usr/share/ca-certificates/"),
mock.call("/etc/ssl/certs/")])
mock_write.assert_called_once_with(
"/etc/ca-certificates.conf", "", mode=0o644)
mock_subp.assert_called_once_with(
('debconf-set-selections', '-'),
"ca-certificates ca-certificates/trust_new_crts select no")
# vi: ts=4 expandtab