Fixes winrm access denied bug for Windows versions 6.0 and 6.1

On Windows Vista, 2008, 2008 R2 and 7, changing the configuration of
the winrm service will fail with an "Access is denied" error if the
User Account Control remote restrictions are enabled.

The solution to this issue is to temporarily disable the User
Account Control remote restrictions.

https://support.microsoft.com/kb/951016

Co-Authored-By: Robert Tingirica <rtingirica@cloudbasesolutions.com>
Closes-Bug: #1379854

Change-Id: I07b61b94900d39ad0aee544e26ee37bbc94bf7b9

cloudbaseinit/plugins/windows/winrmcertificateauth.py

@@ -16,8 +16,10 @@
 
 from cloudbaseinit import exception
 from cloudbaseinit.openstack.common import log as logging
+from cloudbaseinit.osutils import factory as osutils_factory
 from cloudbaseinit.plugins import base
 from cloudbaseinit.plugins import constants
+from cloudbaseinit.utils.windows import security
 from cloudbaseinit.utils.windows import winrmconfig
 from cloudbaseinit.utils.windows import x509
 
@@ -53,6 +55,25 @@ class ConfigWinRMCertificateAuthPlugin(base.BasePlugin):
                      "as a certificate has not been provided in the metadata")
             return (base.PLUGIN_EXECUTION_DONE, False)
 
+        osutils = osutils_factory.get_os_utils()
+        security_utils = security.WindowsSecurityUtils()
+
+        # On Windows Vista, 2008, 2008 R2 and 7, changing the configuration of
+        # the winrm service will fail with an "Access is denied" error if the
+        # User Account Control remote restrictions are enabled.
+        # The solution to this issue is to temporarily disable the User Account
+        # Control remote restrictions.
+        # https://support.microsoft.com/kb/951016
+        disable_uac_remote_restrictions = (osutils.check_os_version(6, 0) and
+                                           not osutils.check_os_version(6, 2)
+                                           and security_utils
+                                           .get_uac_remote_restrictions())
+
+        try:
+            if disable_uac_remote_restrictions:
+                LOG.debug("Disabling UAC remote restrictions")
+                security_utils.set_uac_remote_restrictions(enable=False)
+
             winrm_config = winrmconfig.WinRMConfig()
             winrm_config.set_auth_config(certificate=True)
 
@@ -77,4 +98,9 @@ class ConfigWinRMCertificateAuthPlugin(base.BasePlugin):
                 winrm_config.create_cert_mapping(cert_thumprint, cert_upn,
                                                  user_name, password)
 
+        finally:
+            if disable_uac_remote_restrictions:
+                LOG.debug("Enabling UAC remote restrictions")
+                security_utils.set_uac_remote_restrictions(enable=True)
+
         return (base.PLUGIN_EXECUTION_DONE, False)

cloudbaseinit/plugins/windows/winrmlistener.py

@@ -19,6 +19,7 @@ from oslo.config import cfg
 from cloudbaseinit.openstack.common import log as logging
 from cloudbaseinit.osutils import factory as osutils_factory
 from cloudbaseinit.plugins import base
+from cloudbaseinit.utils.windows import security
 from cloudbaseinit.utils.windows import winrmconfig
 from cloudbaseinit.utils.windows import x509
 
@@ -62,10 +63,27 @@ class ConfigWinRMListenerPlugin(base.BasePlugin):
 
     def execute(self, service, shared_data):
         osutils = osutils_factory.get_os_utils()
+        security_utils = security.WindowsSecurityUtils()
 
         if not self._check_winrm_service(osutils):
             return (base.PLUGIN_EXECUTE_ON_NEXT_BOOT, False)
 
+        # On Windows Vista, 2008, 2008 R2 and 7, changing the configuration of
+        # the winrm service will fail with an "Access is denied" error if the
+        # User Account Control remote restrictions are enabled.
+        # The solution to this issue is to temporarily disable the User Account
+        # Control remote restrictions.
+        # https://support.microsoft.com/kb/951016
+        disable_uac_remote_restrictions = (osutils.check_os_version(6, 0) and
+                                           not osutils.check_os_version(6, 2)
+                                           and security_utils
+                                           .get_uac_remote_restrictions())
+
+        try:
+            if disable_uac_remote_restrictions:
+                LOG.debug("Disabling UAC remote restrictions")
+                security_utils.set_uac_remote_restrictions(enable=False)
+
             winrm_config = winrmconfig.WinRMConfig()
             winrm_config.set_auth_config(basic=CONF.winrm_enable_basic_auth)
 
