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