diff options
| -rw-r--r-- | ChangeLog | 1 | ||||
| -rw-r--r-- | cloudinit/CloudConfig/cc_ca_certs.py | 88 | ||||
| -rw-r--r-- | cloudinit/util.py | 47 | ||||
| -rw-r--r-- | config/cloud.cfg | 1 | ||||
| -rw-r--r-- | debian.trunk/control | 1 | ||||
| -rw-r--r-- | doc/examples/cloud-config-ca-certs.txt | 31 | ||||
| -rw-r--r-- | tests/unittests/test_handler_ca_certs.py | 178 | ||||
| -rw-r--r-- | tests/unittests/test_util.py | 199 |
8 files changed, 540 insertions, 6 deletions
@@ -19,6 +19,7 @@ in the payload parameter. (LP: #874342) - add test case framework [Mike Milner] (LP: #890851) - fix pylint warnings [Juerg Haefliger] (LP: #914739) + - add support for adding and deleting CA Certificates [Mike Milner] (LP: #915232) 0.6.2: - fix bug where update was not done unless update was explicitly set. It would not be run if 'upgrade' or packages were set to be installed diff --git a/cloudinit/CloudConfig/cc_ca_certs.py b/cloudinit/CloudConfig/cc_ca_certs.py new file mode 100644 index 00000000..c18821f9 --- /dev/null +++ b/cloudinit/CloudConfig/cc_ca_certs.py @@ -0,0 +1,88 @@ +# vi: ts=4 expandtab +# +# Author: Mike Milner <mike.milner@canonical.com> +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 3, as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. +import os +from subprocess import check_call +from cloudinit.util import (write_file, get_cfg_option_list_or_str, + delete_dir_contents) + +CA_CERT_PATH = "/usr/share/ca-certificates/" +CA_CERT_FILENAME = "cloud-init-ca-certs.crt" +CA_CERT_CONFIG = "/etc/ca-certificates.conf" +CA_CERT_SYSTEM_PATH = "/etc/ssl/certs/" + + +def update_ca_certs(): + """ + Updates the CA certificate cache on the current machine. + """ + check_call(["update-ca-certificates"]) + + +def add_ca_certs(certs): + """ + Adds certificates to the system. To actually apply the new certificates + you must also call L{update_ca_certs}. + + @param certs: A list of certificate strings. + """ + if certs: + cert_file_contents = "\n".join(certs) + cert_file_fullpath = os.path.join(CA_CERT_PATH, CA_CERT_FILENAME) + write_file(cert_file_fullpath, cert_file_contents, mode=0644) + # Append cert filename to CA_CERT_CONFIG file. + write_file(CA_CERT_CONFIG, "\n%s" % CA_CERT_FILENAME, omode="a") + + +def remove_default_ca_certs(): + """ + Removes all default trusted CA certificates from the system. To actually + apply the change you must also call L{update_ca_certs}. + """ + delete_dir_contents(CA_CERT_PATH) + delete_dir_contents(CA_CERT_SYSTEM_PATH) + write_file(CA_CERT_CONFIG, "", mode=0644) + + +def handle(_name, cfg, _cloud, log, _args): + """ + Call to handle ca-cert sections in cloud-config file. + + @param name: The module name "ca-cert" from cloud.cfg + @param cfg: A nested dict containing the entire cloud config contents. + @param cloud: The L{CloudInit} object in use. + @param log: Pre-initialized Python logger object to use for logging. + @param args: Any module arguments from cloud.cfg + """ + # If there isn't a ca-certs section in the configuration don't do anything + if "ca-certs" not in cfg: + return + ca_cert_cfg = cfg['ca-certs'] + + # If there is a remove-defaults option set to true, remove the system + # default trusted CA certs first. + if ca_cert_cfg.get("remove-defaults", False): + log.debug("removing default certificates") + remove_default_ca_certs() + + # If we are given any new trusted CA certs to add, add them. + if "trusted" in ca_cert_cfg: + trusted_certs = get_cfg_option_list_or_str(ca_cert_cfg, "trusted") + if trusted_certs: + log.debug("adding %d certificates" % len(trusted_certs)) + add_ca_certs(trusted_certs) + + # Update the system with the new cert configuration. + update_ca_certs() diff --git a/cloudinit/util.py b/cloudinit/util.py index d8d735cc..e4337e3a 100644 --- a/cloudinit/util.py +++ b/cloudinit/util.py @@ -18,6 +18,7 @@ import yaml import os import os.path +import shutil import errno import subprocess from Cheetah.Template import Template @@ -94,13 +95,24 @@ def get_cfg_option_str(yobj, key, default=None): def get_cfg_option_list_or_str(yobj, key, default=None): - if key not in yobj: + """ + Gets the C{key} config option from C{yobj} as a list of strings. If the + key is present as a single string it will be returned as a list with one + string arg. + + @param yobj: The configuration object. + @param key: The configuration key to get. + @param default: The default to return if key is not found. + @return: The configuration option as a list of strings or default if key + is not found. + """ + if not key in yobj: return default if yobj[key] is None: return [] if isinstance(yobj[key], list): return yobj[key] - return([yobj[key]]) + return [yobj[key]] # get a cfg entry by its path array @@ -114,9 +126,11 @@ def get_cfg_by_path(yobj, keyp, default=None): return(cur) -# merge values from cand into source -# if src has a key, cand will not override def mergedict(src, cand): + """ + Merge values from C{cand} into C{src}. If C{src} has a key C{cand} will + not override. Nested dictionaries are merged recursively. + """ if isinstance(src, dict) and isinstance(cand, dict): for k, v in cand.iteritems(): if k not in src: @@ -126,7 +140,30 @@ def mergedict(src, cand): return src +def delete_dir_contents(dirname): + """ + Deletes all contents of a directory without deleting the directory itself. + + @param dirname: The directory whose contents should be deleted. + """ + for node in os.listdir(dirname): + node_fullpath = os.path.join(dirname, node) + if os.path.isdir(node_fullpath): + shutil.rmtree(node_fullpath) + else: + os.unlink(node_fullpath) + + def write_file(filename, content, mode=0644, omode="wb"): + """ + Writes a file with the given content and sets the file mode as specified. + Resotres the SELinux context if possible. + + @param filename: The full path of the file to write. + @param content: The content to write to the file. + @param mode: The filesystem mode to set on the file. + @param omode: The open mode used when opening the file (r, rb, a, etc.) + """ try: os.makedirs(os.path.dirname(filename)) except OSError as e: @@ -134,7 +171,7 @@ def write_file(filename, content, mode=0644, omode="wb"): raise e f = open(filename, omode) - if mode != None: + if mode is not None: os.chmod(filename, mode) f.write(content) f.close() diff --git a/config/cloud.cfg b/config/cloud.cfg index 45538955..25d02cee 100644 --- a/config/cloud.cfg +++ b/config/cloud.cfg @@ -9,6 +9,7 @@ cloud_init_modules: - set_hostname - update_hostname - update_etc_hosts + - ca-certs - rsyslog - ssh diff --git a/debian.trunk/control b/debian.trunk/control index c877f673..eef3cd1d 100644 --- a/debian.trunk/control +++ b/debian.trunk/control @@ -8,6 +8,7 @@ Build-Depends: cdbs, python-nose, pyflakes, pylint, + python-mocker, XS-Python-Version: all Standards-Version: 3.9.1 diff --git a/doc/examples/cloud-config-ca-certs.txt b/doc/examples/cloud-config-ca-certs.txt new file mode 100644 index 00000000..5e9115a0 --- /dev/null +++ b/doc/examples/cloud-config-ca-certs.txt @@ -0,0 +1,31 @@ +#cloud-config +# +# This is an example file to configure an instance's trusted CA certificates +# system-wide for SSL/TLS trust establishment when the instance boots for the +# first time. +# +# Make sure that this file is valid yaml before starting instances. +# It should be passed as user-data when starting the instance. + +ca-certs: + # If present and set to True, the 'remove-defaults' parameter will remove + # all the default trusted CA certificates that are normally shipped with + # Ubuntu. + # This is mainly for paranoid admins - most users will not need this + # functionality. + remove-defaults: true + + # If present, the 'trusted' parameter should contain a certificate (or list + # of certificates) to add to the system as trusted CA certificates. + # Pay close attention to the YAML multiline list syntax. The example shown + # here is for a list of multiline certificates. + trusted: + - | + -----BEGIN CERTIFICATE----- + YOUR-ORGS-TRUSTED-CA-CERT-HERE + -----END CERTIFICATE----- + - | + -----BEGIN CERTIFICATE----- + YOUR-ORGS-TRUSTED-CA-CERT-HERE + -----END CERTIFICATE----- + diff --git a/tests/unittests/test_handler_ca_certs.py b/tests/unittests/test_handler_ca_certs.py new file mode 100644 index 00000000..d6513b5b --- /dev/null +++ b/tests/unittests/test_handler_ca_certs.py @@ -0,0 +1,178 @@ +from mocker import MockerTestCase + +from cloudinit.util import write_file, delete_dir_contents +from cloudinit.CloudConfig.cc_ca_certs import ( + handle, update_ca_certs, add_ca_certs, remove_default_ca_certs) +from logging import getLogger + + +class TestNoConfig(MockerTestCase): + def setUp(self): + super(TestNoConfig, self).setUp() + self.name = "ca-certs" + self.cloud_init = None + self.log = getLogger("TestNoConfig") + self.args = [] + + def test_no_config(self): + """ + Test that nothing is done if no ca-certs configuration is provided. + """ + config = {"unknown-key": "value"} + + self.mocker.replace(write_file, passthrough=False) + self.mocker.replace(update_ca_certs, passthrough=False) + self.mocker.replay() + + handle(self.name, config, self.cloud_init, self.log, self.args) + + +class TestConfig(MockerTestCase): + def setUp(self): + super(TestConfig, self).setUp() + self.name = "ca-certs" + self.cloud_init = None + self.log = getLogger("TestNoConfig") + self.args = [] + + # Mock out the functions that actually modify the system + self.mock_add = self.mocker.replace(add_ca_certs, passthrough=False) + self.mock_update = self.mocker.replace(update_ca_certs, + passthrough=False) + self.mock_remove = self.mocker.replace(remove_default_ca_certs, + passthrough=False) + # Order must be correct + self.mocker.order() + + def test_no_trusted_list(self): + """ + Test that no certificates are written if the 'trusted' key is not + present. + """ + config = {"ca-certs": {}} + + # No functions should be called + self.mock_update() + self.mocker.replay() + + handle(self.name, config, self.cloud_init, self.log, self.args) + + def test_empty_trusted_list(self): + """Test that no certificate are written if 'trusted' list is empty""" + config = {"ca-certs": {"trusted": []}} + + # No functions should be called + self.mock_update() + self.mocker.replay() + + handle(self.name, config, self.cloud_init, self.log, self.args) + + def test_single_trusted(self): + """Test that a single cert gets passed to add_ca_certs""" + config = {"ca-certs": {"trusted": ["CERT1"]}} + + self.mock_add(["CERT1"]) + self.mock_update() + self.mocker.replay() + + handle(self.name, config, self.cloud_init, self.log, self.args) + + def test_multiple_trusted(self): + """Test that multiple certs get passed to add_ca_certs""" + config = {"ca-certs": {"trusted": ["CERT1", "CERT2"]}} + + self.mock_add(["CERT1", "CERT2"]) + self.mock_update() + self.mocker.replay() + + handle(self.name, config, self.cloud_init, self.log, self.args) + + def test_remove_default_ca_certs(self): + """Test remove_defaults works as expected""" + config = {"ca-certs": {"remove-defaults": True}} + + self.mock_remove() + self.mock_update() + self.mocker.replay() + + handle(self.name, config, self.cloud_init, self.log, self.args) + + 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}} + + self.mock_update() + self.mocker.replay() + + handle(self.name, config, self.cloud_init, self.log, self.args) + + 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"]}} + + self.mock_remove() + self.mock_add(["CERT1"]) + self.mock_update() + self.mocker.replay() + + handle(self.name, config, self.cloud_init, self.log, self.args) + + +class TestAddCaCerts(MockerTestCase): + def test_no_certs_in_list(self): + """Test that no certificate are written if not provided.""" + self.mocker.replace(write_file, passthrough=False) + self.mocker.replay() + + add_ca_certs([]) + + def test_single_cert(self): + """Test adding a single certificate to the trusted CAs""" + cert = "CERT1\nLINE2\nLINE3" + + mock_write = self.mocker.replace(write_file, passthrough=False) + mock_write("/usr/share/ca-certificates/cloud-init-ca-certs.crt", + cert, mode=0644) + mock_write("/etc/ca-certificates.conf", + "\ncloud-init-ca-certs.crt", omode="a") + self.mocker.replay() + + add_ca_certs([cert]) + + 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) + + mock_write = self.mocker.replace(write_file, passthrough=False) + mock_write("/usr/share/ca-certificates/cloud-init-ca-certs.crt", + expected_cert_file, mode=0644) + mock_write("/etc/ca-certificates.conf", + "\ncloud-init-ca-certs.crt", omode="a") + self.mocker.replay() + + add_ca_certs(certs) + + +class TestUpdateCaCerts(MockerTestCase): + def test_commands(self): + mock_check_call = self.mocker.replace("subprocess.check_call", + passthrough=False) + mock_check_call(["update-ca-certificates"]) + self.mocker.replay() + + update_ca_certs() + + +class TestRemoveDefaultCaCerts(MockerTestCase): + def test_commands(self): + mock_delete_dir_contents = self.mocker.replace(delete_dir_contents, + passthrough=False) + mock_write = self.mocker.replace(write_file, passthrough=False) + + mock_delete_dir_contents("/usr/share/ca-certificates/") + mock_delete_dir_contents("/etc/ssl/certs/") + mock_write("/etc/ca-certificates.conf", "", mode=0644) + self.mocker.replay() + + remove_default_ca_certs() diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py index 34a073d9..d8da8bc9 100644 --- a/tests/unittests/test_util.py +++ b/tests/unittests/test_util.py @@ -1,15 +1,24 @@ from unittest import TestCase +from mocker import MockerTestCase +from tempfile import mkdtemp +from shutil import rmtree +import os +import stat + +from cloudinit.util import (mergedict, get_cfg_option_list_or_str, write_file, + delete_dir_contents) -from cloudinit.util import mergedict class TestMergeDict(TestCase): def test_simple_merge(self): + """Test simple non-conflict merge.""" source = {"key1": "value1"} candidate = {"key2": "value2"} result = mergedict(source, candidate) self.assertEqual({"key1": "value1", "key2": "value2"}, result) def test_nested_merge(self): + """Test nested merge.""" source = {"key1": {"key1.1": "value1.1"}} candidate = {"key1": {"key1.2": "value1.2"}} result = mergedict(source, candidate) @@ -17,37 +26,225 @@ class TestMergeDict(TestCase): {"key1": {"key1.1": "value1.1", "key1.2": "value1.2"}}, result) def test_merge_does_not_override(self): + """Test that candidate doesn't override source.""" source = {"key1": "value1", "key2": "value2"} candidate = {"key2": "value2", "key2": "NEW VALUE"} result = mergedict(source, candidate) self.assertEqual(source, result) def test_empty_candidate(self): + """Test empty candidate doesn't change source.""" source = {"key": "value"} candidate = {} result = mergedict(source, candidate) self.assertEqual(source, result) def test_empty_source(self): + """Test empty source is replaced by candidate.""" source = {} candidate = {"key": "value"} result = mergedict(source, candidate) self.assertEqual(candidate, result) def test_non_dict_candidate(self): + """Test non-dict candidate is discarded.""" source = {"key": "value"} candidate = "not a dict" result = mergedict(source, candidate) self.assertEqual(source, result) def test_non_dict_source(self): + """Test non-dict source is not modified with a dict candidate.""" source = "not a dict" candidate = {"key": "value"} result = mergedict(source, candidate) self.assertEqual(source, result) def test_neither_dict(self): + """Test if neither candidate or source is dict source wins.""" source = "source" candidate = "candidate" result = mergedict(source, candidate) self.assertEqual(source, result) + + +class TestGetCfgOptionListOrStr(TestCase): + def test_not_found_no_default(self): + """None is returned if key is not found and no default given.""" + config = {} + result = get_cfg_option_list_or_str(config, "key") + self.assertIsNone(result) + + def test_not_found_with_default(self): + """Default is returned if key is not found.""" + config = {} + result = get_cfg_option_list_or_str(config, "key", default=["DEFAULT"]) + self.assertEqual(["DEFAULT"], result) + + def test_found_with_default(self): + """Default is not returned if key is found.""" + config = {"key": ["value1"]} + result = get_cfg_option_list_or_str(config, "key", default=["DEFAULT"]) + self.assertEqual(["value1"], result) + + def test_found_convert_to_list(self): + """Single string is converted to one element list.""" + config = {"key": "value1"} + result = get_cfg_option_list_or_str(config, "key") + self.assertEqual(["value1"], result) + + def test_value_is_none(self): + """If value is None empty list is returned.""" + config = {"key": None} + result = get_cfg_option_list_or_str(config, "key") + self.assertEqual([], result) + + +class TestWriteFile(MockerTestCase): + def setUp(self): + super(TestWriteFile, self).setUp() + # Make a temp directoy for tests to use. + self.tmp = mkdtemp(prefix="unittest_") + + def tearDown(self): + super(TestWriteFile, self).tearDown() + # Clean up temp directory + rmtree(self.tmp) + + def test_basic_usage(self): + """Verify basic usage with default args.""" + path = os.path.join(self.tmp, "NewFile.txt") + contents = "Hey there" + + write_file(path, contents) + + self.assertTrue(os.path.exists(path)) + self.assertTrue(os.path.isfile(path)) + with open(path) as f: + create_contents = f.read() + self.assertEqual(contents, create_contents) + file_stat = os.stat(path) + self.assertEqual(0644, stat.S_IMODE(file_stat.st_mode)) + + def test_dir_is_created_if_required(self): + """Verifiy that directories are created is required.""" + dirname = os.path.join(self.tmp, "subdir") + path = os.path.join(dirname, "NewFile.txt") + contents = "Hey there" + + write_file(path, contents) + + self.assertTrue(os.path.isdir(dirname)) + self.assertTrue(os.path.isfile(path)) + + def test_custom_mode(self): + """Verify custom mode works properly.""" + path = os.path.join(self.tmp, "NewFile.txt") + contents = "Hey there" + + write_file(path, contents, mode=0666) + + self.assertTrue(os.path.exists(path)) + self.assertTrue(os.path.isfile(path)) + file_stat = os.stat(path) + self.assertEqual(0666, stat.S_IMODE(file_stat.st_mode)) + + def test_custom_omode(self): + """Verify custom omode works properly.""" + path = os.path.join(self.tmp, "NewFile.txt") + contents = "Hey there" + + # Create file first with basic content + with open(path, "wb") as f: + f.write("LINE1\n") + write_file(path, contents, omode="a") + + self.assertTrue(os.path.exists(path)) + self.assertTrue(os.path.isfile(path)) + with open(path) as f: + create_contents = f.read() + self.assertEqual("LINE1\nHey there", create_contents) + + def test_restorecon_if_possible_is_called(self): + """Make sure the restorecon_if_possible is called correctly.""" + path = os.path.join(self.tmp, "NewFile.txt") + contents = "Hey there" + + # Mock out the restorecon_if_possible call to test if it's called. + mock_restorecon = self.mocker.replace( + "cloudinit.util.restorecon_if_possible", passthrough=False) + mock_restorecon(path) + self.mocker.replay() + + write_file(path, contents) + + +class TestDeleteDirContents(TestCase): + def setUp(self): + super(TestDeleteDirContents, self).setUp() + # Make a temp directoy for tests to use. + self.tmp = mkdtemp(prefix="unittest_") + + def tearDown(self): + super(TestDeleteDirContents, self).tearDown() + # Clean up temp directory + rmtree(self.tmp) + + def assertDirEmpty(self, dirname): + self.assertEqual([], os.listdir(dirname)) + + def test_does_not_delete_dir(self): + """Ensure directory itself is not deleted.""" + delete_dir_contents(self.tmp) + + self.assertTrue(os.path.isdir(self.tmp)) + self.assertDirEmpty(self.tmp) + + def test_deletes_files(self): + """Single file should be deleted.""" + with open(os.path.join(self.tmp, "new_file.txt"), "wb") as f: + f.write("DELETE ME") + + delete_dir_contents(self.tmp) + + self.assertDirEmpty(self.tmp) + + def test_deletes_empty_dirs(self): + """Empty directories should be deleted.""" + os.mkdir(os.path.join(self.tmp, "new_dir")) + + delete_dir_contents(self.tmp) + + self.assertDirEmpty(self.tmp) + + def test_deletes_nested_dirs(self): + """Nested directories should be deleted.""" + os.mkdir(os.path.join(self.tmp, "new_dir")) + os.mkdir(os.path.join(self.tmp, "new_dir", "new_subdir")) + + delete_dir_contents(self.tmp) + + self.assertDirEmpty(self.tmp) + + def test_deletes_non_empty_dirs(self): + """Non-empty directories should be deleted.""" + os.mkdir(os.path.join(self.tmp, "new_dir")) + f_name = os.path.join(self.tmp, "new_dir", "new_file.txt") + with open(f_name, "wb") as f: + f.write("DELETE ME") + + delete_dir_contents(self.tmp) + + self.assertDirEmpty(self.tmp) + + def test_deletes_symlinks(self): + """Symlinks should be deleted.""" + file_name = os.path.join(self.tmp, "new_file.txt") + link_name = os.path.join(self.tmp, "new_file_link.txt") + with open(file_name, "wb") as f: + f.write("DELETE ME") + os.symlink(file_name, link_name) + + delete_dir_contents(self.tmp) + + self.assertDirEmpty(self.tmp) |