cloud-init-dev team mailing list archive
-
cloud-init-dev team
-
Mailing list archive
-
Message #05445
[Merge] ~raharper/cloud-init:ubuntu/devel/newupstream-20180910 into cloud-init:ubuntu/devel
Ryan Harper has proposed merging ~raharper/cloud-init:ubuntu/devel/newupstream-20180910 into cloud-init:ubuntu/devel.
Commit message:
cloud-init (18.3-39-g757247f9-0ubuntu1) cosmic; urgency=medium
* New upstream snapshot.
- config: disable ssh access to a configured user account
-- Ryan Harper <ryan.harper@xxxxxxxxxxxxx> Mon, 10 Sep 2018 17:11:51 -0500
Requested reviews:
cloud-init commiters (cloud-init-dev)
For more details, see:
https://code.launchpad.net/~raharper/cloud-init/+git/cloud-init/+merge/354656
--
Your team cloud-init commiters is requested to review the proposed merge of ~raharper/cloud-init:ubuntu/devel/newupstream-20180910 into cloud-init:ubuntu/devel.
diff --git a/cloudinit/config/cc_ssh.py b/cloudinit/config/cc_ssh.py
index 45204a0..f8f7cb3 100755
--- a/cloudinit/config/cc_ssh.py
+++ b/cloudinit/config/cc_ssh.py
@@ -101,10 +101,6 @@ from cloudinit.distros import ug_util
from cloudinit import ssh_util
from cloudinit import util
-DISABLE_ROOT_OPTS = (
- "no-port-forwarding,no-agent-forwarding,"
- "no-X11-forwarding,command=\"echo \'Please login as the user \\\"$USER\\\""
- " rather than the user \\\"root\\\".\';echo;sleep 10\"")
GENERATE_KEY_NAMES = ['rsa', 'dsa', 'ecdsa', 'ed25519']
KEY_FILE_TPL = '/etc/ssh/ssh_host_%s_key'
@@ -185,7 +181,7 @@ def handle(_name, cfg, cloud, log, _args):
(user, _user_config) = ug_util.extract_default(users)
disable_root = util.get_cfg_option_bool(cfg, "disable_root", True)
disable_root_opts = util.get_cfg_option_str(cfg, "disable_root_opts",
- DISABLE_ROOT_OPTS)
+ ssh_util.DISABLE_USER_OPTS)
keys = cloud.get_public_ssh_keys() or []
if "ssh_authorized_keys" in cfg:
@@ -207,6 +203,7 @@ def apply_credentials(keys, user, disable_root, disable_root_opts):
if not user:
user = "NONE"
key_prefix = disable_root_opts.replace('$USER', user)
+ key_prefix = key_prefix.replace('$DISABLE_USER', 'root')
else:
key_prefix = ''
diff --git a/cloudinit/config/cc_users_groups.py b/cloudinit/config/cc_users_groups.py
index c95bdaa..c32a743 100644
--- a/cloudinit/config/cc_users_groups.py
+++ b/cloudinit/config/cc_users_groups.py
@@ -52,8 +52,17 @@ config keys for an entry in ``users`` are as follows:
associated with the address, username and SSH keys will be requested from
there. Default: none
- ``ssh_authorized_keys``: Optional. List of ssh keys to add to user's
- authkeys file. Default: none
- - ``ssh_import_id``: Optional. SSH id to import for user. Default: none
+ authkeys file. Default: none. This key can not be combined with
+ ``ssh_redirect_user``.
+ - ``ssh_import_id``: Optional. SSH id to import for user. Default: none.
+ This key can not be combined with ``ssh_redirect_user``.
+ - ``ssh_redirect_user``: Optional. Boolean set to true to disable SSH
+ logins for this user. When specified, all cloud meta-data public ssh
+ keys will be set up in a disabled state for this username. Any ssh login
+ as this username will timeout and prompt with a message to login instead
+ as the configured <default_username> for this instance. Default: false.
+ This key can not be combined with ``ssh_import_id`` or
+ ``ssh_authorized_keys``.
- ``sudo``: Optional. Sudo rule to use, list of sudo rules to use or False.
Default: none. An absence of sudo key, or a value of none or false
will result in no sudo rules being written for the user.
@@ -101,6 +110,7 @@ config keys for an entry in ``users`` are as follows:
selinux_user: <selinux username>
shell: <shell path>
snapuser: <email>
+ ssh_redirect_user: <true/false>
ssh_authorized_keys:
- <key>
- <key>
@@ -114,17 +124,44 @@ config keys for an entry in ``users`` are as follows:
# since the module attribute 'distros'
# is a list of distros that are supported, not a sub-module
from cloudinit.distros import ug_util
+from cloudinit import log as logging
from cloudinit.settings import PER_INSTANCE
+LOG = logging.getLogger(__name__)
+
frequency = PER_INSTANCE
def handle(name, cfg, cloud, _log, _args):
(users, groups) = ug_util.normalize_users_groups(cfg, cloud.distro)
+ (default_user, _user_config) = ug_util.extract_default(users)
+ cloud_keys = cloud.get_public_ssh_keys() or []
for (name, members) in groups.items():
cloud.distro.create_group(name, members)
for (user, config) in users.items():
+ ssh_redirect_user = config.pop("ssh_redirect_user", False)
+ if ssh_redirect_user:
+ if 'ssh_authorized_keys' in config or 'ssh_import_id' in config:
+ raise ValueError(
+ 'Not creating user %s. ssh_redirect_user cannot be'
+ ' provided with ssh_import_id or ssh_authorized_keys' %
+ user)
+ if ssh_redirect_user not in (True, 'default'):
+ raise ValueError(
+ 'Not creating user %s. Invalid value of'
+ ' ssh_redirect_user: %s. Expected values: true, default'
+ ' or false.' % (user, ssh_redirect_user))
+ if default_user is None:
+ LOG.warning(
+ 'Ignoring ssh_redirect_user: %s for %s.'
+ ' No default_user defined.'
+ ' Perhaps missing cloud configuration users: '
+ ' [default, ..].',
+ ssh_redirect_user, user)
+ else:
+ config['ssh_redirect_user'] = default_user
+ config['cloud_public_ssh_keys'] = cloud_keys
cloud.distro.create_user(user, **config)
# vi: ts=4 expandtab
diff --git a/cloudinit/config/tests/test_ssh.py b/cloudinit/config/tests/test_ssh.py
index 7441d9e..c8a4271 100644
--- a/cloudinit/config/tests/test_ssh.py
+++ b/cloudinit/config/tests/test_ssh.py
@@ -2,6 +2,7 @@
from cloudinit.config import cc_ssh
+from cloudinit import ssh_util
from cloudinit.tests.helpers import CiTestCase, mock
MODPATH = "cloudinit.config.cc_ssh."
@@ -15,8 +16,7 @@ class TestHandleSsh(CiTestCase):
"""Apply keys for the given user and root."""
keys = ["key1"]
user = "clouduser"
- options = cc_ssh.DISABLE_ROOT_OPTS
- cc_ssh.apply_credentials(keys, user, False, options)
+ cc_ssh.apply_credentials(keys, user, False, ssh_util.DISABLE_USER_OPTS)
self.assertEqual([mock.call(set(keys), user),
mock.call(set(keys), "root", options="")],
m_setup_keys.call_args_list)
@@ -25,8 +25,7 @@ class TestHandleSsh(CiTestCase):
"""Apply keys for root only."""
keys = ["key1"]
user = None
- options = cc_ssh.DISABLE_ROOT_OPTS
- cc_ssh.apply_credentials(keys, user, False, options)
+ cc_ssh.apply_credentials(keys, user, False, ssh_util.DISABLE_USER_OPTS)
self.assertEqual([mock.call(set(keys), "root", options="")],
m_setup_keys.call_args_list)
@@ -34,9 +33,10 @@ class TestHandleSsh(CiTestCase):
"""Apply keys for the given user and disable root ssh."""
keys = ["key1"]
user = "clouduser"
- options = cc_ssh.DISABLE_ROOT_OPTS
+ options = ssh_util.DISABLE_USER_OPTS
cc_ssh.apply_credentials(keys, user, True, options)
options = options.replace("$USER", user)
+ options = options.replace("$DISABLE_USER", "root")
self.assertEqual([mock.call(set(keys), user),
mock.call(set(keys), "root", options=options)],
m_setup_keys.call_args_list)
@@ -45,9 +45,10 @@ class TestHandleSsh(CiTestCase):
"""Apply keys no user and disable root ssh."""
keys = ["key1"]
user = None
- options = cc_ssh.DISABLE_ROOT_OPTS
+ options = ssh_util.DISABLE_USER_OPTS
cc_ssh.apply_credentials(keys, user, True, options)
options = options.replace("$USER", "NONE")
+ options = options.replace("$DISABLE_USER", "root")
self.assertEqual([mock.call(set(keys), "root", options=options)],
m_setup_keys.call_args_list)
@@ -66,7 +67,8 @@ class TestHandleSsh(CiTestCase):
cloud = self.tmp_cloud(
distro='ubuntu', metadata={'public-keys': keys})
cc_ssh.handle("name", cfg, cloud, None, None)
- options = cc_ssh.DISABLE_ROOT_OPTS.replace("$USER", "NONE")
+ options = ssh_util.DISABLE_USER_OPTS.replace("$USER", "NONE")
+ options = options.replace("$DISABLE_USER", "root")
m_glob.assert_called_once_with('/etc/ssh/ssh_host_*key*')
self.assertIn(
[mock.call('/etc/ssh/ssh_host_rsa_key'),
@@ -94,7 +96,8 @@ class TestHandleSsh(CiTestCase):
distro='ubuntu', metadata={'public-keys': keys})
cc_ssh.handle("name", cfg, cloud, None, None)
- options = cc_ssh.DISABLE_ROOT_OPTS.replace("$USER", user)
+ options = ssh_util.DISABLE_USER_OPTS.replace("$USER", user)
+ options = options.replace("$DISABLE_USER", "root")
self.assertEqual([mock.call(set(keys), user),
mock.call(set(keys), "root", options=options)],
m_setup_keys.call_args_list)
@@ -118,7 +121,8 @@ class TestHandleSsh(CiTestCase):
distro='ubuntu', metadata={'public-keys': keys})
cc_ssh.handle("name", cfg, cloud, None, None)
- options = cc_ssh.DISABLE_ROOT_OPTS.replace("$USER", user)
+ options = ssh_util.DISABLE_USER_OPTS.replace("$USER", user)
+ options = options.replace("$DISABLE_USER", "root")
self.assertEqual([mock.call(set(keys), user),
mock.call(set(keys), "root", options=options)],
m_setup_keys.call_args_list)
diff --git a/cloudinit/config/tests/test_users_groups.py b/cloudinit/config/tests/test_users_groups.py
new file mode 100644
index 0000000..ba0afae
--- /dev/null
+++ b/cloudinit/config/tests/test_users_groups.py
@@ -0,0 +1,144 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+
+from cloudinit.config import cc_users_groups
+from cloudinit.tests.helpers import CiTestCase, mock
+
+MODPATH = "cloudinit.config.cc_users_groups"
+
+
+@mock.patch('cloudinit.distros.ubuntu.Distro.create_group')
+@mock.patch('cloudinit.distros.ubuntu.Distro.create_user')
+class TestHandleUsersGroups(CiTestCase):
+ """Test cc_users_groups handling of config."""
+
+ with_logs = True
+
+ def test_handle_no_cfg_creates_no_users_or_groups(self, m_user, m_group):
+ """Test handle with no config will not create users or groups."""
+ cfg = {} # merged cloud-config
+ # System config defines a default user for the distro.
+ sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+ 'groups': ['lxd', 'sudo'],
+ 'shell': '/bin/bash'}}
+ metadata = {}
+ cloud = self.tmp_cloud(
+ distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+ cc_users_groups.handle('modulename', cfg, cloud, None, None)
+ m_user.assert_not_called()
+ m_group.assert_not_called()
+
+ def test_handle_users_in_cfg_calls_create_users(self, m_user, m_group):
+ """When users in config, create users with distro.create_user."""
+ cfg = {'users': ['default', {'name': 'me2'}]} # merged cloud-config
+ # System config defines a default user for the distro.
+ sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+ 'groups': ['lxd', 'sudo'],
+ 'shell': '/bin/bash'}}
+ metadata = {}
+ cloud = self.tmp_cloud(
+ distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+ cc_users_groups.handle('modulename', cfg, cloud, None, None)
+ self.assertItemsEqual(
+ m_user.call_args_list,
+ [mock.call('ubuntu', groups='lxd,sudo', lock_passwd=True,
+ shell='/bin/bash'),
+ mock.call('me2', default=False)])
+ m_group.assert_not_called()
+
+ def test_users_with_ssh_redirect_user_passes_keys(self, m_user, m_group):
+ """When ssh_redirect_user is True pass default user and cloud keys."""
+ cfg = {
+ 'users': ['default', {'name': 'me2', 'ssh_redirect_user': True}]}
+ # System config defines a default user for the distro.
+ sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+ 'groups': ['lxd', 'sudo'],
+ 'shell': '/bin/bash'}}
+ metadata = {'public-keys': ['key1']}
+ cloud = self.tmp_cloud(
+ distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+ cc_users_groups.handle('modulename', cfg, cloud, None, None)
+ self.assertItemsEqual(
+ m_user.call_args_list,
+ [mock.call('ubuntu', groups='lxd,sudo', lock_passwd=True,
+ shell='/bin/bash'),
+ mock.call('me2', cloud_public_ssh_keys=['key1'], default=False,
+ ssh_redirect_user='ubuntu')])
+ m_group.assert_not_called()
+
+ def test_users_with_ssh_redirect_user_default_str(self, m_user, m_group):
+ """When ssh_redirect_user is 'default' pass default username."""
+ cfg = {
+ 'users': ['default', {'name': 'me2',
+ 'ssh_redirect_user': 'default'}]}
+ # System config defines a default user for the distro.
+ sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+ 'groups': ['lxd', 'sudo'],
+ 'shell': '/bin/bash'}}
+ metadata = {'public-keys': ['key1']}
+ cloud = self.tmp_cloud(
+ distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+ cc_users_groups.handle('modulename', cfg, cloud, None, None)
+ self.assertItemsEqual(
+ m_user.call_args_list,
+ [mock.call('ubuntu', groups='lxd,sudo', lock_passwd=True,
+ shell='/bin/bash'),
+ mock.call('me2', cloud_public_ssh_keys=['key1'], default=False,
+ ssh_redirect_user='ubuntu')])
+ m_group.assert_not_called()
+
+ def test_users_with_ssh_redirect_user_non_default(self, m_user, m_group):
+ """Warn when ssh_redirect_user is not 'default'."""
+ cfg = {
+ 'users': ['default', {'name': 'me2',
+ 'ssh_redirect_user': 'snowflake'}]}
+ # System config defines a default user for the distro.
+ sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+ 'groups': ['lxd', 'sudo'],
+ 'shell': '/bin/bash'}}
+ metadata = {'public-keys': ['key1']}
+ cloud = self.tmp_cloud(
+ distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+ with self.assertRaises(ValueError) as context_manager:
+ cc_users_groups.handle('modulename', cfg, cloud, None, None)
+ m_group.assert_not_called()
+ self.assertEqual(
+ 'Not creating user me2. Invalid value of ssh_redirect_user:'
+ ' snowflake. Expected values: true, default or false.',
+ str(context_manager.exception))
+
+ def test_users_with_ssh_redirect_user_default_false(self, m_user, m_group):
+ """When unspecified ssh_redirect_user is false and not set up."""
+ cfg = {'users': ['default', {'name': 'me2'}]}
+ # System config defines a default user for the distro.
+ sys_cfg = {'default_user': {'name': 'ubuntu', 'lock_passwd': True,
+ 'groups': ['lxd', 'sudo'],
+ 'shell': '/bin/bash'}}
+ metadata = {'public-keys': ['key1']}
+ cloud = self.tmp_cloud(
+ distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+ cc_users_groups.handle('modulename', cfg, cloud, None, None)
+ self.assertItemsEqual(
+ m_user.call_args_list,
+ [mock.call('ubuntu', groups='lxd,sudo', lock_passwd=True,
+ shell='/bin/bash'),
+ mock.call('me2', default=False)])
+ m_group.assert_not_called()
+
+ def test_users_ssh_redirect_user_and_no_default(self, m_user, m_group):
+ """Warn when ssh_redirect_user is True and no default user present."""
+ cfg = {
+ 'users': ['default', {'name': 'me2', 'ssh_redirect_user': True}]}
+ # System config defines *no* default user for the distro.
+ sys_cfg = {}
+ metadata = {} # no public-keys defined
+ cloud = self.tmp_cloud(
+ distro='ubuntu', sys_cfg=sys_cfg, metadata=metadata)
+ cc_users_groups.handle('modulename', cfg, cloud, None, None)
+ m_user.assert_called_once_with('me2', default=False)
+ m_group.assert_not_called()
+ self.assertEqual(
+ 'WARNING: Ignoring ssh_redirect_user: True for me2. No'
+ ' default_user defined. Perhaps missing'
+ ' cloud configuration users: [default, ..].\n',
+ self.logs.getvalue())
diff --git a/cloudinit/distros/__init__.py b/cloudinit/distros/__init__.py
old mode 100755
new mode 100644
index d9101ce..b8a48e8
--- a/cloudinit/distros/__init__.py
+++ b/cloudinit/distros/__init__.py
@@ -381,6 +381,9 @@ class Distro(object):
"""
Add a user to the system using standard GNU tools
"""
+ # XXX need to make add_user idempotent somehow as we
+ # still want to add groups or modify ssh keys on pre-existing
+ # users in the image.
if util.is_user(name):
LOG.info("User %s already exists, skipping.", name)
return
@@ -547,10 +550,24 @@ class Distro(object):
LOG.warning("Invalid type '%s' detected for"
" 'ssh_authorized_keys', expected list,"
" string, dict, or set.", type(keys))
+ keys = []
else:
keys = set(keys) or []
- ssh_util.setup_user_keys(keys, name, options=None)
-
+ ssh_util.setup_user_keys(set(keys), name)
+ if 'ssh_redirect_user' in kwargs:
+ cloud_keys = kwargs.get('cloud_public_ssh_keys', [])
+ if not cloud_keys:
+ LOG.warning(
+ 'Unable to disable ssh logins for %s given'
+ ' ssh_redirect_user: %s. No cloud public-keys present.',
+ name, kwargs['ssh_redirect_user'])
+ else:
+ redirect_user = kwargs['ssh_redirect_user']
+ disable_option = ssh_util.DISABLE_USER_OPTS
+ disable_option = disable_option.replace('$USER', redirect_user)
+ disable_option = disable_option.replace('$DISABLE_USER', name)
+ ssh_util.setup_user_keys(
+ set(cloud_keys), name, options=disable_option)
return True
def lock_passwd(self, name):
diff --git a/cloudinit/ssh_util.py b/cloudinit/ssh_util.py
index 73c3177..3f99b58 100644
--- a/cloudinit/ssh_util.py
+++ b/cloudinit/ssh_util.py
@@ -41,6 +41,12 @@ VALID_KEY_TYPES = (
)
+DISABLE_USER_OPTS = (
+ "no-port-forwarding,no-agent-forwarding,"
+ "no-X11-forwarding,command=\"echo \'Please login as the user \\\"$USER\\\""
+ " rather than the user \\\"$DISABLE_USER\\\".\';echo;sleep 10\"")
+
+
class AuthKeyLine(object):
def __init__(self, source, keytype=None, base64=None,
comment=None, options=None):
diff --git a/debian/changelog b/debian/changelog
index 5f738e9..6b1a35f 100644
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,3 +1,10 @@
+cloud-init (18.3-39-g757247f9-0ubuntu1) cosmic; urgency=medium
+
+ * New upstream snapshot.
+ - config: disable ssh access to a configured user account
+
+ -- Ryan Harper <ryan.harper@xxxxxxxxxxxxx> Mon, 10 Sep 2018 17:11:51 -0500
+
cloud-init (18.3-38-gd47d404e-0ubuntu1) cosmic; urgency=medium
* New upstream snapshot.
diff --git a/doc/examples/cloud-config-user-groups.txt b/doc/examples/cloud-config-user-groups.txt
index 01ecad7..6a363b7 100644
--- a/doc/examples/cloud-config-user-groups.txt
+++ b/doc/examples/cloud-config-user-groups.txt
@@ -36,6 +36,8 @@ users:
- <ssh pub key 1>
- <ssh pub key 2>
- snapuser: joe@xxxxxxxxxx
+ - name: nosshlogins
+ ssh_redirect_user: true
# Valid Values:
# name: The user's login name
@@ -76,6 +78,13 @@ users:
# no_log_init: When set to true, do not initialize lastlog and faillog database.
# ssh_import_id: Optional. Import SSH ids
# ssh_authorized_keys: Optional. [list] Add keys to user's authorized keys file
+# ssh_redirect_user: Optional. [bool] Set true to block ssh logins for cloud
+# ssh public keys and emit a message redirecting logins to
+# use <default_username> instead. This option only disables cloud
+# provided public-keys. An error will be raised if ssh_authorized_keys
+# or ssh_import_id is provided for the same user.
+#
+# ssh_authorized_keys.
# sudo: Defaults to none. Accepts a sudo rule string, a list of sudo rule
# strings or False to explicitly deny sudo usage. Examples:
#
diff --git a/doc/examples/cloud-config.txt b/doc/examples/cloud-config.txt
index 774f66b..eb84dcf 100644
--- a/doc/examples/cloud-config.txt
+++ b/doc/examples/cloud-config.txt
@@ -232,9 +232,22 @@ disable_root: false
# respective key in /root/.ssh/authorized_keys if disable_root is true
# see 'man authorized_keys' for more information on what you can do here
#
-# The string '$USER' will be replaced with the username of the default user
-#
-# disable_root_opts: no-port-forwarding,no-agent-forwarding,no-X11-forwarding,command="echo 'Please login as the user \"$USER\" rather than the user \"root\".';echo;sleep 10"
+# The string '$USER' will be replaced with the username of the default user.
+# The string '$DISABLE_USER' will be replaced with the username to disable.
+#
+# disable_root_opts: no-port-forwarding,no-agent-forwarding,no-X11-forwarding,command="echo 'Please login as the user \"$USER\" rather than the user \"$DISABLE_USER\".';echo;sleep 10"
+
+# disable ssh access for non-root-users
+# To disable ssh access for non-root users, ssh_redirect_user: true can be
+# provided for any use in the 'users' list. This will prompt any ssh login
+# attempts as that user with a message like that in disable_root_opts which
+# redirects the person to login as <default_username>
+# This option can not be combined with either ssh_authorized_keys or
+# ssh_import_id.
+users:
+ - default
+ - name: blockeduser
+ ssh_redirect_user: true
# set the locale to a given locale
diff --git a/tests/unittests/test_distros/test_create_users.py b/tests/unittests/test_distros/test_create_users.py
index 07176ca..c3f258d 100644
--- a/tests/unittests/test_distros/test_create_users.py
+++ b/tests/unittests/test_distros/test_create_users.py
@@ -1,7 +1,10 @@
# This file is part of cloud-init. See LICENSE file for license information.
+import re
+
from cloudinit import distros
-from cloudinit.tests.helpers import (TestCase, mock)
+from cloudinit import ssh_util
+from cloudinit.tests.helpers import (CiTestCase, mock)
class MyBaseDistro(distros.Distro):
@@ -44,8 +47,12 @@ class MyBaseDistro(distros.Distro):
@mock.patch("cloudinit.distros.util.system_is_snappy", return_value=False)
@mock.patch("cloudinit.distros.util.subp")
-class TestCreateUser(TestCase):
+class TestCreateUser(CiTestCase):
+
+ with_logs = True
+
def setUp(self):
+ super(TestCreateUser, self).setUp()
self.dist = MyBaseDistro()
def _useradd2call(self, args):
@@ -153,4 +160,84 @@ class TestCreateUser(TestCase):
[self._useradd2call([user, '-m']),
mock.call(['passwd', '-l', user])])
+ @mock.patch('cloudinit.ssh_util.setup_user_keys')
+ def test_setup_ssh_authorized_keys_with_string(
+ self, m_setup_user_keys, m_subp, m_is_snappy):
+ """ssh_authorized_keys allows string and calls setup_user_keys."""
+ user = 'foouser'
+ self.dist.create_user(user, ssh_authorized_keys='mykey')
+ self.assertEqual(
+ m_subp.call_args_list,
+ [self._useradd2call([user, '-m']),
+ mock.call(['passwd', '-l', user])])
+ m_setup_user_keys.assert_called_once_with(set(['mykey']), user)
+
+ @mock.patch('cloudinit.ssh_util.setup_user_keys')
+ def test_setup_ssh_authorized_keys_with_list(
+ self, m_setup_user_keys, m_subp, m_is_snappy):
+ """ssh_authorized_keys allows lists and calls setup_user_keys."""
+ user = 'foouser'
+ self.dist.create_user(user, ssh_authorized_keys=['key1', 'key2'])
+ self.assertEqual(
+ m_subp.call_args_list,
+ [self._useradd2call([user, '-m']),
+ mock.call(['passwd', '-l', user])])
+ m_setup_user_keys.assert_called_once_with(set(['key1', 'key2']), user)
+
+ @mock.patch('cloudinit.ssh_util.setup_user_keys')
+ def test_setup_ssh_authorized_keys_with_integer(
+ self, m_setup_user_keys, m_subp, m_is_snappy):
+ """ssh_authorized_keys warns on non-iterable/string type."""
+ user = 'foouser'
+ self.dist.create_user(user, ssh_authorized_keys=-1)
+ m_setup_user_keys.assert_called_once_with(set([]), user)
+ match = re.match(
+ r'.*WARNING: Invalid type \'<(type|class) \'int\'>\' detected for'
+ ' \'ssh_authorized_keys\'.*',
+ self.logs.getvalue(),
+ re.DOTALL)
+ self.assertIsNotNone(
+ match, 'Missing ssh_authorized_keys invalid type warning')
+
+ @mock.patch('cloudinit.ssh_util.setup_user_keys')
+ def test_create_user_with_ssh_redirect_user_no_cloud_keys(
+ self, m_setup_user_keys, m_subp, m_is_snappy):
+ """Log a warning when trying to redirect a user no cloud ssh keys."""
+ user = 'foouser'
+ self.dist.create_user(user, ssh_redirect_user='someuser')
+ self.assertIn(
+ 'WARNING: Unable to disable ssh logins for foouser given '
+ 'ssh_redirect_user: someuser. No cloud public-keys present.\n',
+ self.logs.getvalue())
+ m_setup_user_keys.assert_not_called()
+
+ @mock.patch('cloudinit.ssh_util.setup_user_keys')
+ def test_create_user_with_ssh_redirect_user_with_cloud_keys(
+ self, m_setup_user_keys, m_subp, m_is_snappy):
+ """Disable ssh when ssh_redirect_user and cloud ssh keys are set."""
+ user = 'foouser'
+ self.dist.create_user(
+ user, ssh_redirect_user='someuser', cloud_public_ssh_keys=['key1'])
+ disable_prefix = ssh_util.DISABLE_USER_OPTS
+ disable_prefix = disable_prefix.replace('$USER', 'someuser')
+ disable_prefix = disable_prefix.replace('$DISABLE_USER', user)
+ m_setup_user_keys.assert_called_once_with(
+ set(['key1']), 'foouser', options=disable_prefix)
+
+ @mock.patch('cloudinit.ssh_util.setup_user_keys')
+ def test_create_user_with_ssh_redirect_user_does_not_disable_auth_keys(
+ self, m_setup_user_keys, m_subp, m_is_snappy):
+ """Do not disable ssh_authorized_keys when ssh_redirect_user is set."""
+ user = 'foouser'
+ self.dist.create_user(
+ user, ssh_authorized_keys='auth1', ssh_redirect_user='someuser',
+ cloud_public_ssh_keys=['key1'])
+ disable_prefix = ssh_util.DISABLE_USER_OPTS
+ disable_prefix = disable_prefix.replace('$USER', 'someuser')
+ disable_prefix = disable_prefix.replace('$DISABLE_USER', user)
+ self.assertEqual(
+ m_setup_user_keys.call_args_list,
+ [mock.call(set(['auth1']), user), # not disabled
+ mock.call(set(['key1']), 'foouser', options=disable_prefix)])
+
# vi: ts=4 expandtab
Follow ups
