| # Copyright (c) 2012 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Base class for tests that need to update the policies enforced by Chrome. |
| |
| Subclasses can call SetUserPolicy (ChromeOS, Linux, Windows) and |
| SetDevicePolicy (ChromeOS only) with a dictionary of the policies to install. |
| |
| The current implementation depends on the platform. The implementations might |
| change in the future, but tests relying on the above calls will keep working. |
| """ |
| |
| # On ChromeOS, a mock DMServer is started and enterprise enrollment faked |
| # against it. The mock DMServer then serves user and device policy to Chrome. |
| # |
| # For this setup to work, the DNS, GAIA and TPM (if present) are mocked as well: |
| # |
| # * The mock DNS resolves all addresses to 127.0.0.1. This allows the mock GAIA |
| # to handle all login attempts. It also eliminates the impact of flaky network |
| # connections on tests. Beware though that no cloud services can be accessed |
| # due to this DNS redirect. |
| # |
| # * The mock GAIA permits login with arbitrary credentials and accepts any OAuth |
| # tokens sent to it for verification as valid. |
| # |
| # * When running on a real device, its TPM is disabled. If the TPM were enabled, |
| # enrollment could not be undone without a reboot. Disabling the TPM makes |
| # cryptohomed behave as if no TPM was present, allowing enrollment to be |
| # undone by removing the install attributes. |
| # |
| # To disable the TPM, 0 must be written to /sys/class/misc/tpm0/device/enabled. |
| # Since this file is not writeable, a tpmfs is mounted that shadows the file |
| # with a writeable copy. |
| |
| import json |
| import logging |
| import os |
| import subprocess |
| |
| import pyauto |
| |
| if pyauto.PyUITest.IsChromeOS(): |
| import sys |
| import warnings |
| |
| import pyauto_paths |
| |
| # Ignore deprecation warnings, they make our output more cluttered. |
| warnings.filterwarnings('ignore', category=DeprecationWarning) |
| |
| # Find the path to the pyproto and add it to sys.path. |
| # Prepend it so that google.protobuf is loaded from here. |
| for path in pyauto_paths.GetBuildDirs(): |
| p = os.path.join(path, 'pyproto') |
| if os.path.isdir(p): |
| sys.path = [p, os.path.join(p, 'chrome', 'browser', 'policy', |
| 'proto')] + sys.path |
| break |
| sys.path.append('/usr/local') # to import autotest libs. |
| |
| import dbus |
| import device_management_backend_pb2 as dm |
| import pyauto_utils |
| import string |
| import tempfile |
| import urllib |
| import urllib2 |
| import uuid |
| from autotest.cros import auth_server |
| from autotest.cros import constants |
| from autotest.cros import cros_ui |
| from autotest.cros import dns_server |
| elif pyauto.PyUITest.IsWin(): |
| import _winreg as winreg |
| elif pyauto.PyUITest.IsMac(): |
| import getpass |
| import plistlib |
| |
| # ASN.1 object identifier for PKCS#1/RSA. |
| PKCS1_RSA_OID = '\x2a\x86\x48\x86\xf7\x0d\x01\x01\x01' |
| |
| TPM_SYSFS_PATH = '/sys/class/misc/tpm0' |
| TPM_SYSFS_ENABLED_FILE = os.path.join(TPM_SYSFS_PATH, 'device/enabled') |
| |
| |
| class PolicyTestBase(pyauto.PyUITest): |
| """A base class for tests that need to set up and modify policies. |
| |
| Subclasses can use the methods SetUserPolicy (ChromeOS, Linux, Windows) and |
| SetDevicePolicy (ChromeOS only) to set the policies seen by Chrome. |
| """ |
| |
| if pyauto.PyUITest.IsChromeOS(): |
| # TODO(bartfab): Extend the C++ wrapper that starts the mock DMServer so |
| # that an owner can be passed in. Without this, the server will assume that |
| # the owner is user@example.com and for consistency, so must we. |
| owner = 'user@example.com' |
| # Subclasses may override these credentials to fake enrollment into another |
| # mode or use different device and machine IDs. |
| mode = 'enterprise' |
| device_id = string.upper(str(uuid.uuid4())) |
| machine_id = 'CROSTEST' |
| |
| _auth_server = None |
| _dns_server = None |
| |
| def ShouldAutoLogin(self): |
| return False |
| |
| @staticmethod |
| def _Call(command, check=False): |
| """Invokes a subprocess and optionally asserts the return value is zero.""" |
| with open(os.devnull, 'w') as devnull: |
| if check: |
| return subprocess.check_call(command.split(' '), stdout=devnull) |
| else: |
| return subprocess.call(command.split(' '), stdout=devnull) |
| |
| def _WriteFile(self, path, content): |
| """Writes content to path, creating any intermediary directories.""" |
| if not os.path.exists(os.path.dirname(path)): |
| os.makedirs(os.path.dirname(path)) |
| f = open(path, 'w') |
| f.write(content) |
| f.close() |
| |
| def _GetTestServerPoliciesFilePath(self): |
| """Returns the path of the cloud policy configuration file.""" |
| assert self.IsChromeOS() |
| return os.path.join(self._temp_data_dir, 'device_management') |
| |
| def _GetHttpURLForDeviceManagement(self): |
| """Returns the URL at which the TestServer is serving user policy.""" |
| assert self.IsChromeOS() |
| return self._http_server.GetURL('device_management').spec() |
| |
| def _RemoveIfExists(self, filename): |
| """Removes a file if it exists.""" |
| if os.path.exists(filename): |
| os.remove(filename) |
| |
| def _StartSessionManagerAndChrome(self): |
| """Starts the session manager and Chrome. |
| |
| Requires that the session manager be stopped already. |
| """ |
| # Ugly hack: session manager will not spawn Chrome if this file exists. That |
| # is usually a good thing (to keep the automation channel open), but in this |
| # case we really want to restart chrome. PyUITest.setUp() will be called |
| # after session manager and chrome have restarted, and will setup the |
| # automation channel. |
| restore_magic_file = False |
| if os.path.exists(constants.DISABLE_BROWSER_RESTART_MAGIC_FILE): |
| logging.debug('DISABLE_BROWSER_RESTART_MAGIC_FILE found. ' |
| 'Removing temporarily for the next restart.') |
| restore_magic_file = True |
| os.remove(constants.DISABLE_BROWSER_RESTART_MAGIC_FILE) |
| assert not os.path.exists(constants.DISABLE_BROWSER_RESTART_MAGIC_FILE) |
| |
| logging.debug('Starting session manager again') |
| cros_ui.start() |
| |
| # cros_ui.start() waits for the login prompt to be visible, so Chrome has |
| # already started once it returns. |
| if restore_magic_file: |
| open(constants.DISABLE_BROWSER_RESTART_MAGIC_FILE, 'w').close() |
| assert os.path.exists(constants.DISABLE_BROWSER_RESTART_MAGIC_FILE) |
| |
| def _WritePolicyOnChromeOS(self): |
| """Updates the mock DMServer's input file with current policy.""" |
| assert self.IsChromeOS() |
| policy_dict = { |
| 'google/chromeos/device': self._device_policy, |
| 'google/chromeos/user': { |
| 'mandatory': self._user_policy, |
| 'recommended': {}, |
| }, |
| 'managed_users': ['*'], |
| } |
| self._WriteFile(self._GetTestServerPoliciesFilePath(), |
| json.dumps(policy_dict, sort_keys=True, indent=2) + '\n') |
| |
| @staticmethod |
| def _IsCryptohomedReadyOnChromeOS(): |
| """Checks whether cryptohomed is running and ready to accept DBus calls.""" |
| assert pyauto.PyUITest.IsChromeOS() |
| try: |
| bus = dbus.SystemBus() |
| proxy = bus.get_object('org.chromium.Cryptohome', |
| '/org/chromium/Cryptohome') |
| dbus.Interface(proxy, 'org.chromium.CryptohomeInterface') |
| except dbus.DBusException: |
| return False |
| return True |
| |
| def _ClearInstallAttributesOnChromeOS(self): |
| """Resets the install attributes.""" |
| assert self.IsChromeOS() |
| self._RemoveIfExists('/home/.shadow/install_attributes.pb') |
| self._Call('restart cryptohomed', check=True) |
| assert self.WaitUntil(self._IsCryptohomedReadyOnChromeOS) |
| |
| def _DMPostRequest(self, request_type, request, headers): |
| """Posts a request to the mock DMServer.""" |
| assert self.IsChromeOS() |
| url = self._GetHttpURLForDeviceManagement() |
| url += '?' + urllib.urlencode({ |
| 'deviceid': self.device_id, |
| 'oauth_token': 'dummy_oauth_token_that_is_not_checked_anyway', |
| 'request': request_type, |
| 'devicetype': 2, |
| 'apptype': 'Chrome', |
| 'agent': 'Chrome', |
| }) |
| response = dm.DeviceManagementResponse() |
| response.ParseFromString(urllib2.urlopen(urllib2.Request( |
| url, request.SerializeToString(), headers)).read()) |
| return response |
| |
| def _DMRegisterDevice(self): |
| """Registers with the mock DMServer and returns the DMToken.""" |
| assert self.IsChromeOS() |
| dm_request = dm.DeviceManagementRequest() |
| request = dm_request.register_request |
| request.type = dm.DeviceRegisterRequest.DEVICE |
| request.machine_id = self.machine_id |
| dm_response = self._DMPostRequest('register', dm_request, {}) |
| return dm_response.register_response.device_management_token |
| |
| def _DMFetchPolicy(self, dm_token): |
| """Fetches device policy from the mock DMServer.""" |
| assert self.IsChromeOS() |
| dm_request = dm.DeviceManagementRequest() |
| policy_request = dm_request.policy_request |
| request = policy_request.request.add() |
| request.policy_type = 'google/chromeos/device' |
| request.signature_type = dm.PolicyFetchRequest.SHA1_RSA |
| headers = {'Authorization': 'GoogleDMToken token=' + dm_token} |
| dm_response = self._DMPostRequest('policy', dm_request, headers) |
| response = dm_response.policy_response.response[0] |
| assert response.policy_data |
| assert response.policy_data_signature |
| assert response.new_public_key |
| return response |
| |
| def ExtraChromeFlags(self): |
| """Sets up Chrome to use cloud policies on ChromeOS.""" |
| flags = pyauto.PyUITest.ExtraChromeFlags(self) |
| if self.IsChromeOS(): |
| while '--skip-oauth-login' in flags: |
| flags.remove('--skip-oauth-login') |
| url = self._GetHttpURLForDeviceManagement() |
| flags.append('--device-management-url=' + url) |
| flags.append('--disable-sync') |
| return flags |
| |
| def _SetUpWithSessionManagerStopped(self): |
| """Sets up the test environment after stopping the session manager.""" |
| assert self.IsChromeOS() |
| logging.debug('Stopping session manager') |
| cros_ui.stop(allow_fail=True) |
| |
| # Start mock GAIA server. |
| self._auth_server = auth_server.GoogleAuthServer() |
| self._auth_server.run() |
| |
| # Disable TPM if present. |
| if os.path.exists(TPM_SYSFS_PATH): |
| self._Call('mount -t tmpfs -o size=1k tmpfs %s' |
| % os.path.realpath(TPM_SYSFS_PATH), check=True) |
| self._WriteFile(TPM_SYSFS_ENABLED_FILE, '0') |
| |
| # Clear install attributes and restart cryptohomed to pick up the change. |
| self._ClearInstallAttributesOnChromeOS() |
| |
| # Set install attributes to mock enterprise enrollment. |
| bus = dbus.SystemBus() |
| proxy = bus.get_object('org.chromium.Cryptohome', |
| '/org/chromium/Cryptohome') |
| install_attributes = { |
| 'enterprise.device_id': self.device_id, |
| 'enterprise.domain': string.split(self.owner, '@')[-1], |
| 'enterprise.mode': self.mode, |
| 'enterprise.owned': 'true', |
| 'enterprise.user': self.owner |
| } |
| interface = dbus.Interface(proxy, 'org.chromium.CryptohomeInterface') |
| for name, value in install_attributes.iteritems(): |
| interface.InstallAttributesSet(name, '%s\0' % value) |
| interface.InstallAttributesFinalize() |
| |
| # Start mock DNS server that redirects all traffic to 127.0.0.1. |
| self._dns_server = dns_server.LocalDns() |
| self._dns_server.run() |
| |
| # Start mock DMServer. |
| source_dir = os.path.normpath(pyauto_paths.GetSourceDir()) |
| self._temp_data_dir = tempfile.mkdtemp(dir=source_dir) |
| logging.debug('TestServer input path: %s' % self._temp_data_dir) |
| relative_temp_data_dir = os.path.basename(self._temp_data_dir) |
| self._http_server = self.StartHTTPServer(relative_temp_data_dir) |
| |
| # Initialize the policy served. |
| self._device_policy = {} |
| self._user_policy = {} |
| self._WritePolicyOnChromeOS() |
| |
| # Register with mock DMServer and retrieve initial device policy blob. |
| dm_token = self._DMRegisterDevice() |
| policy = self._DMFetchPolicy(dm_token) |
| |
| # Write the initial device policy blob. |
| self._WriteFile(constants.OWNER_KEY_FILE, policy.new_public_key) |
| self._WriteFile(constants.SIGNED_POLICY_FILE, policy.SerializeToString()) |
| |
| # Remove any existing vaults. |
| self.RemoveAllCryptohomeVaultsOnChromeOS() |
| |
| # Restart session manager and Chrome. |
| self._StartSessionManagerAndChrome() |
| |
| def _tearDownWithSessionManagerStopped(self): |
| """Resets the test environment after stopping the session manager.""" |
| assert self.IsChromeOS() |
| logging.debug('Stopping session manager') |
| cros_ui.stop(allow_fail=True) |
| |
| # Stop mock GAIA server. |
| if self._auth_server: |
| self._auth_server.stop() |
| |
| # Reenable TPM if present. |
| if os.path.exists(TPM_SYSFS_PATH): |
| self._Call('umount %s' % os.path.realpath(TPM_SYSFS_PATH)) |
| |
| # Clear install attributes and restart cryptohomed to pick up the change. |
| self._ClearInstallAttributesOnChromeOS() |
| |
| # Stop mock DNS server. |
| if self._dns_server: |
| self._dns_server.stop() |
| |
| # Stop mock DMServer. |
| self.StopHTTPServer(self._http_server) |
| |
| # Clear the policy served. |
| pyauto_utils.RemovePath(self._temp_data_dir) |
| |
| # Remove the device policy blob. |
| self._RemoveIfExists(constants.OWNER_KEY_FILE) |
| self._RemoveIfExists(constants.SIGNED_POLICY_FILE) |
| |
| # Remove any existing vaults. |
| self.RemoveAllCryptohomeVaultsOnChromeOS() |
| |
| # Restart session manager and Chrome. |
| self._StartSessionManagerAndChrome() |
| |
| def setUp(self): |
| """Sets up the platform for policy testing. |
| |
| On ChromeOS, part of the setup involves restarting the session manager to |
| inject an initial device policy blob. |
| """ |
| if self.IsChromeOS(): |
| # Perform the remainder of the setup with the device manager stopped. |
| try: |
| self.WaitForSessionManagerRestart( |
| self._SetUpWithSessionManagerStopped) |
| except: |
| # Destroy the non re-entrant services. |
| if self._auth_server: |
| self._auth_server.stop() |
| if self._dns_server: |
| self._dns_server.stop() |
| raise |
| |
| pyauto.PyUITest.setUp(self) |
| self._branding = self.GetBrowserInfo()['properties']['branding'] |
| |
| def tearDown(self): |
| """Cleans up the policies and related files created in tests.""" |
| if self.IsChromeOS(): |
| # Perform the cleanup with the device manager stopped. |
| self.WaitForSessionManagerRestart(self._tearDownWithSessionManagerStopped) |
| else: |
| # On other platforms, there is only user policy to clear. |
| self.SetUserPolicy(refresh=False) |
| |
| pyauto.PyUITest.tearDown(self) |
| |
| def LoginWithTestAccount(self, account='prod_enterprise_test_user'): |
| """Convenience method for logging in with one of the test accounts.""" |
| assert self.IsChromeOS() |
| credentials = self.GetPrivateInfo()[account] |
| self.Login(credentials['username'], credentials['password']) |
| assert self.GetLoginInfo()['is_logged_in'] |
| |
| def _GetCurrentLoginScreenId(self): |
| return self.ExecuteJavascriptInOOBEWebUI( |
| """window.domAutomationController.send( |
| String(cr.ui.Oobe.getInstance().currentScreen.id)); |
| """) |
| |
| def _WaitForLoginScreenId(self, id): |
| self.assertTrue( |
| self.WaitUntil(function=self._GetCurrentLoginScreenId, |
| expect_retval=id), |
| msg='Expected login screen "%s" to be visible.' % id) |
| |
| def _CheckLoginFormLoading(self): |
| return self.ExecuteJavascriptInOOBEWebUI( |
| """window.domAutomationController.send( |
| cr.ui.Oobe.getInstance().currentScreen.loading); |
| """) |
| |
| def PrepareToWaitForLoginFormReload(self): |
| self.assertEqual('gaia-signin', |
| self._GetCurrentLoginScreenId(), |
| msg='Expected the login form to be visible.') |
| self.assertTrue( |
| self.WaitUntil(function=self._CheckLoginFormLoading, |
| expect_retval=False), |
| msg='Expected the login form to finish loading.') |
| # Set up a sentinel variable that is false now and will toggle to true when |
| # the login form starts reloading. |
| self.ExecuteJavascriptInOOBEWebUI( |
| """var screen = cr.ui.Oobe.getInstance().currentScreen; |
| if (!('reload_started' in screen)) { |
| screen.orig_loadAuthExtension_ = screen.loadAuthExtension_; |
| screen.loadAuthExtension_ = function(data) { |
| this.orig_loadAuthExtension_(data); |
| if (this.loading) |
| this.reload_started = true; |
| } |
| } |
| screen.reload_started = false; |
| window.domAutomationController.send(true);""") |
| |
| def _CheckLoginFormReloaded(self): |
| return self.ExecuteJavascriptInOOBEWebUI( |
| """window.domAutomationController.send( |
| cr.ui.Oobe.getInstance().currentScreen.reload_started && |
| !cr.ui.Oobe.getInstance().currentScreen.loading); |
| """) |
| |
| def WaitForLoginFormReload(self): |
| self.assertEqual('gaia-signin', |
| self._GetCurrentLoginScreenId(), |
| msg='Expected the login form to be visible.') |
| self.assertTrue( |
| self.WaitUntil(function=self._CheckLoginFormReloaded), |
| msg='Expected the login form to finish reloading.') |
| |
| def _SetUserPolicyChromeOS(self, user_policy=None): |
| """Writes the given user policy to the mock DMServer's input file.""" |
| self._user_policy = user_policy or {} |
| self._WritePolicyOnChromeOS() |
| |
| def _SetUserPolicyWin(self, user_policy=None): |
| """Writes the given user policy to the Windows registry.""" |
| def SetValueEx(key, sub_key, value): |
| if isinstance(value, int): |
| winreg.SetValueEx(key, sub_key, 0, winreg.REG_DWORD, int(value)) |
| elif isinstance(value, basestring): |
| winreg.SetValueEx(key, sub_key, 0, winreg.REG_SZ, value.encode('ascii')) |
| elif isinstance(value, list): |
| k = winreg.CreateKey(key, sub_key) |
| for index, v in list(enumerate(value)): |
| SetValueEx(k, str(index + 1), v) |
| winreg.CloseKey(k) |
| else: |
| raise TypeError('Unsupported data type: "%s"' % value) |
| |
| assert self.IsWin() |
| if self._branding == 'Google Chrome': |
| reg_base = r'SOFTWARE\Policies\Google\Chrome' |
| else: |
| reg_base = r'SOFTWARE\Policies\Chromium' |
| |
| if subprocess.call( |
| r'reg query HKEY_LOCAL_MACHINE\%s' % reg_base) == 0: |
| logging.debug(r'Removing %s' % reg_base) |
| subprocess.call(r'reg delete HKLM\%s /f' % reg_base) |
| |
| if user_policy is not None: |
| root_key = winreg.CreateKey(winreg.HKEY_LOCAL_MACHINE, reg_base) |
| for k, v in user_policy.iteritems(): |
| SetValueEx(root_key, k, v) |
| winreg.CloseKey(root_key) |
| |
| def _SetUserPolicyLinux(self, user_policy=None): |
| """Writes the given user policy to the JSON policy file read by Chrome.""" |
| assert self.IsLinux() |
| sudo_cmd_file = os.path.join(os.path.dirname(__file__), |
| 'policy_posix_util.py') |
| |
| if self._branding == 'Google Chrome': |
| policies_location_base = '/etc/opt/chrome' |
| else: |
| policies_location_base = '/etc/chromium' |
| |
| if os.path.exists(policies_location_base): |
| logging.debug('Removing directory %s' % policies_location_base) |
| subprocess.call(['suid-python', sudo_cmd_file, |
| 'remove_dir', policies_location_base]) |
| |
| if user_policy is not None: |
| self._WriteFile('/tmp/chrome.json', |
| json.dumps(user_policy, sort_keys=True, indent=2) + '\n') |
| |
| policies_location = '%s/policies/managed' % policies_location_base |
| subprocess.call(['suid-python', sudo_cmd_file, |
| 'setup_dir', policies_location]) |
| subprocess.call(['suid-python', sudo_cmd_file, |
| 'perm_dir', policies_location]) |
| # Copy chrome.json file to the managed directory |
| subprocess.call(['suid-python', sudo_cmd_file, |
| 'copy', '/tmp/chrome.json', policies_location]) |
| os.remove('/tmp/chrome.json') |
| |
| def _SetUserPolicyMac(self, user_policy=None): |
| """Writes the given user policy to the plist policy file read by Chrome.""" |
| assert self.IsMac() |
| sudo_cmd_file = os.path.join(os.path.dirname(__file__), |
| 'policy_posix_util.py') |
| |
| if self._branding == 'Google Chrome': |
| policies_file_base = 'com.google.Chrome.plist' |
| else: |
| policies_file_base = 'org.chromium.Chromium.plist' |
| |
| policies_location = os.path.join('/Library', 'Managed Preferences', |
| getpass.getuser()) |
| |
| if os.path.exists(policies_location): |
| logging.debug('Removing directory %s' % policies_location) |
| subprocess.call(['suid-python', sudo_cmd_file, |
| 'remove_dir', policies_location]) |
| |
| if user_policy is not None: |
| policies_tmp_file = os.path.join('/tmp', policies_file_base) |
| plistlib.writePlist(user_policy, policies_tmp_file) |
| subprocess.call(['suid-python', sudo_cmd_file, |
| 'setup_dir', policies_location]) |
| # Copy policy file to the managed directory |
| subprocess.call(['suid-python', sudo_cmd_file, |
| 'copy', policies_tmp_file, policies_location]) |
| os.remove(policies_tmp_file) |
| |
| def SetUserPolicy(self, user_policy=None, refresh=True): |
| """Sets the user policy provided as a dict. |
| |
| Args: |
| user_policy: The user policy to set. None clears it. |
| refresh: If True, Chrome will refresh and apply the new policy. |
| Requires Chrome to be alive for it. |
| """ |
| if self.IsChromeOS(): |
| self._SetUserPolicyChromeOS(user_policy=user_policy) |
| elif self.IsWin(): |
| self._SetUserPolicyWin(user_policy=user_policy) |
| elif self.IsLinux(): |
| self._SetUserPolicyLinux(user_policy=user_policy) |
| elif self.IsMac(): |
| self._SetUserPolicyMac(user_policy=user_policy) |
| else: |
| raise NotImplementedError('Not available on this platform.') |
| |
| if refresh: |
| self.RefreshPolicies() |
| |
| def SetDevicePolicy(self, device_policy=None, refresh=True): |
| """Sets the device policy provided as a dict. |
| |
| Args: |
| device_policy: The device policy to set. None clears it. |
| refresh: If True, Chrome will refresh and apply the new policy. |
| Requires Chrome to be alive for it. |
| """ |
| assert self.IsChromeOS() |
| self._device_policy = device_policy or {} |
| self._WritePolicyOnChromeOS() |
| if refresh: |
| self.RefreshPolicies() |