@@ -78,8 +96,7 @@ class ConfigWinRMListenerPlugin(base.BasePlugin):
             if winrm_config.get_listener(protocol=protocol):
                 winrm_config.delete_listener(protocol=protocol)
 
-        winrm_config.create_listener(
+            winrm_config.create_listener(cert_thumbprint=cert_thumbprint,
-            cert_thumbprint=cert_thumbprint,
                                          protocol=protocol)
 
             listener_config = winrm_config.get_listener(protocol=protocol)
@@ -89,4 +106,9 @@ class ConfigWinRMListenerPlugin(base.BasePlugin):
             osutils.firewall_create_rule(rule_name, listener_port,
                                          osutils.PROTOCOL_TCP)
 
+        finally:
+            if disable_uac_remote_restrictions:
+                LOG.debug("Enabling UAC remote restrictions")
+                security_utils.set_uac_remote_restrictions(enable=True)
+
         return (base.PLUGIN_EXECUTION_DONE, False)

cloudbaseinit/tests/plugins/windows/test_winrmcertificateauth.py

@@ -32,14 +32,19 @@ class ConfigWinRMCertificateAuthPluginTests(unittest.TestCase):
         self._ctypes_mock = mock.MagicMock()
         self._win32com_mock = mock.MagicMock()
         self._pywintypes_mock = mock.MagicMock()
+        self._moves_mock = mock.MagicMock()
 
         self._module_patcher = mock.patch.dict(
             'sys.modules',
             {'ctypes': self._ctypes_mock,
              'win32com': self._win32com_mock,
-             'pywintypes': self._pywintypes_mock})
+             'pywintypes': self._pywintypes_mock,
+             'six.moves': self._moves_mock})
+
         self._module_patcher.start()
 
+        self._winreg_mock = self._moves_mock.winreg
+
         self.winrmcert = importlib.import_module(
             'cloudbaseinit.plugins.windows.winrmcertificateauth')
         self._certif_auth = self.winrmcert.ConfigWinRMCertificateAuthPlugin()
@@ -82,22 +87,47 @@ class ConfigWinRMCertificateAuthPluginTests(unittest.TestCase):
     @mock.patch('cloudbaseinit.utils.windows.winrmconfig.WinRMConfig')
     @mock.patch('cloudbaseinit.utils.windows.x509.CryptoAPICertManager.'
                 'import_cert')
-    def _test_execute(self, mock_import_cert, mock_WinRMConfig,
+    @mock.patch('cloudbaseinit.osutils.factory.get_os_utils')
+    @mock.patch('cloudbaseinit.utils.windows.security.WindowsSecurityUtils'
+                '.set_uac_remote_restrictions')
+    @mock.patch('cloudbaseinit.utils.windows.security.WindowsSecurityUtils'
+                '.get_uac_remote_restrictions')
+    def _test_execute(self, get_uac_rs, set_uac_rs, mock_get_os_utils,
+                      mock_import_cert, mock_WinRMConfig,
                       mock_get_credentials, cert_data, cert_upn):
+        mock_osutils = mock.MagicMock()
         mock_service = mock.MagicMock()
         mock_cert_thumprint = mock.MagicMock()
         fake_credentials = ('fake user', 'fake password')
         mock_get_credentials.return_value = fake_credentials
+
         mock_import_cert.return_value = (mock_cert_thumprint, cert_upn)
         mock_WinRMConfig.get_cert_mapping.return_value = True
         mock_service.get_client_auth_certs.return_value = [cert_data]
 
+        mock_get_os_utils.return_value = mock_osutils
+
+        expected_set_token_calls = [mock.call(enable=False),
+                                    mock.call(enable=True)]
+
+        mock_osutils.check_os_version.side_effect = [True, False]
+        get_uac_rs.return_value = True
+
+        expected_check_version_calls = [mock.call(6, 0), mock.call(6, 2)]
+
         response = self._certif_auth.execute(mock_service,
                                              shared_data='fake data')
-        mock_service.get_client_auth_certs.assert_called_once_with()
+
         if not cert_data:
             self.assertEqual((base.PLUGIN_EXECUTION_DONE, False), response)
         else:
+            mock_service.get_client_auth_certs.assert_called_once_with()
+            self.assertEqual(expected_check_version_calls,
+                             mock_osutils.check_os_version.call_args_list)
+            mock_get_os_utils.assert_called_once_with()
+            self.assertEqual(expected_set_token_calls,
+                             set_uac_rs.call_args_list)
+
             mock_get_credentials.assert_called_once_with('fake data')
             mock_import_cert.assert_called_once_with(
                 cert_data, store_name=self.winrmcert.x509.STORE_NAME_ROOT)

cloudbaseinit/tests/plugins/windows/test_winrmlistener.py

@@ -29,14 +29,17 @@ class ConfigWinRMListenerPluginTests(unittest.TestCase):
         self._mock_wintypes = mock.MagicMock()
         self._mock_pywintypes = mock.MagicMock()
         self._mock_win32 = mock.MagicMock()
+        self._moves_mock = mock.MagicMock()
 
         self._module_patcher = mock.patch.dict(
             'sys.modules',
             {'ctypes': self._mock_wintypes,
              'ctypes.wintypes': self._mock_wintypes,
              'pywintypes': self._mock_pywintypes,
-             'win32com': self._mock_win32})
+             'win32com': self._mock_win32,
+             'six.moves': self._moves_mock})
         self._module_patcher.start()
