1 # Copyright 2017 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 logging 6 import os 7 import pyudev 8 import re 9 import select 10 import struct 11 import subprocess 12 import threading 13 import time 14 from autotest_lib.client.common_lib import error 15 16 17 JAIL_CONTROL_PATH = '/dev/jail-control' 18 JAIL_REQUEST_PATH = '/dev/jail-request' 19 20 # From linux/device_jail.h. 21 REQUEST_ALLOW = 0 22 REQUEST_ALLOW_WITH_LOCKDOWN = 1 23 REQUEST_ALLOW_WITH_DETACH = 2 24 REQUEST_DENY = 3 25 26 27 class OSFile: 28 """Simple context manager for file descriptors.""" 29 def __init__(self, path, flag): 30 self._fd = os.open(path, flag) 31 32 def close(self): 33 os.close(self._fd) 34 35 def __enter__(self): 36 """Returns the fd so it can be used in with-blocks.""" 37 return self._fd 38 39 def __exit__(self, exc_type, exc_val, traceback): 40 self.close() 41 42 43 class ConcurrentFunc: 44 """Simple context manager that starts and joins a thread.""" 45 def __init__(self, target_func, timeout_func): 46 self._thread = threading.Thread(target=target_func) 47 self._timeout_func = timeout_func 48 self._target_name = target_func.__name__ 49 50 def __enter__(self): 51 self._thread.start() 52 53 def __exit__(self, exc_type, exc_val, traceback): 54 self._thread.join(self._timeout_func()) 55 if self._thread.is_alive() and not exc_val: 56 raise error.TestError('Function %s timed out' % self._target_name) 57 58 59 class JailDevice: 60 TIMEOUT_SEC = 3 61 PATH_MAX = 4096 62 63 def __init__(self, path_to_jail): 64 self._path_to_jail = path_to_jail 65 66 67 def __enter__(self): 68 """ 69 Creates a jail device for the device located at self._path_to_jail. 70 If the jail already exists, don't take ownership of it. 71 """ 72 try: 73 output = subprocess.check_output( 74 ['device_jail_utility', 75 '--add={0}'.format(self._path_to_jail)], 76 stderr=subprocess.STDOUT) 77 78 match = re.search('created jail at (.*)', output) 79 if match: 80 self._path = match.group(1) 81 self._owns_device = True 82 return self 83 84 match = re.search('jail already exists at (.*)', output) 85 if match: 86 self._path = match.group(1) 87 self._owns_device = False 88 return self 89 90 raise error.TestError('Failed to create device jail') 91 except subprocess.CalledProcessError as e: 92 raise error.TestError('Failed to call device_jail_utility') 93 94 95 def expect_open(self, verdict): 96 """ 97 Tries to open the jail device. This method mocks out the 98 device_jail request server which is normally run by permission_broker. 99 This allows us to set the verdict we want to test. Since the open 100 call will block until we return the verdict, we have to use a 101 separate thread to perform the open call, as well. 102 """ 103 # Python 2 does not support "nonlocal" so this closure can't 104 # set the values of identifiers it closes over unless they 105 # are in global scope. Work around this by using a list and 106 # value-mutation. 107 dev_file_wrapper = [None] 108 def open_device(): 109 try: 110 dev_file_wrapper[0] = OSFile(self._path, os.O_RDWR) 111 except OSError as e: 112 # We don't throw an error because this might be intentional, 113 # such as when the verdict is REQUEST_DENY. 114 logging.info("Failed to open jail device: %s", e.strerror) 115 116 # timeout_sec should be used for the timeouts below. 117 # This ensures we don't spend much longer than TIMEOUT_SEC in 118 # this method. 119 deadline = time.time() + self.TIMEOUT_SEC 120 def timeout_sec(): 121 return max(deadline - time.time(), 0.01) 122 123 # We have to use FDs because polling works with FDs and 124 # buffering is silly. 125 try: 126 req_f = OSFile(JAIL_REQUEST_PATH, os.O_RDWR) 127 except OSError as e: 128 raise error.TestError( 129 'Failed to open request device: %s' % e.strerror) 130 131 with req_f as req_fd: 132 poll_obj = select.poll() 133 poll_obj.register(req_fd, select.POLLIN) 134 135 # Starting open_device should ensure we have a request waiting 136 # on the request device. 137 with ConcurrentFunc(open_device, timeout_sec): 138 ready_fds = poll_obj.poll(timeout_sec() * 1000) 139 if not ready_fds: 140 raise error.TestError('Timed out waiting for jail-request') 141 142 # Sanity check the request. 143 path = os.read(req_fd, self.PATH_MAX) 144 logging.info('Received jail-request for path %s', path) 145 if path != self._path_to_jail: 146 raise error.TestError('Got request for the wrong path') 147 148 os.write(req_fd, struct.pack('I', verdict)) 149 logging.info('Responded to jail-request') 150 151 return dev_file_wrapper[0] 152 153 154 def __exit__(self, exc_type, exc_val, traceback): 155 if self._owns_device: 156 subprocess.call(['device_jail_utility', 157 '--remove={0}'.format(self._path)]) 158 159 160 def get_usb_devices(): 161 context = pyudev.Context() 162 return [device for device in context.list_devices() 163 if device.device_node and device.device_node.startswith('/dev/bus/usb')] 164
