# Copyright 2015 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import json
import logging
import os
from autotest_lib.client.bin import test
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib.cros import chrome
from autotest_lib.client.cros import cryptohome
from autotest_lib.client.cros import httpd
from autotest_lib.client.cros.enterprise import enterprise_fake_dmserver
CROSQA_FLAGS = [
'--gaia-url=https://gaiastaging.corp.google.com',
'--lso-url=https://gaiastaging.corp.google.com',
'--google-apis-url=https://www-googleapis-test.sandbox.google.com',
'--oauth2-client-id=236834563817.apps.googleusercontent.com',
'--oauth2-client-secret=RsKv5AwFKSzNgE0yjnurkPVI',
('--cloud-print-url='
'https://cloudprint-nightly-ps.sandbox.google.com/cloudprint'),
'--ignore-urlfetcher-cert-requests']
CROSALPHA_FLAGS = [
('--cloud-print-url='
'https://cloudprint-nightly-ps.sandbox.google.com/cloudprint'),
'--ignore-urlfetcher-cert-requests']
TESTDMS_FLAGS = [
'--ignore-urlfetcher-cert-requests',
'--disable-policy-key-verification']
FLAGS_DICT = {
'prod': [],
'crosman-qa': CROSQA_FLAGS,
'crosman-alpha': CROSALPHA_FLAGS,
'dm-test': TESTDMS_FLAGS,
'dm-fake': TESTDMS_FLAGS
}
DMS_URL_DICT = {
'prod': 'http://m.google.com/devicemanagement/data/api',
'crosman-qa':
'https://crosman-qa.sandbox.google.com/devicemanagement/data/api',
'crosman-alpha':
'https://crosman-alpha.sandbox.google.com/devicemanagement/data/api',
'dm-test': 'http://chromium-dm-test.appspot.com/d/%s',
'dm-fake': 'http://127.0.0.1:%d/'
}
DMSERVER = '--device-management-url=%s'
# Username and password for the fake dm server can be anything, since
# they are not used to authenticate against GAIA.
USERNAME = 'fake-user@managedchrome.com'
PASSWORD = 'fakepassword'
class EnterprisePolicyTest(test.test):
"""Base class for Enterprise Policy Tests."""
WEB_PORT = 8080
WEB_HOST = 'http://localhost:%d' % WEB_PORT
CHROME_SETTINGS_PAGE = 'chrome://settings'
CHROME_POLICY_PAGE = 'chrome://policy'
SETTING_LABEL = 0
SETTING_CHECKED = 1
SETTING_DISABLED = 2
def setup(self):
os.chdir(self.srcdir)
utils.make()
def initialize(self, **kwargs):
self._initialize_enterprise_policy_test(**kwargs)
def _initialize_enterprise_policy_test(
self, case, env='dm-fake', dms_name=None,
username=USERNAME, password=PASSWORD):
"""Initialize test parameters, fake DM Server, and Chrome flags.
@param case: String name of the test case to run.
@param env: String environment of DMS and Gaia servers.
@param username: String user name login credential.
@param password: String password login credential.
@param dms_name: String name of test DM Server.
"""
self.case = case
self.env = env
self.username = username
self.password = password
self.dms_name = dms_name
self.dms_is_fake = (env == 'dm-fake')
self._enforce_variable_restrictions()
# Initialize later variables to prevent error after an early failure.
self._web_server = None
self.cr = None
# Start AutoTest DM Server if using local fake server.
if self.dms_is_fake:
self.fake_dm_server = enterprise_fake_dmserver.FakeDMServer(
self.srcdir)
self.fake_dm_server.start(self.tmpdir, self.debugdir)
# Get enterprise directory of shared resources.
client_dir = os.path.dirname(os.path.dirname(self.bindir))
self.enterprise_dir = os.path.join(client_dir, 'cros/enterprise')
# Log the test context parameters.
logging.info('Test Context Parameters:')
logging.info(' Case: %r', self.case)
logging.info(' Environment: %r', self.env)
logging.info(' Username: %r', self.username)
logging.info(' Password: %r', self.password)
logging.info(' Test DMS Name: %r', self.dms_name)
def cleanup(self):
# Clean up AutoTest DM Server if using local fake server.
if self.dms_is_fake:
self.fake_dm_server.stop()
# Stop web server if it was started.
if self._web_server:
self._web_server.stop()
# Close Chrome instance if opened.
if self.cr:
self.cr.close()
def start_webserver(self):
"""Set up HTTP Server to serve pages from enterprise directory."""
self._web_server = httpd.HTTPListener(
self.WEB_PORT, docroot=self.enterprise_dir)
self._web_server.run()
def _enforce_variable_restrictions(self):
"""Validate class-level test context parameters.
@raises error.TestError if context parameter has an invalid value,
or a combination of parameters have incompatible values.
"""
# Verify |env| is a valid environment.
if self.env not in FLAGS_DICT:
raise error.TestError('Environment is invalid: %s' % self.env)
# Verify test |dms_name| is given iff |env| is 'dm-test'.
if self.env == 'dm-test' and not self.dms_name:
raise error.TestError('dms_name must be given when using '
'env=dm-test.')
if self.env != 'dm-test' and self.dms_name:
raise error.TestError('dms_name must not be given when not using '
'env=dm-test.')
def setup_case(self, policy_name, policy_value, mandatory_policies={},
suggested_policies={}, policy_name_is_suggested=False,
skip_policy_value_verification=False):
"""Set up and confirm the preconditions of a test case.
If the AutoTest fake DM Server is used, make a JSON policy blob
and upload it to the fake DM server.
Launch Chrome and sign in to Chrome OS. Examine the user's
cryptohome vault, to confirm user is signed in successfully.
Open the Policies page, and confirm that it shows the specified
|policy_name| and has the correct |policy_value|.
@param policy_name: Name of the policy under test.
@param policy_value: Expected value for the policy under test.
@param mandatory_policies: optional dict of mandatory policies
(not policy_name) in name -> value format.
@param suggested_policies: optional dict of suggested policies
(not policy_name) in name -> value format.
@param policy_name_is_suggested: True if policy_name a suggested policy.
@param skip_policy_value_verification: True if setup_case should not
verify that the correct policy value shows on policy page.
@raises error.TestError if cryptohome vault is not mounted for user.
@raises error.TestFail if |policy_name| and |policy_value| are not
shown on the Policies page.
"""
logging.info('Setting up case: (%s: %s)', policy_name, policy_value)
logging.info('Mandatory policies: %s', mandatory_policies)
logging.info('Suggested policies: %s', suggested_policies)
if self.dms_is_fake:
if policy_name_is_suggested:
suggested_policies[policy_name] = policy_value
else:
mandatory_policies[policy_name] = policy_value
self.fake_dm_server.setup_policy(self._make_json_blob(
mandatory_policies, suggested_policies))
self._launch_chrome_browser()
if not cryptohome.is_vault_mounted(user=self.username,
allow_fail=True):
raise error.TestError('Expected to find a mounted vault for %s.'
% self.username)
if not skip_policy_value_verification:
self.verify_policy_value(policy_name, policy_value)
def _make_json_blob(self, mandatory_policies, suggested_policies):
"""Create JSON policy blob from mandatory and suggested policies.
For the status of a policy to be shown as "Not set" on the
chrome://policy page, the policy dictionary must contain no NVP for
for that policy. Remove policy NVPs if value is None.
@param mandatory_policies: dict of mandatory policies -> values.
@param suggested_policies: dict of suggested policies -> values.
@returns: JSON policy blob to send to the fake DM server.
"""
# Remove "Not set" policies and json-ify dicts because the
# FakeDMServer expects "policy": "{value}" not "policy": {value}.
for policies_dict in [mandatory_policies, suggested_policies]:
policies_to_pop = []
for policy in policies_dict:
value = policies_dict[policy]
if value is None:
policies_to_pop.append(policy)
elif isinstance(value, dict):
policies_dict[policy] = encode_json_string(value)
elif isinstance(value, list):
if len(value) > 0 and isinstance(value[0], dict):
policies_dict[policy] = encode_json_string(value)
for policy in policies_to_pop:
policies_dict.pop(policy)
modes_dict = {}
if mandatory_policies:
modes_dict['mandatory'] = mandatory_policies
if suggested_policies:
modes_dict['suggested'] = suggested_policies
device_management_dict = {
'google/chromeos/user': modes_dict,
'managed_users': ['*'],
'policy_user': self.username,
'current_key_index': 0,
'invalidation_source': 16,
'invalidation_name': 'test_policy'
}
logging.info('Created policy blob: %s', device_management_dict)
return encode_json_string(device_management_dict)
def _get_policy_value_shown(self, policy_tab, policy_name):
"""Get the value shown for |policy_name| from the |policy_tab| page.
Return the policy value for the policy given by |policy_name|, from
from the chrome://policy page given by |policy_tab|.
CAVEAT: the policy page does not display proper JSON. For example, lists
are generally shown without the [ ] and cannot be distinguished from
strings. This function decodes what it can and returns the string it
found when in doubt.
@param policy_tab: Tab displaying the Policies page.
@param policy_name: The name of the policy.
@returns: The decoded value shown for the policy on the Policies page,
with the aforementioned caveat.
"""
row_values = policy_tab.EvaluateJavaScript('''
var section = document.getElementsByClassName(
"policy-table-section")[0];
var table = section.getElementsByTagName('table')[0];
rowValues = '';
for (var i = 1, row; row = table.rows[i]; i++) {
if (row.className !== 'expanded-value-container') {
var name_div = row.getElementsByClassName('name elide')[0];
var name = name_div.textContent;
if (name === '%s') {
var value_span = row.getElementsByClassName('value')[0];
var value = value_span.textContent;
var status_div = row.getElementsByClassName(
'status elide')[0];
var status = status_div.textContent;
rowValues = [name, value, status];
break;
}
}
}
rowValues;
''' % policy_name)
value_shown = row_values[1].encode('ascii', 'ignore')
status_shown = row_values[2].encode('ascii', 'ignore')
logging.debug('Policy %s row: %s', policy_name, row_values)
if status_shown == 'Not set.':
return None
return decode_json_string(value_shown)
def _get_policy_value_from_new_tab(self, policy_name):
"""Get the policy value for |policy_name| from the Policies page.
@param policy_name: string of policy name.
@returns: decoded value of the policy as shown on chrome://policy.
"""
values = self._get_policy_values_from_new_tab([policy_name])
return values[policy_name]
def _get_policy_values_from_new_tab(self, policy_names):
"""Get a given policy value by opening a new tab then closing it.
@param policy_names: list of strings of policy names.
@returns: dict of policy name mapped to decoded values of the policy as
shown on chrome://policy.
"""
values = {}
tab = self.navigate_to_url(self.CHROME_POLICY_PAGE)
for policy_name in policy_names:
values[policy_name] = self._get_policy_value_shown(tab, policy_name)
tab.Close()
return values
def verify_policy_value(self, policy_name, expected_value):
"""
Verify that the correct policy values shows in chrome://policy.
@param policy_name: the policy we are checking.
@param expected_value: the expected value for policy_name.
@raises error.TestError if value does not match expected.
"""
value_shown = self._get_policy_value_from_new_tab(policy_name)
logging.info('Value decoded from chrome://policy: %s', value_shown)
# If we expect a list and don't have a list, modify the value_shown.
if isinstance(expected_value, list):
if isinstance(value_shown, str):
if '{' in value_shown: # List of dicts.
value_shown = decode_json_string('[%s]' % value_shown)
elif ',' in value_shown: # List of strs.
value_shown = value_shown.split(',')
else: # List with one str.
value_shown = [value_shown]
elif not isinstance(value_shown, list): # List with one element.
value_shown = [value_shown]
if not expected_value == value_shown:
raise error.TestError('chrome://policy shows the incorrect value '
'for %s! Expected %s, got %s.' % (
policy_name, expected_value,
value_shown))
def _initialize_chrome_extra_flags(self):
"""
Initialize flags used to create Chrome instance.
@returns: list of extra Chrome flags.
"""
# Construct DM Server URL flags if not using production server.
env_flag_list = []
if self.env != 'prod':
if self.dms_is_fake:
# Use URL provided by the fake AutoTest DM server.
dmserver_str = (DMSERVER % self.fake_dm_server.server_url)
else:
# Use URL defined in the DMS URL dictionary.
dmserver_str = (DMSERVER % (DMS_URL_DICT[self.env]))
if self.env == 'dm-test':
dmserver_str = (dmserver_str % self.dms_name)
# Merge with other flags needed by non-prod enviornment.
env_flag_list = ([dmserver_str] + FLAGS_DICT[self.env])
return env_flag_list
def _launch_chrome_browser(self):
"""Launch Chrome browser and sign in."""
extra_flags = self._initialize_chrome_extra_flags()
logging.info('Chrome Browser Arguments:')
logging.info(' extra_browser_args: %s', extra_flags)
logging.info(' username: %s', self.username)
logging.info(' password: %s', self.password)
logging.info(' gaia_login: %s', not self.dms_is_fake)
self.cr = chrome.Chrome(extra_browser_args=extra_flags,
username=self.username,
password=self.password,
gaia_login=not self.dms_is_fake,
disable_gaia_services=False,
autotest_ext=True)
def navigate_to_url(self, url, tab=None):
"""Navigate tab to the specified |url|. Create new tab if none given.
@param url: URL of web page to load.
@param tab: browser tab to load (if any).
@returns: browser tab loaded with web page.
"""
logging.info('Navigating to URL: %r', url)
if not tab:
tab = self.cr.browser.tabs.New()
tab.Activate()
tab.Navigate(url, timeout=8)
tab.WaitForDocumentReadyStateToBeComplete()
return tab
def get_elements_from_page(self, tab, cmd):
"""Get collection of page elements that match the |cmd| filter.
@param tab: tab containing the page to be scraped.
@param cmd: JavaScript command to evaluate on the page.
@returns object containing elements on page that match the cmd.
@raises: TestFail if matching elements are not found on the page.
"""
try:
elements = tab.EvaluateJavaScript(cmd)
except Exception as err:
raise error.TestFail('Unable to find matching elements on '
'the test page: %s\n %r' %(tab.url, err))
return elements
def _get_settings_checkbox_properties(self, pref):
"""Get properties of the |pref| check box on the settings page.
@param pref: pref attribute value of the check box setting.
@returns: element properties of the check box setting.
"""
js_cmd_get_props = ('''
settings = document.getElementsByClassName(
'checkbox controlled-setting-with-label');
settingValues = '';
for (var i = 0, setting; setting = settings[i]; i++) {
var setting_label = setting.getElementsByTagName("label")[0];
var label_input = setting_label.getElementsByTagName("input")[0];
var input_pref = label_input.getAttribute("pref");
if (input_pref == '%s') {
settingValues = [setting_label.innerText.trim(),
label_input.checked, label_input.disabled];
break;
}
}
settingValues;
''' % pref)
tab = self.navigate_to_url(self.CHROME_SETTINGS_PAGE)
checkbox_props = self.get_elements_from_page(tab, js_cmd_get_props)
tab.Close()
return checkbox_props
def run_once(self):
"""The run_once() method is required by all AutoTest tests.
run_once() is defined herein to automatically determine which test
case in the test class to run. The test class must have a public
run_test_case() method defined. Note: The test class may override
run_once() if it determines which test case to run.
"""
logging.info('Running test case: %s', self.case)
self.run_test_case(self.case)
def encode_json_string(object_value):
"""Convert given value to JSON format string.
@param object_value: object to be converted.
@returns: string in JSON format.
"""
return json.dumps(object_value)
def decode_json_string(json_string):
"""Convert given JSON format string to an object.
If no object is found, return json_string instead. This is to allow
us to "decode" items off the policy page that aren't real JSON.
@param json_string: the JSON string to be decoded.
@returns: Python object represented by json_string or json_string.
"""
def _decode_list(json_list):
result = []
for value in json_list:
if isinstance(value, unicode):
value = value.encode('ascii')
if isinstance(value, list):
value = _decode_list(value)
if isinstance(value, dict):
value = _decode_dict(value)
result.append(value)
return result
def _decode_dict(json_dict):
result = {}
for key, value in json_dict.iteritems():
if isinstance(key, unicode):
key = key.encode('ascii')
if isinstance(value, unicode):
value = value.encode('ascii')
elif isinstance(value, list):
value = _decode_list(value)
result[key] = value
return result
try:
# Decode JSON turning all unicode strings into ascii.
# object_hook will get called on all dicts, so also handle lists.
result = json.loads(json_string, encoding='ascii',
object_hook=_decode_dict)
if isinstance(result, list):
result = _decode_list(result)
return result
except ValueError as e:
# Input not valid, e.g. '1, 2, "c"' instead of '[1, 2, "c"]'.
logging.warning('Could not unload: %s (%s)', json_string, e)
return json_string