Add support for changing the password at next logon

This patch exports a new option, `password_next_logon`, which determines
what will happen at the next logon in certain conditions. The option
can accept three possible arguments: `always`, which forces the user
to change the password at next logon, 'no', which doesn't change anything
and `clear_text_injected_only`, which forces the user to change the password
at the next logon if the password comes in clear text from the metadata.

Change-Id: Ic6a0526ea9c9902e183898c42497133a135b5c53
This commit is contained in:
Claudiu Popa 2015-06-19 23:25:58 +03:00
parent 13bec72fbc
commit 47e52e2e34
8 changed files with 204 additions and 20 deletions

View File

@ -71,8 +71,8 @@ plugin.
+------------+--------------------------------+------------------+
cloudbaseinit.plugins.common.setuserpassword.SetUserPasswordPlugin
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
cloudbaseinit.plugins.windows.setuserpassword.SetUserPasswordPlugin
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Sets the cloud user's password. If a password has been provided in the
metadata during boot (user_data) it will be used, otherwise a random password
@ -80,11 +80,17 @@ will be generated, encrypted with the user's SSH public key and posted to the
metadata provider (currently supported only by the OpenStack HTTP metadata
provider).
+------------------------+-------------------------------------------------------------------------------------+---------+
| Option | Description | Default |
+========================+=====================================================================================+=========+
| *inject_user_password* | Can be set to false to avoid the injection of the password provided in the metadata | *True* |
+------------------------+-------------------------------------------------------------------------------------+---------+
+-------------------------+-------------------------------------------------------------------------------------+---------------------------+
| Option | Description | Default |
+=========================+=====================================================================================+===========================+
| *inject_user_password* | Can be set to false to avoid the injection of the password provided in the metadata | *True* |
+-------------------------+-------------------------------------------------------------------------------------+---------------------------+
| | Can control what happens with the password at the next logon. If this option | |
| | is set to `always`, the user will be forced to change the password at the next | |
| *first_logon_behaviour* | logon. If it is set to `clear_text_injected_only`, the user will be forced to | *clear_text_injected_only*|
| | change the password only if the password is a clear text password, coming from the | |
| | metadata. The last option is `no`, when the user is never forced. | |
+-------------------------+-------------------------------------------------------------------------------------+---------------------------+
cloudbaseinit.plugins.common.networkconfig.NetworkConfigPlugin
@ -109,7 +115,7 @@ cloudbaseinit.plugins.common.sshpublickeys.SetUserSSHPublicKeysPlugin
Creates an "authorized_keys" file in the user's home directory containing the
SSH keys provided in the metadata.
It is needed by the
*cloudbaseinit.plugins.common.setuserpassword.SetUserPasswordPlugin* plugin.
*cloudbaseinit.plugins.windows.setuserpassword.SetUserPasswordPlugin* plugin.
cloudbaseinit.plugins.windows.extendvolumes.ExtendVolumesPlugin

View File

@ -263,6 +263,9 @@ class WindowsUtils(base.BaseOSUtils):
ERROR_OLD_WIN_VERSION = 1150
ERROR_NO_MORE_FILES = 18
ADS_UF_PASSWORD_EXPIRED = 0x800000
PASSWORD_CHANGED_FLAG = 1
INVALID_HANDLE_VALUE = 0xFFFFFFFF
FILE_SHARE_READ = 1
@ -1108,3 +1111,11 @@ class WindowsUtils(base.BaseOSUtils):
raise exception.CloudbaseInitException(
"The given timezone name is unrecognised: %r" % timezone_name)
timezone.Timezone(windows_name).set(self)
def change_password_next_logon(self, username):
"""Force the given user to change the password at next logon."""
user = self._get_adsi_object(object_name=username,
object_type='user')
user.Put('PasswordExpired', self.PASSWORD_CHANGED_FLAG)
user.Put('UserFlags', self.ADS_UF_PASSWORD_EXPIRED)
user.SetInfo()

View File

