1 # Copyright (c) 2012 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 import dbus, gobject, logging, os, random, re, shutil, string, sys, time 6 from dbus.mainloop.glib import DBusGMainLoop 7 8 import common, constants 9 from autotest_lib.client.bin import utils 10 from autotest_lib.client.common_lib import error 11 from autotest_lib.client.cros.cros_disks import DBusClient 12 13 CRYPTOHOME_CMD = '/usr/sbin/cryptohome' 14 GUEST_USER_NAME = '$guest' 15 UNAVAILABLE_ACTION = 'Unknown action or no action given.' 16 MOUNT_RETRY_COUNT = 20 17 TEMP_MOUNT_PATTERN = '/home/.shadow/%s/temporary_mount' 18 VAULT_PATH_PATTERN = '/home/.shadow/%s/vault' 19 20 DBUS_PROTOS_DEP = 'dbus_protos' 21 22 23 class ChromiumOSError(error.TestError): 24 """Generic error for ChromiumOS-specific exceptions.""" 25 pass 26 27 def __run_cmd(cmd): 28 return utils.system_output(cmd + ' 2>&1', retain_output=True, 29 ignore_status=True).strip() 30 31 def get_user_hash(user): 32 """Get the user hash for the given user.""" 33 return utils.system_output(['cryptohome', '--action=obfuscate_user', 34 '--user=%s' % user]) 35 36 37 def user_path(user): 38 """Get the user mount point for the given user.""" 39 return utils.system_output(['cryptohome-path', 'user', user]) 40 41 42 def system_path(user): 43 """Get the system mount point for the given user.""" 44 return utils.system_output(['cryptohome-path', 'system', user]) 45 46 47 def temporary_mount_path(user): 48 """Get the vault mount path used during crypto-migration for the user. 49 50 @param user: user the temporary mount should be for 51 """ 52 return TEMP_MOUNT_PATTERN % (get_user_hash(user)) 53 54 55 def vault_path(user): 56 """ Get the vault path for the given user. 57 58 @param user: The user who's vault path should be returned. 59 """ 60 return VAULT_PATH_PATTERN % (get_user_hash(user)) 61 62 63 def ensure_clean_cryptohome_for(user, password=None): 64 """Ensure a fresh cryptohome exists for user. 65 66 @param user: user who needs a shiny new cryptohome. 67 @param password: if unset, a random password will be used. 68 """ 69 if not password: 70 password = ''.join(random.sample(string.ascii_lowercase, 6)) 71 unmount_vault(user) 72 remove_vault(user) 73 mount_vault(user, password, create=True) 74 75 76 def get_tpm_status(): 77 """Get the TPM status. 78 79 Returns: 80 A TPM status dictionary, for example: 81 { 'Enabled': True, 82 'Owned': True, 83 'Being Owned': False, 84 'Ready': True, 85 'Password': '' 86 } 87 """ 88 out = __run_cmd(CRYPTOHOME_CMD + ' --action=tpm_status') 89 status = {} 90 for field in ['Enabled', 'Owned', 'Being Owned', 'Ready']: 91 match = re.search('TPM %s: (true|false)' % field, out) 92 if not match: 93 raise ChromiumOSError('Invalid TPM status: "%s".' % out) 94 status[field] = match.group(1) == 'true' 95 match = re.search('TPM Password: (\w*)', out) 96 status['Password'] = '' 97 if match: 98 status['Password'] = match.group(1) 99 return status 100 101 102 def get_tpm_more_status(): 103 """Get more of the TPM status. 104 105 Returns: 106 A TPM more status dictionary, for example: 107 { 'dictionary_attack_lockout_in_effect': False, 108 'attestation_prepared': False, 109 'boot_lockbox_finalized': False, 110 'enabled': True, 111 'owned': True, 112 'owner_password': '' 113 'dictionary_attack_counter': 0, 114 'dictionary_attack_lockout_seconds_remaining': 0, 115 'dictionary_attack_threshold': 10, 116 'attestation_enrolled': False, 117 'initialized': True, 118 'verified_boot_measured': False, 119 'install_lockbox_finalized': True 120 } 121 An empty dictionary is returned if the command is not supported. 122 """ 123 status = {} 124 out = __run_cmd(CRYPTOHOME_CMD + ' --action=tpm_more_status | grep :') 125 if out.startswith(UNAVAILABLE_ACTION): 126 # --action=tpm_more_status only exists >= 41. 127 logging.info('Method not supported!') 128 return status 129 for line in out.splitlines(): 130 items = line.strip().split(':') 131 if items[1].strip() == 'false': 132 value = False 133 elif items[1].strip() == 'true': 134 value = True 135 elif items[1].strip().isdigit(): 136 value = int(items[1].strip()) 137 else: 138 value = items[1].strip(' "') 139 status[items[0]] = value 140 return status 141 142 143 def get_fwmp(cleared_fwmp=False): 144 """Get the firmware management parameters. 145 146 Args: 147 cleared_fwmp: True if the space should not exist. 148 149 Returns: 150 The dictionary with the FWMP contents, for example: 151 { 'flags': 0xbb41, 152 'developer_key_hash': 153 "\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\ 154 000\000\000\000\000\000\000\000\000\000\000", 155 } 156 or a dictionary with the Error if the FWMP doesn't exist and 157 cleared_fwmp is True 158 { 'error': 'CRYPTOHOME_ERROR_FIRMWARE_MANAGEMENT_PARAMETERS_INVALID' } 159 160 Raises: 161 ChromiumOSError if any expected field is not found in the cryptohome 162 output. This would typically happen when FWMP state does not match 163 'clreared_fwmp' 164 """ 165 out = __run_cmd(CRYPTOHOME_CMD + 166 ' --action=get_firmware_management_parameters') 167 168 if cleared_fwmp: 169 fields = ['error'] 170 else: 171 fields = ['flags', 'developer_key_hash'] 172 173 status = {} 174 for field in fields: 175 match = re.search('%s: (\S+)\n' % field, out) 176 if not match: 177 raise ChromiumOSError('Invalid FWMP field %s: "%s".' % 178 (field, out)) 179 status[field] = match.group(1) 180 return status 181 182 183 def set_fwmp(flags, developer_key_hash=None): 184 """Set the firmware management parameter contents. 185 186 Args: 187 developer_key_hash: a string with the developer key hash 188 189 Raises: 190 ChromiumOSError cryptohome cannot set the FWMP contents 191 """ 192 cmd = (CRYPTOHOME_CMD + 193 ' --action=set_firmware_management_parameters ' 194 '--flags=' + flags) 195 if developer_key_hash: 196 cmd += ' --developer_key_hash=' + developer_key_hash 197 198 out = __run_cmd(cmd) 199 if 'SetFirmwareManagementParameters success' not in out: 200 raise ChromiumOSError('failed to set FWMP: %s' % out) 201 202 203 def is_tpm_lockout_in_effect(): 204 """Returns true if the TPM lockout is in effect; false otherwise.""" 205 status = get_tpm_more_status() 206 return status.get('dictionary_attack_lockout_in_effect', None) 207 208 209 def get_login_status(): 210 """Query the login status 211 212 Returns: 213 A login status dictionary containing: 214 { 'owner_user_exists': True|False, 215 'boot_lockbox_finalized': True|False 216 } 217 """ 218 out = __run_cmd(CRYPTOHOME_CMD + ' --action=get_login_status') 219 status = {} 220 for field in ['owner_user_exists', 'boot_lockbox_finalized']: 221 match = re.search('%s: (true|false)' % field, out) 222 if not match: 223 raise ChromiumOSError('Invalid login status: "%s".' % out) 224 status[field] = match.group(1) == 'true' 225 return status 226 227 228 def get_tpm_attestation_status(): 229 """Get the TPM attestation status. Works similar to get_tpm_status(). 230 """ 231 out = __run_cmd(CRYPTOHOME_CMD + ' --action=tpm_attestation_status') 232 status = {} 233 for field in ['Prepared', 'Enrolled']: 234 match = re.search('Attestation %s: (true|false)' % field, out) 235 if not match: 236 raise ChromiumOSError('Invalid attestation status: "%s".' % out) 237 status[field] = match.group(1) == 'true' 238 return status 239 240 241 def take_tpm_ownership(wait_for_ownership=True): 242 """Take TPM owernship. 243 244 Args: 245 wait_for_ownership: block until TPM is owned if true 246 """ 247 __run_cmd(CRYPTOHOME_CMD + ' --action=tpm_take_ownership') 248 if wait_for_ownership: 249 # Note that waiting for the 'Ready' flag is more correct than waiting 250 # for the 'Owned' flag, as the latter is set by cryptohomed before some 251 # of the ownership tasks are completed. 252 utils.poll_for_condition( 253 lambda: get_tpm_status()['Ready'], 254 timeout=300, 255 exception=error.TestError('Timeout waiting for TPM ownership')) 256 257 258 def verify_ek(): 259 """Verify the TPM endorsement key. 260 261 Returns true if EK is valid. 262 """ 263 cmd = CRYPTOHOME_CMD + ' --action=tpm_verify_ek' 264 return (utils.system(cmd, ignore_status=True) == 0) 265 266 267 def remove_vault(user): 268 """Remove the given user's vault from the shadow directory.""" 269 logging.debug('user is %s', user) 270 user_hash = get_user_hash(user) 271 logging.debug('Removing vault for user %s with hash %s', user, user_hash) 272 cmd = CRYPTOHOME_CMD + ' --action=remove --force --user=%s' % user 273 __run_cmd(cmd) 274 # Ensure that the vault does not exist. 275 if os.path.exists(os.path.join(constants.SHADOW_ROOT, user_hash)): 276 raise ChromiumOSError('Cryptohome could not remove the user\'s vault.') 277 278 279 def remove_all_vaults(): 280 """Remove any existing vaults from the shadow directory. 281 282 This function must be run with root privileges. 283 """ 284 for item in os.listdir(constants.SHADOW_ROOT): 285 abs_item = os.path.join(constants.SHADOW_ROOT, item) 286 if os.path.isdir(os.path.join(abs_item, 'vault')): 287 logging.debug('Removing vault for user with hash %s', item) 288 shutil.rmtree(abs_item) 289 290 291 def mount_vault(user, password, create=False, key_label=None): 292 """Mount the given user's vault. Mounts should be created by calling this 293 function with create=True, and can be used afterwards with create=False. 294 Only try to mount existing vaults created with this function. 295 296 """ 297 args = [CRYPTOHOME_CMD, '--action=mount_ex', '--user=%s' % user, 298 '--password=%s' % password, '--async'] 299 if create: 300 args += ['--create'] 301 if key_label is None: 302 key_label = 'bar' 303 if key_label is not None: 304 args += ['--key_label=%s' % key_label] 305 logging.info(__run_cmd(' '.join(args))) 306 # Ensure that the vault exists in the shadow directory. 307 user_hash = get_user_hash(user) 308 if not os.path.exists(os.path.join(constants.SHADOW_ROOT, user_hash)): 309 retry = 0 310 mounted = False 311 while retry < MOUNT_RETRY_COUNT and not mounted: 312 time.sleep(1) 313 logging.info("Retry " + str(retry + 1)) 314 __run_cmd(' '.join(args)) 315 # TODO: Remove this additional call to get_user_hash(user) when 316 # crbug.com/690994 is fixed 317 user_hash = get_user_hash(user) 318 if os.path.exists(os.path.join(constants.SHADOW_ROOT, user_hash)): 319 mounted = True 320 retry += 1 321 if not mounted: 322 raise ChromiumOSError('Cryptohome vault not found after mount.') 323 # Ensure that the vault is mounted. 324 if not is_permanent_vault_mounted(user=user, allow_fail=True): 325 raise ChromiumOSError('Cryptohome created a vault but did not mount.') 326 327 328 def mount_guest(): 329 """Mount the guest vault.""" 330 args = [CRYPTOHOME_CMD, '--action=mount_guest_ex'] 331 logging.info(__run_cmd(' '.join(args))) 332 # Ensure that the guest vault is mounted. 333 if not is_guest_vault_mounted(allow_fail=True): 334 raise ChromiumOSError('Cryptohome did not mount guest vault.') 335 336 337 def test_auth(user, password): 338 cmd = [CRYPTOHOME_CMD, '--action=check_key_ex', '--user=%s' % user, 339 '--password=%s' % password, '--async'] 340 out = __run_cmd(' '.join(cmd)) 341 logging.info(out) 342 return 'Key authenticated.' in out 343 344 345 def add_le_key(user, password, new_password, new_key_label): 346 args = [CRYPTOHOME_CMD, '--action=add_key_ex', '--key_policy=le', 347 '--user=%s' % user, '--password=%s' % password, 348 '--new_key_label=%s' % new_key_label, 349 '--new_password=%s' % new_password] 350 logging.info(__run_cmd(' '.join(args))) 351 352 353 def remove_key(user, password, remove_key_label): 354 args = [CRYPTOHOME_CMD, '--action=remove_key_ex', '--user=%s' % user, 355 '--password=%s' % password, 356 '--remove_key_label=%s' % remove_key_label] 357 logging.info(__run_cmd(' '.join(args))) 358 359 360 def get_supported_key_policies(): 361 args = [CRYPTOHOME_CMD, '--action=get_supported_key_policies'] 362 out = __run_cmd(' '.join(args)) 363 logging.info(out) 364 policies = {} 365 for line in out.splitlines(): 366 match = re.search(' ([^:]+): (true|false)', line) 367 if match: 368 policies[match.group(1)] = match.group(2) == 'true' 369 return policies 370 371 372 def unmount_vault(user=None): 373 """Unmount the given user's vault. 374 375 Once unmounting for a specific user is supported, the user parameter will 376 name the target user. See crosbug.com/20778. 377 """ 378 __run_cmd(CRYPTOHOME_CMD + ' --action=unmount') 379 # Ensure that the vault is not mounted. 380 if user is not None and is_vault_mounted(user, allow_fail=True): 381 raise ChromiumOSError('Cryptohome did not unmount the user.') 382 383 384 def __get_mount_info(mount_point, allow_fail=False): 385 """Get information about the active mount at a given mount point.""" 386 cryptohomed_path = '/proc/$(pgrep cryptohomed)/mounts' 387 try: 388 logging.debug("Active cryptohome mounts:\n" + 389 utils.system_output('cat %s' % cryptohomed_path)) 390 mount_line = utils.system_output( 391 'grep %s %s' % (mount_point, cryptohomed_path), 392 ignore_status=allow_fail) 393 except Exception as e: 394 logging.error(e) 395 raise ChromiumOSError('Could not get info about cryptohome vault ' 396 'through %s. See logs for complete mount-point.' 397 % os.path.dirname(str(mount_point))) 398 return mount_line.split() 399 400 401 def __get_user_mount_info(user, allow_fail=False): 402 """Get information about the active mounts for a given user. 403 404 Returns the active mounts at the user's user and system mount points. If no 405 user is given, the active mount at the shared mount point is returned 406 (regular users have a bind-mount at this mount point for backwards 407 compatibility; the guest user has a mount at this mount point only). 408 """ 409 return [__get_mount_info(mount_point=user_path(user), 410 allow_fail=allow_fail), 411 __get_mount_info(mount_point=system_path(user), 412 allow_fail=allow_fail)] 413 414 def is_vault_mounted(user, regexes=None, allow_fail=False): 415 """Check whether a vault is mounted for the given user. 416 417 user: If no user is given, the shared mount point is checked, determining 418 whether a vault is mounted for any user. 419 regexes: dictionary of regexes to matches against the mount information. 420 The mount filesystem for the user's user and system mounts point must 421 match one of the keys. 422 The mount source point must match the selected device regex. 423 424 In addition, if mounted over ext4, we check the directory is encrypted. 425 """ 426 if regexes is None: 427 regexes = { 428 constants.CRYPTOHOME_FS_REGEX_ANY : 429 constants.CRYPTOHOME_DEV_REGEX_ANY 430 } 431 user_mount_info = __get_user_mount_info(user=user, allow_fail=allow_fail) 432 for mount_info in user_mount_info: 433 # Look at each /proc/../mount lines that match mount point for a given 434 # user user/system mount (/home/user/.... /home/root/...) 435 436 # We should have at least 3 arguments (source, mount, type of mount) 437 if len(mount_info) < 3: 438 return False 439 440 device_regex = None 441 for fs_regex in regexes.keys(): 442 if re.match(fs_regex, mount_info[2]): 443 device_regex = regexes[fs_regex] 444 break 445 446 if not device_regex: 447 # The thrid argument in not the expectd mount point type. 448 return False 449 450 # Check if the mount source match the device regex: it can be loose, 451 # (anything) or stricter if we expect guest filesystem. 452 if not re.match(device_regex, mount_info[0]): 453 return False 454 455 if (re.match(constants.CRYPTOHOME_FS_REGEX_EXT4, mount_info[2]) 456 and not(re.match(constants.CRYPTOHOME_DEV_REGEX_LOOP_DEVICE, 457 mount_info[0]))): 458 # Ephemeral cryptohome uses ext4 mount from a loop device, 459 # otherwise it should be ext4 crypto. Check there is an encryption 460 # key for that directory. 461 find_key_cmd_list = ['e4crypt get_policy %s' % (mount_info[1]), 462 'cut -d \' \' -f 2'] 463 key = __run_cmd(' | ' .join(find_key_cmd_list)) 464 cmd_list = ['keyctl show @s', 465 'grep %s' % (key), 466 'wc -l'] 467 out = __run_cmd(' | '.join(cmd_list)) 468 if int(out) != 1: 469 return False 470 return True 471 472 473 def is_guest_vault_mounted(allow_fail=False): 474 """Check whether a vault is mounted for the guest user. 475 It should be a mount of an ext4 partition on a loop device 476 or be backed by tmpfs. 477 """ 478 return is_vault_mounted( 479 user=GUEST_USER_NAME, 480 regexes={ 481 # Remove tmpfs support when it becomes unnecessary as all guest 482 # modes will use ext4 on a loop device. 483 constants.CRYPTOHOME_FS_REGEX_EXT4 : 484 constants.CRYPTOHOME_DEV_REGEX_LOOP_DEVICE, 485 constants.CRYPTOHOME_FS_REGEX_TMPFS : 486 constants.CRYPTOHOME_DEV_REGEX_GUEST, 487 }, 488 allow_fail=allow_fail) 489 490 def is_permanent_vault_mounted(user, allow_fail=False): 491 """Check if user is mounted over ecryptfs or ext4 crypto. """ 492 return is_vault_mounted( 493 user=user, 494 regexes={ 495 constants.CRYPTOHOME_FS_REGEX_ECRYPTFS : 496 constants.CRYPTOHOME_DEV_REGEX_REGULAR_USER_SHADOW, 497 constants.CRYPTOHOME_FS_REGEX_EXT4 : 498 constants.CRYPTOHOME_DEV_REGEX_REGULAR_USER_DEVICE, 499 }, 500 allow_fail=allow_fail) 501 502 def get_mounted_vault_path(user, allow_fail=False): 503 """Get the path where the decrypted data for the user is located.""" 504 return os.path.join(constants.SHADOW_ROOT, get_user_hash(user), 'mount') 505 506 507 def canonicalize(credential): 508 """Perform basic canonicalization of |email_address|. 509 510 Perform basic canonicalization of |email_address|, taking into account that 511 gmail does not consider '.' or caps inside a username to matter. It also 512 ignores everything after a '+'. For example, 513 c.masone+abc (at] gmail.com == cMaSone (at] gmail.com, per 514 http://mail.google.com/support/bin/answer.py?hl=en&ctx=mail&answer=10313 515 """ 516 if not credential: 517 return None 518 519 parts = credential.split('@') 520 if len(parts) != 2: 521 raise error.TestError('Malformed email: ' + credential) 522 523 (name, domain) = parts 524 name = name.partition('+')[0] 525 if (domain == constants.SPECIAL_CASE_DOMAIN): 526 name = name.replace('.', '') 527 return '@'.join([name, domain]).lower() 528 529 530 def crash_cryptohomed(): 531 # Try to kill cryptohomed so we get something to work with. 532 pid = __run_cmd('pgrep cryptohomed') 533 try: 534 pid = int(pid) 535 except ValueError, e: # empty or invalid string 536 raise error.TestError('Cryptohomed was not running') 537 utils.system('kill -ABRT %d' % pid) 538 # CONT just in case cryptohomed had a spurious STOP. 539 utils.system('kill -CONT %d' % pid) 540 utils.poll_for_condition( 541 lambda: utils.system('ps -p %d' % pid, 542 ignore_status=True) != 0, 543 timeout=180, 544 exception=error.TestError( 545 'Timeout waiting for cryptohomed to coredump')) 546 547 548 def create_ecryptfs_homedir(user, password): 549 """Creates a new home directory as ecryptfs. 550 551 If a home directory for the user exists already, it will be removed. 552 The resulting home directory will be mounted. 553 554 @param user: Username to create the home directory for. 555 @param password: Password to use when creating the home directory. 556 """ 557 unmount_vault(user) 558 remove_vault(user) 559 args = [ 560 CRYPTOHOME_CMD, 561 '--action=mount_ex', 562 '--user=%s' % user, 563 '--password=%s' % password, 564 '--key_label=foo', 565 '--ecryptfs', 566 '--create'] 567 logging.info(__run_cmd(' '.join(args))) 568 if not is_vault_mounted(user, regexes={ 569 constants.CRYPTOHOME_FS_REGEX_ECRYPTFS : 570 constants.CRYPTOHOME_DEV_REGEX_REGULAR_USER_SHADOW 571 }, allow_fail=True): 572 raise ChromiumOSError('Ecryptfs home could not be created') 573 574 575 def do_dircrypto_migration(user, password, timeout=600): 576 """Start dircrypto migration for the user. 577 578 @param user: The user to migrate. 579 @param password: The password used to mount the users vault 580 @param timeout: How long in seconds to wait for the migration to finish 581 before failing. 582 """ 583 unmount_vault(user) 584 args = [ 585 CRYPTOHOME_CMD, 586 '--action=mount_ex', 587 '--to_migrate_from_ecryptfs', 588 '--user=%s' % user, 589 '--password=%s' % password] 590 logging.info(__run_cmd(' '.join(args))) 591 if not __get_mount_info(temporary_mount_path(user), allow_fail=True): 592 raise ChromiumOSError('Failed to mount home for migration') 593 args = [CRYPTOHOME_CMD, '--action=migrate_to_dircrypto', '--user=%s' % user] 594 logging.info(__run_cmd(' '.join(args))) 595 utils.poll_for_condition( 596 lambda: not __get_mount_info( 597 temporary_mount_path(user), allow_fail=True), 598 timeout=timeout, 599 exception=error.TestError( 600 'Timeout waiting for dircrypto migration to finish')) 601 602 603 def change_password(user, password, new_password): 604 args = [ 605 CRYPTOHOME_CMD, 606 '--action=migrate_key_ex', 607 '--user=%s' % user, 608 '--old_password=%s' % password, 609 '--password=%s' % new_password] 610 out = __run_cmd(' '.join(args)) 611 logging.info(out) 612 if 'Key migration succeeded.' not in out: 613 raise ChromiumOSError('Key migration failed.') 614 615 616 class CryptohomeProxy(DBusClient): 617 """A DBus proxy client for testing the Cryptohome DBus server. 618 """ 619 CRYPTOHOME_BUS_NAME = 'org.chromium.Cryptohome' 620 CRYPTOHOME_OBJECT_PATH = '/org/chromium/Cryptohome' 621 CRYPTOHOME_INTERFACE = 'org.chromium.CryptohomeInterface' 622 ASYNC_CALL_STATUS_SIGNAL = 'AsyncCallStatus' 623 ASYNC_CALL_STATUS_SIGNAL_ARGUMENTS = ( 624 'async_id', 'return_status', 'return_code' 625 ) 626 DBUS_PROPERTIES_INTERFACE = 'org.freedesktop.DBus.Properties' 627 628 # Default timeout in seconds for the D-Bus connection. 629 DEFAULT_DBUS_TIMEOUT = 30 630 631 def __init__(self, bus_loop=None, autodir=None, job=None, 632 timeout=DEFAULT_DBUS_TIMEOUT): 633 if autodir and job: 634 # Install D-Bus protos necessary for some methods. 635 dep_dir = os.path.join(autodir, 'deps', DBUS_PROTOS_DEP) 636 job.install_pkg(DBUS_PROTOS_DEP, 'dep', dep_dir) 637 sys.path.append(dep_dir) 638 639 # Set up D-Bus main loop and interface. 640 self.main_loop = gobject.MainLoop() 641 if bus_loop is None: 642 bus_loop = DBusGMainLoop(set_as_default=True) 643 self.bus = dbus.SystemBus(mainloop=bus_loop) 644 super(CryptohomeProxy, self).__init__(self.main_loop, self.bus, 645 self.CRYPTOHOME_BUS_NAME, 646 self.CRYPTOHOME_OBJECT_PATH, 647 timeout) 648 self.iface = dbus.Interface(self.proxy_object, 649 self.CRYPTOHOME_INTERFACE) 650 self.properties = dbus.Interface(self.proxy_object, 651 self.DBUS_PROPERTIES_INTERFACE) 652 self.handle_signal(self.CRYPTOHOME_INTERFACE, 653 self.ASYNC_CALL_STATUS_SIGNAL, 654 self.ASYNC_CALL_STATUS_SIGNAL_ARGUMENTS) 655 656 657 # Wrap all proxied calls to catch cryptohomed failures. 658 def __call(self, method, *args): 659 try: 660 return method(*args, timeout=180) 661 except dbus.exceptions.DBusException, e: 662 if e.get_dbus_name() == 'org.freedesktop.DBus.Error.NoReply': 663 logging.error('Cryptohome is not responding. Sending ABRT') 664 crash_cryptohomed() 665 raise ChromiumOSError('cryptohomed aborted. Check crashes!') 666 raise e 667 668 669 def __wait_for_specific_signal(self, signal, data): 670 """Wait for the |signal| with matching |data| 671 Returns the resulting dict on success or {} on error. 672 """ 673 # Do not bubble up the timeout here, just return {}. 674 result = {} 675 try: 676 result = self.wait_for_signal(signal) 677 except utils.TimeoutError: 678 return {} 679 for k in data.keys(): 680 if not result.has_key(k) or result[k] != data[k]: 681 return {} 682 return result 683 684 685 # Perform a data-less async call. 686 # TODO(wad) Add __async_data_call. 687 def __async_call(self, method, *args): 688 # Clear out any superfluous async call signals. 689 self.clear_signal_content(self.ASYNC_CALL_STATUS_SIGNAL) 690 out = self.__call(method, *args) 691 logging.debug('Issued call ' + str(method) + 692 ' with async_id ' + str(out)) 693 result = {} 694 try: 695 # __wait_for_specific_signal has a 10s timeout 696 result = utils.poll_for_condition( 697 lambda: self.__wait_for_specific_signal( 698 self.ASYNC_CALL_STATUS_SIGNAL, {'async_id' : out}), 699 timeout=180, 700 desc='matching %s signal' % self.ASYNC_CALL_STATUS_SIGNAL) 701 except utils.TimeoutError, e: 702 logging.error('Cryptohome timed out. Sending ABRT.') 703 crash_cryptohomed() 704 raise ChromiumOSError('cryptohomed aborted. Check crashes!') 705 return result 706 707 708 def mount(self, user, password, create=False, async=True, key_label='bar'): 709 """Mounts a cryptohome. 710 711 Returns True if the mount succeeds or False otherwise. 712 """ 713 import rpc_pb2 714 715 acc = rpc_pb2.AccountIdentifier() 716 acc.account_id = user 717 718 auth = rpc_pb2.AuthorizationRequest() 719 auth.key.secret = password 720 auth.key.data.label = key_label 721 722 mount_req = rpc_pb2.MountRequest() 723 if create: 724 mount_req.create.copy_authorization_key = True 725 726 out = self.__call(self.iface.MountEx, acc.SerializeToString(), 727 auth.SerializeToString(), mount_req.SerializeToString()) 728 parsed_out = rpc_pb2.BaseReply() 729 parsed_out.ParseFromString(''.join(map(chr, out))) 730 return parsed_out.error == rpc_pb2.CRYPTOHOME_ERROR_NOT_SET 731 732 733 def unmount(self, user): 734 """Unmounts a cryptohome. 735 736 Returns True if the unmount suceeds or false otherwise. 737 """ 738 import rpc_pb2 739 740 req = rpc_pb2.UnmountRequest() 741 742 out = self.__call(self.iface.UnmountEx, req.SerializeToString()) 743 parsed_out = rpc_pb2.BaseReply() 744 parsed_out.ParseFromString(''.join(map(chr, out))) 745 return parsed_out.error == rpc_pb2.CRYPTOHOME_ERROR_NOT_SET 746 747 748 def is_mounted(self, user): 749 """Tests whether a user's cryptohome is mounted.""" 750 return (utils.is_mountpoint(user_path(user)) 751 and utils.is_mountpoint(system_path(user))) 752 753 754 def require_mounted(self, user): 755 """Raises a test failure if a user's cryptohome is not mounted.""" 756 utils.require_mountpoint(user_path(user)) 757 utils.require_mountpoint(system_path(user)) 758 759 760 def remove(self, user, async=True): 761 """Removes a users crypothome. 762 763 Returns True if the operation succeeds or False otherwise. 764 """ 765 import rpc_pb2 766 767 acc = rpc_pb2.AccountIdentifier() 768 acc.account_id = user 769 770 out = self.__call(self.iface.RemoveEx, acc.SerializeToString()) 771 parsed_out = rpc_pb2.BaseReply() 772 parsed_out.ParseFromString(''.join(map(chr, out))) 773 return parsed_out.error == rpc_pb2.CRYPTOHOME_ERROR_NOT_SET 774 775 776 def ensure_clean_cryptohome_for(self, user, password=None): 777 """Ensure a fresh cryptohome exists for user. 778 779 @param user: user who needs a shiny new cryptohome. 780 @param password: if unset, a random password will be used. 781 """ 782 if not password: 783 password = ''.join(random.sample(string.ascii_lowercase, 6)) 784 self.remove(user) 785 self.mount(user, password, create=True) 786 787 def lock_install_attributes(self, attrs): 788 """Set and lock install attributes for the device. 789 790 @param attrs: dict of install attributes. 791 """ 792 take_tpm_ownership() 793 self.wait_for_install_attributes_ready() 794 for key, value in attrs.items(): 795 if not self.__call(self.iface.InstallAttributesSet, key, 796 dbus.ByteArray(value + '\0')): 797 return False 798 return self.__call(self.iface.InstallAttributesFinalize) 799 800 def wait_for_install_attributes_ready(self): 801 """Wait until install attributes are ready. 802 """ 803 utils.poll_for_condition( 804 lambda: self.__call(self.iface.InstallAttributesIsReady), 805 timeout=300, 806 exception=error.TestError( 807 'Timeout waiting for install attributes are ready')) 808