+        self._winreg_mock = self._moves_mock.winreg
 
         winrmlistener = importlib.import_module('cloudbaseinit.plugins.'
                                                 'windows.winrmlistener')
@@ -84,7 +87,12 @@ class ConfigWinRMListenerPluginTests(unittest.TestCase):
     @mock.patch('cloudbaseinit.utils.windows.winrmconfig.WinRMConfig')
     @mock.patch('cloudbaseinit.utils.windows.x509.CryptoAPICertManager'
                 '.create_self_signed_cert')
-    def _test_execute(self, mock_create_cert, mock_WinRMConfig,
+    @mock.patch('cloudbaseinit.utils.windows.security.WindowsSecurityUtils'
+                '.set_uac_remote_restrictions')
+    @mock.patch('cloudbaseinit.utils.windows.security.WindowsSecurityUtils'
+                '.get_uac_remote_restrictions')
+    def _test_execute(self, get_uac_rs, set_uac_rs, mock_create_cert,
+                      mock_WinRMConfig,
                       mock_check_winrm_service, mock_get_os_utils,
                       service_status):
         mock_service = mock.MagicMock()
@@ -98,6 +106,13 @@ class ConfigWinRMListenerPluginTests(unittest.TestCase):
         mock_WinRMConfig().get_listener.return_value = mock_listener_config
         mock_listener_config.get.return_value = 9999
 
+        mock_osutils.check_os_version.side_effect = [True, False]
+        get_uac_rs.return_value = True
+
+        expected_check_version_calls = [mock.call(6, 0), mock.call(6, 2)]
+        expected_set_token_calls = [mock.call(enable=False),
+                                    mock.call(enable=True)]
+
         response = self._winrmlistener.execute(mock_service, shared_data)
 
         mock_get_os_utils.assert_called_once_with()
@@ -107,6 +122,10 @@ class ConfigWinRMListenerPluginTests(unittest.TestCase):
             self.assertEqual((base.PLUGIN_EXECUTE_ON_NEXT_BOOT,
                               service_status), response)
         else:
+            self.assertEqual(expected_check_version_calls,
+                             mock_osutils.check_os_version.call_args_list)
+            self.assertEqual(expected_set_token_calls,
+                             set_uac_rs.call_args_list)
             mock_WinRMConfig().set_auth_config.assert_called_once_with(
                 basic=CONF.winrm_enable_basic_auth)
             mock_create_cert.assert_called_once_with(

cloudbaseinit/tests/utils/windows/test_security.py

@@ -0,0 +1,101 @@
+# Copyright 2014 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 importlib
+import mock
+import unittest
+
+
+class FakeWindowsError(Exception):
+    """WindowsError is available on Windows only."""
+    def __init__(self, errno):
+        self.errno = errno
+
+
+class WindowsSecurityUtilsTests(unittest.TestCase):
+
+    def setUp(self):
+        self._moves_mock = mock.MagicMock()
+
+        self._module_patcher = mock.patch.dict(
+            'sys.modules',
+            {'six.moves': self._moves_mock})
+
+        self._module_patcher.start()
+        self._winreg_mock = self._moves_mock.winreg
+
+        self.security = importlib.import_module(
+            "cloudbaseinit.utils.windows.security")
+        self.security.WindowsError = FakeWindowsError
+
+        self._security_utils = self.security.WindowsSecurityUtils()
+
+    def tearDown(self):
+        self._module_patcher.stop()
+
+    def test_set_uac_remote_restrictions(self):
+        fake_value = False
+        self._security_utils.set_uac_remote_restrictions(
+            enable=fake_value)
+
+        self._winreg_mock.SetValueEx.assert_called_once_with(
+            self._winreg_mock.CreateKey.return_value.__enter__(),
+            self._security_utils._LATFP_VALUE_NAME, 0,
+            self._winreg_mock.REG_DWORD, not fake_value)
+
+        self._winreg_mock.CreateKey.assert_called_once_with(
+            self._winreg_mock.HKEY_LOCAL_MACHINE,
+            self._security_utils._SYSTEM_POLICIES_KEY)
+
+    def _test_get_uac_remote_restrictions_win_error(self, ret_error=False):
+        fake_errno = 2
+        if ret_error:
+            fake_errno = 0
+
+        self._winreg_mock.OpenKey.side_effect = [
+            self.security.WindowsError(fake_errno)]
+        if ret_error:
+            self.assertRaises(self.security.WindowsError,
+                              self._security_utils.get_uac_remote_restrictions)
+        else:
+            response = self._security_utils.get_uac_remote_restrictions()
+            self.assertTrue(response)
+
+    def test_get_uac_remote_restrictions_win_error_ret_error(self):
+        self._test_get_uac_remote_restrictions_win_error(ret_error=True)
+
+    def test_get_uac_remote_restrictions_win_error(self):
+        self._test_get_uac_remote_restrictions_win_error(ret_error=False)
+
+    def test_get_uac_remote_restrictions_no_error(self):
+        key = mock.MagicMock()
+        fake_key_value = 0
+        key.__enter__.return_value = fake_key_value
+        fake_reg_type = mock.sentinel.fake_reg_type
+
+        self._winreg_mock.OpenKey.return_value = key
+        self._winreg_mock.QueryValueEx.return_value = (fake_key_value,
+                                                       fake_reg_type)
+
+        response = self._security_utils.get_uac_remote_restrictions()
+
+        self._winreg_mock.QueryValueEx.assert_called_once_with(
+            fake_key_value,
+            self._security_utils._LATFP_VALUE_NAME)
+
+        self._winreg_mock.OpenKey.assert_called_once_with(
+            self._winreg_mock.HKEY_LOCAL_MACHINE,
+            self._security_utils._SYSTEM_POLICIES_KEY)
+
+        self.assertTrue(bool(response))

cloudbaseinit/utils/windows/security.py

@@ -0,0 +1,41 @@
+# Copyright 2014 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 six.moves import winreg
+
+
+class WindowsSecurityUtils(object):
+    _SYSTEM_POLICIES_KEY = ("SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\"
+                            "Policies\\System")
+    _LATFP_VALUE_NAME = "LocalAccountTokenFilterPolicy"
+
+    # https://support.microsoft.com/kb/951016
+    def set_uac_remote_restrictions(self, enable=True):
+        with winreg.CreateKey(winreg.HKEY_LOCAL_MACHINE,
+                              self._SYSTEM_POLICIES_KEY) as key_name:
+            winreg.SetValueEx(key_name, self._LATFP_VALUE_NAME, 0,
+                              winreg.REG_DWORD, int(not enable))
+
+    def get_uac_remote_restrictions(self):
+        try:
+            with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE,
+                                self._SYSTEM_POLICIES_KEY) as key:
+                (value, regtype) = winreg.QueryValueEx(key,
+                                                       self._LATFP_VALUE_NAME)
+                return not bool(value)
+        except WindowsError as e:
+            if e.errno == 0x2:
+                return True
+            else:
+                raise