@ -31,7 +31,7 @@ opts = [
'SetUserSSHPublicKeysPlugin',
'cloudbaseinit.plugins.windows.extendvolumes.ExtendVolumesPlugin',
'cloudbaseinit.plugins.common.userdata.UserDataPlugin',
'cloudbaseinit.plugins.common.setuserpassword.'
'cloudbaseinit.plugins.windows.setuserpassword.'
'SetUserPasswordPlugin',
'cloudbaseinit.plugins.windows.winrmlistener.'
'ConfigWinRMListenerPlugin',
@ -71,8 +71,8 @@ OLD_PLUGINS = {
'cloudbaseinit.plugins.windows.userdata.UserDataPlugin':
'cloudbaseinit.plugins.common.userdata.UserDataPlugin',
'cloudbaseinit.plugins.windows.setuserpassword.SetUserPasswordPlugin':
'cloudbaseinit.plugins.common.setuserpassword.SetUserPasswordPlugin',
'cloudbaseinit.plugins.common.setuserpassword.SetUserPasswordPlugin':
'cloudbaseinit.plugins.windows.setuserpassword.SetUserPasswordPlugin',
'cloudbaseinit.plugins.windows.localscripts.LocalScriptsPlugin':
'cloudbaseinit.plugins.common.localscripts.LocalScriptsPlugin',

View File

@ -50,18 +50,20 @@ class SetUserPasswordPlugin(base.BasePlugin):
return list(public_keys)[0]
def _get_password(self, service, shared_data):
injected = False
if CONF.inject_user_password:
password = service.get_admin_password()
else:
password = None
if password:
injected = True
LOG.warn('Using admin_pass metadata user password. Consider '
'changing it as soon as possible')
else:
password = shared_data.get(constants.SHARED_DATA_PASSWORD)
return password
return password, injected
def _set_metadata_password(self, password, service):
if service.is_password_set:
@ -93,7 +95,7 @@ class SetUserPasswordPlugin(base.BasePlugin):
LOG.info('Updating password is not required.')
return None
password = self._get_password(service, shared_data)
password, injected = self._get_password(service, shared_data)
if not password:
LOG.debug('Generating a random user password')
maximum_length = osutils.get_maximum_password_length()
@ -101,8 +103,16 @@ class SetUserPasswordPlugin(base.BasePlugin):
maximum_length)
osutils.set_user_password(user_name, password)
self.post_set_password(user_name, password,
password_injected=injected)
return password
def post_set_password(self, username, password, password_injected=False):
"""Executes post set password logic.
This is called by :meth:`execute` after the password was set.
"""
def execute(self, service, shared_data):
# TODO(alexpilotti): The username selection logic must be set in the
# CreateUserPlugin instead if using CONF.username

View File

@ -0,0 +1,65 @@
# Copyright 2015 Cloudbase Solutions Srl
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo.config import cfg
from cloudbaseinit.osutils import factory
from cloudbaseinit.plugins.common import setuserpassword
CLEAR_TEXT_INJECTED_ONLY = 'clear_text_injected_only'
ALWAYS_CHANGE = 'always'
NEVER_CHANGE = 'no'
LOGON_PASSWORD_CHANGE_OPTIONS = [
CLEAR_TEXT_INJECTED_ONLY,
NEVER_CHANGE,
ALWAYS_CHANGE,
]
opts = [
cfg.StrOpt('first_logon_behaviour',
default=CLEAR_TEXT_INJECTED_ONLY,
choices=LOGON_PASSWORD_CHANGE_OPTIONS,
help='Control the behaviour of what happens at '
'next logon. If this option is set to `always`, '
'then the user will be forced to change the password '
'at next logon. If it is set to '
'`clear_text_injected_only`, '
'then the user will have to change the password only if '
'the password is a clear text password, coming from the '
'metadata. The last option is `no`, when the user is '
'never forced to change the password.'),
]
CONF = cfg.CONF
CONF.register_opts(opts)
class SetUserPasswordPlugin(setuserpassword.SetUserPasswordPlugin):
"""Plugin for changing the password, tailored to Windows."""
def post_set_password(self, username, _, password_injected=False):
"""Post set password logic
If the option is activated, force the user to change the
password at next logon.
"""
if CONF.first_logon_behaviour == NEVER_CHANGE:
return
clear_text = CONF.first_logon_behaviour == CLEAR_TEXT_INJECTED_ONLY
always = CONF.first_logon_behaviour == ALWAYS_CHANGE
if always or (clear_text and password_injected):
osutils = factory.get_os_utils()
osutils.change_password_next_logon(username)

View File

@ -1689,3 +1689,18 @@ class TestWindowsUtils(testutils.CloudbaseInitTestBase):
self.assertEqual('dwForwardNextHop', given_route[2])
self.assertEqual('dwForwardIfIndex', given_route[3])
self.assertEqual('dwForwardMetric1', given_route[4])
@mock.patch('cloudbaseinit.osutils.windows.WindowsUtils.'
'_get_adsi_object')
def test_change_password_next_logon(self, mock_get_adsi_object):
self._winutils.change_password_next_logon(mock.sentinel.username)
mock_get_adsi_object.called_once_with(mock.sentinel.username)
user = mock_get_adsi_object.return_value
expected_put_call = [
mock.call('PasswordExpired',
self._winutils.PASSWORD_CHANGED_FLAG),
mock.call('UserFlags', self._winutils.ADS_UF_PASSWORD_EXPIRED)
]
self.assertEqual(expected_put_call, user.Put.mock_calls)
user.SetInfo.assert_called_once_with()

View File

@ -88,9 +88,10 @@ class SetUserPasswordPluginTests(unittest.TestCase):
shared_data)
if inject_password:
mock_service.get_admin_password.assert_called_with()
expected_password = (expected_password, True)
else:
self.assertFalse(mock_service.get_admin_password.called)
expected_password = mock.sentinel.create_user_password
expected_password = (mock.sentinel.create_user_password, False)
self.assertEqual(expected_password, response)
@ -154,26 +155,30 @@ class SetUserPasswordPluginTests(unittest.TestCase):
'updated in the instance metadata']
self.assertEqual(expected_logging, snatcher.output)
@mock.patch('cloudbaseinit.plugins.common.setuserpassword.'
'SetUserPasswordPlugin.post_set_password')
@mock.patch('cloudbaseinit.plugins.common.setuserpassword.'
'SetUserPasswordPlugin._get_password')
def _test_set_password(self, mock_get_password, password,
can_update_password, is_password_changed):
def _test_set_password(self, mock_get_password, mock_post_set_password,
password, can_update_password,
is_password_changed, injected=False):
expected_password = password
expected_logging = []
user = 'fake_user'
mock_get_password.return_value = password
mock_get_password.return_value = (password, injected)
mock_service = mock.MagicMock()
mock_osutils = mock.MagicMock()
mock_osutils.get_maximum_password_length.return_value = None
mock_osutils.generate_random_password.return_value = 'fake-password'
mock_osutils.generate_random_password.return_value = expected_password
mock_service.can_update_password = can_update_password
mock_service.is_password_changed.return_value = is_password_changed
with testutils.LogSnatcher('cloudbaseinit.plugins.common.'
'setuserpassword') as snatcher:
response = self._setpassword_plugin._set_password(
mock_service, mock_osutils, 'fake_user',
mock_service, mock_osutils, user,
mock.sentinel.shared_data)
if can_update_password and not is_password_changed:
@ -182,7 +187,7 @@ class SetUserPasswordPluginTests(unittest.TestCase):
if not password:
expected_logging.append('Generating a random user password')
expected_password = 'fake-password'
expected_password = password
if not can_update_password or is_password_changed:
mock_get_password.assert_called_once_with(
@ -190,6 +195,9 @@ class SetUserPasswordPluginTests(unittest.TestCase):
self.assertEqual(expected_password, response)
self.assertEqual(expected_logging, snatcher.output)
if password and can_update_password and is_password_changed:
mock_post_set_password.assert_called_once_with(
user, expected_password, password_injected=injected)
def test_set_password(self):
self._test_set_password(password='Password',

View File

@ -0,0 +1,69 @@
# Copyright 2015 Cloudbase Solutions Srl
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import unittest
import mock
from cloudbaseinit.plugins.windows import setuserpassword
from cloudbaseinit.tests import testutils
@mock.patch.object(setuserpassword.factory, 'get_os_utils')
class TestSetUserPassword(unittest.TestCase):
def setUp(self):
self._plugin = setuserpassword.SetUserPasswordPlugin()
@testutils.ConfPatcher('first_logon_behaviour',
setuserpassword.NEVER_CHANGE)
def test_post_set_password_never_change(self, mock_get_os_utils):
self._plugin.post_set_password(mock.sentinel.username,
mock.sentinel.password)
self.assertFalse(mock_get_os_utils.called)
@testutils.ConfPatcher('first_logon_behaviour',
setuserpassword.ALWAYS_CHANGE)
def test_post_set_password_always(self, mock_get_os_utils):
self._plugin.post_set_password(mock.sentinel.username,
mock.sentinel.password)
self.assertTrue(mock_get_os_utils.called)
osutils = mock_get_os_utils.return_value
osutils.change_password_next_logon.assert_called_once_with(
mock.sentinel.username)
@testutils.ConfPatcher('first_logon_behaviour',
setuserpassword.CLEAR_TEXT_INJECTED_ONLY)
def test_post_set_password_clear_text_password_not_injected(
self, mock_get_os_utils):
self._plugin.post_set_password(mock.sentinel.username,
mock.sentinel.password,
password_injected=False)
self.assertFalse(mock_get_os_utils.called)
@testutils.ConfPatcher('first_logon_behaviour',
setuserpassword.CLEAR_TEXT_INJECTED_ONLY)
def test_post_set_password_clear_text_password_injected(
self, mock_get_os_utils):
self._plugin.post_set_password(mock.sentinel.username,
mock.sentinel.password,
password_injected=True)
self.assertTrue(mock_get_os_utils.called)
osutils = mock_get_os_utils.return_value
osutils.change_password_next_logon.assert_called_once_with(
mock.sentinel.username)