diff --git a/core/qubesutils.py b/core/qubesutils.py index 689138a3..417a609d 100644 --- a/core/qubesutils.py +++ b/core/qubesutils.py @@ -25,6 +25,7 @@ from __future__ import absolute_import import string +import errno from lxml import etree from lxml.etree import ElementTree, SubElement, Element @@ -463,260 +464,184 @@ def block_detach_all(vm): usb_ver_re = re.compile(r"^(1|2)$") usb_device_re = re.compile(r"^[0-9]+-[0-9]+(_[0-9]+)?$") usb_port_re = re.compile(r"^$|^[0-9]+-[0-9]+(\.[0-9]+)?$") +usb_desc_re = re.compile(r"^[ -~]{1,255}$") +# should match valid VM name +usb_connected_to_re = re.compile(r"^[a-zA-Z][a-zA-Z0-9_.-]*$") -def usb_setup(backend_vm_xid, vm_xid, devid, usb_ver): - """ - Attach frontend to the backend. - backend_vm_xid - id of the backend domain - vm_xid - id of the frontend domain - devid - id of the pvusb controller - """ - num_ports = 8 - trans = vmm.xs.transaction_start() - - be_path = "/local/domain/%d/backend/vusb/%d/%d" % (backend_vm_xid, vm_xid, devid) - fe_path = "/local/domain/%d/device/vusb/%d" % (vm_xid, devid) - - be_perm = [{'dom': backend_vm_xid}, {'dom': vm_xid, 'read': True} ] - fe_perm = [{'dom': vm_xid}, {'dom': backend_vm_xid, 'read': True} ] - - # Create directories and set permissions - vmm.xs.write(trans, be_path, "") - vmm.xs.set_permissions(trans, be_path, be_perm) - - vmm.xs.write(trans, fe_path, "") - vmm.xs.set_permissions(trans, fe_path, fe_perm) - - # Write backend information into the location that frontend looks for - vmm.xs.write(trans, "%s/backend-id" % fe_path, str(backend_vm_xid)) - vmm.xs.write(trans, "%s/backend" % fe_path, be_path) - - # Write frontend information into the location that backend looks for - vmm.xs.write(trans, "%s/frontend-id" % be_path, str(vm_xid)) - vmm.xs.write(trans, "%s/frontend" % be_path, fe_path) - - # Write USB Spec version field. - vmm.xs.write(trans, "%s/usb-ver" % be_path, usb_ver) - - # Write virtual root hub field. - vmm.xs.write(trans, "%s/num-ports" % be_path, str(num_ports)) - for port in range(1, num_ports+1): - # Set all port to disconnected state - vmm.xs.write(trans, "%s/port/%d" % (be_path, port), "") - - # Set state to XenbusStateInitialising - vmm.xs.write(trans, "%s/state" % fe_path, "1") - vmm.xs.write(trans, "%s/state" % be_path, "1") - vmm.xs.write(trans, "%s/online" % be_path, "1") - - vmm.xs.transaction_end(trans) - -def usb_decode_device_from_xs(xs_encoded_device): +def usb_decode_device_from_qdb(qdb_encoded_device): """ recover actual device name (xenstore doesn't allow dot in key names, so it was translated to underscore) """ - return xs_encoded_device.replace('_', '.') + return qdb_encoded_device.replace('_', '.') -def usb_encode_device_for_xs(device): +def usb_encode_device_for_qdb(device): """ encode actual device name (xenstore doesn't allow dot in key names, so translated it into underscore) """ return device.replace('.', '_') -def usb_list(): +def usb_list_vm(vm): + if not vm.is_running(): + return {} + + try: + untrusted_devices = vm.qdb.multiread('/qubes-usb-devices/') + except Error: + vm.refresh() + return {} + + def get_dev_item(dev, item): + return untrusted_devices.get( + '/qubes-usb-devices/%s/%s' % (dev, item), + None) + + devices = {} + + untrusted_devices_names = list(set(map(lambda x: x.split("/")[2], + untrusted_devices.keys()))) + for untrusted_dev_name in untrusted_devices_names: + if usb_device_re.match(untrusted_dev_name): + dev_name = untrusted_dev_name + untrusted_device_desc = get_dev_item(dev_name, 'desc') + if not usb_desc_re.match(untrusted_device_desc): + print >> sys.stderr, "Invalid %s device desc in VM '%s'" % ( + dev_name, vm.name) + continue + device_desc = untrusted_device_desc + + untrusted_connected_to = get_dev_item(dev_name, 'connected-to') + if untrusted_connected_to: + if not usb_connected_to_re.match(untrusted_connected_to): + print >>sys.stderr, \ + "Invalid %s device 'connected-to' in VM '%s'" % ( + dev_name, vm.name) + continue + connected_to = untrusted_connected_to + else: + connected_to = None + + device = usb_decode_device_from_qdb(dev_name) + + full_name = vm.name + ':' + device + + devices[full_name] = { + 'vm': vm, + 'device': device, + 'qdb_path': '/qubes-usb-devices/' + dev_name, + 'name': full_name, + 'desc': device_desc, + 'connected-to': connected_to, + } + return devices + + +def usb_list(qvmc=None, vm=None): """ Returns a dictionary of USB devices (for PVUSB backends running in all VM). The dictionary is keyed by 'name' (see below), each element is a dictionary itself: - vm = name of the backend domain - xid = xid of the backend domain - device = - - name = :- + vm = backend domain object + device = device ID + name = : desc = description """ - # FIXME: any better idea of desc_re? - desc_re = re.compile(r"^.{1,255}$") + if vm is not None: + if not vm.is_running(): + return {} + else: + vm_list = [vm] + else: + if qvmc is None: + raise QubesException("You must pass either qvm or vm argument") + vm_list = qvmc.values() devices_list = {} - - xs_trans = vmm.xs.transaction_start() - vm_list = vmm.xs.ls(xs_trans, '/local/domain') - - for xid in vm_list: - vm_name = vmm.xs.read(xs_trans, '/local/domain/%s/name' % xid) - vm_devices = vmm.xs.ls(xs_trans, '/local/domain/%s/qubes-usb-devices' % xid) - if vm_devices is None: - continue - # when listing devices in xenstore we get encoded names - for xs_encoded_device in vm_devices: - # Sanitize device id - if not usb_device_re.match(xs_encoded_device): - print >> sys.stderr, "Invalid device id in backend VM '%s'" % vm_name - continue - device = usb_decode_device_from_xs(xs_encoded_device) - device_desc = vmm.xs.read(xs_trans, '/local/domain/%s/qubes-usb-devices/%s/desc' % (xid, xs_encoded_device)) - if not desc_re.match(device_desc): - print >> sys.stderr, "Invalid %s device desc in VM '%s'" % (device, vm_name) - continue - visible_name = "%s:%s" % (vm_name, device) - # grab version - usb_ver = vmm.xs.read(xs_trans, '/local/domain/%s/qubes-usb-devices/%s/usb-ver' % (xid, xs_encoded_device)) - if usb_ver is None or not usb_ver_re.match(usb_ver): - print >> sys.stderr, "Invalid %s device USB version in VM '%s'" % (device, vm_name) - continue - devices_list[visible_name] = {"name": visible_name, "xid":int(xid), - "vm": vm_name, "device":device, - "desc":device_desc, - "usb_ver":usb_ver} - - vmm.xs.transaction_end(xs_trans) + for vm in vm_list: + devices_list.update(usb_list_vm(vm)) return devices_list -def usb_check_attached(xs_trans, backend_vm, device): - """ - Checks if the given device in the given backend attached to any frontend. - Parameters: - backend_vm - xid of the backend domain - device - device name in the backend domain - Returns None or a dictionary: - vm - the name of the frontend domain - xid - xid of the frontend domain - frontend - frontend device number FIXME - devid - frontend port number FIXME - """ - # sample xs content: /local/domain/0/backend/vusb/4/0/port/1 = "7-5" - attached_dev = None - vms = vmm.xs.ls(xs_trans, '/local/domain/%d/backend/vusb' % backend_vm) - if vms is None: - return None - for vm in vms: - if not vm.isdigit(): - print >> sys.stderr, "Invalid VM id" - continue - frontend_devs = vmm.xs.ls(xs_trans, '/local/domain/%d/backend/vusb/%s' % (backend_vm, vm)) - if frontend_devs is None: - continue - for frontend_dev in frontend_devs: - if not frontend_dev.isdigit(): - print >> sys.stderr, "Invalid frontend in VM %s" % vm - continue - ports = vmm.xs.ls(xs_trans, '/local/domain/%d/backend/vusb/%s/%s/port' % (backend_vm, vm, frontend_dev)) - if ports is None: - continue - for port in ports: - # FIXME: refactor, see similar loop in usb_find_unused_frontend(), use usb_list() instead? - if not port.isdigit(): - print >> sys.stderr, "Invalid port in VM %s frontend %s" % (vm, frontend) - continue - dev = vmm.xs.read(xs_trans, '/local/domain/%d/backend/vusb/%s/%s/port/%s' % (backend_vm, vm, frontend_dev, port)) - if dev == "": - continue - # Sanitize device id - if not usb_port_re.match(dev): - print >> sys.stderr, "Invalid device id in backend VM %d @ %s/%s/port/%s" % \ - (backend_vm, vm, frontend_dev, port) - continue - if dev == device: - frontend = "%s-%s" % (frontend_dev, port) - #TODO - vm_name = xl_ctx.domid_to_name(int(vm)) - if vm_name is None: - # FIXME: should we wipe references to frontends running on nonexistent VMs? - continue - attached_dev = {"xid":int(vm), "frontend": frontend, "devid": device, "vm": vm_name} - break - return attached_dev - -#def usb_check_frontend_busy(vm, front_dev, port): -# devport = frontend.split("-") -# if len(devport) != 2: -# raise QubesException("Malformed frontend syntax, must be in device-port format") -# # FIXME: -# # return vmm.xs.read('', '/local/domain/%d/device/vusb/%d/state' % (vm.xid, frontend)) == '4' -# return False - -def usb_find_unused_frontend(xs_trans, backend_vm_xid, vm_xid, usb_ver): - """ - Find an unused frontend/port to link the given backend with the given frontend. - Creates new frontend if needed. - Returns frontend specification in - format. - """ - - # This variable holds an index of last frontend scanned by the loop below. - # If nothing found, this value will be used to derive the index of a new frontend. - last_frontend_dev = -1 - - frontend_devs = vmm.xs.ls(xs_trans, "/local/domain/%d/device/vusb" % vm_xid) - if frontend_devs is not None: - for frontend_dev in frontend_devs: - if not frontend_dev.isdigit(): - print >> sys.stderr, "Invalid frontend_dev in VM %d" % vm_xid - continue - frontend_dev = int(frontend_dev) - fe_path = "/local/domain/%d/device/vusb/%d" % (vm_xid, frontend_dev) - if vmm.xs.read(xs_trans, "%s/backend-id" % fe_path) == str(backend_vm_xid): - if vmm.xs.read(xs_trans, '/local/domain/%d/backend/vusb/%d/%d/usb-ver' % (backend_vm_xid, vm_xid, frontend_dev)) != usb_ver: - last_frontend_dev = frontend_dev - continue - # here: found an existing frontend already connected to right backend using an appropriate USB version - ports = vmm.xs.ls(xs_trans, '/local/domain/%d/backend/vusb/%d/%d/port' % (backend_vm_xid, vm_xid, frontend_dev)) - if ports is None: - print >> sys.stderr, "No ports in VM %d frontend_dev %d?" % (vm_xid, frontend_dev) - last_frontend_dev = frontend_dev - continue - for port in ports: - # FIXME: refactor, see similar loop in usb_check_attached(), use usb_list() instead? - if not port.isdigit(): - print >> sys.stderr, "Invalid port in VM %d frontend_dev %d" % (vm_xid, frontend_dev) - continue - port = int(port) - dev = vmm.xs.read(xs_trans, '/local/domain/%d/backend/vusb/%d/%s/port/%s' % (backend_vm_xid, vm_xid, frontend_dev, port)) - # Sanitize device id - if not usb_port_re.match(dev): - print >> sys.stderr, "Invalid device id in backend VM %d @ %d/%d/port/%d" % \ - (backend_vm_xid, vm_xid, frontend_dev, port) - continue - if dev == "": - return '%d-%d' % (frontend_dev, port) - last_frontend_dev = frontend_dev - - # create a new frontend_dev and link it to the backend - frontend_dev = last_frontend_dev + 1 - usb_setup(backend_vm_xid, vm_xid, frontend_dev, usb_ver) - return '%d-%d' % (frontend_dev, 1) - -def usb_attach(vm, backend_vm, device, frontend=None, auto_detach=False, wait=True): - device_attach_check(vm, backend_vm, device, frontend) - - xs_trans = vmm.xs.transaction_start() - - xs_encoded_device = usb_encode_device_for_xs(device) - usb_ver = vmm.xs.read(xs_trans, '/local/domain/%s/qubes-usb-devices/%s/usb-ver' % (backend_vm.xid, xs_encoded_device)) - if usb_ver is None or not usb_ver_re.match(usb_ver): - vmm.xs.transaction_end(xs_trans) - raise QubesException("Invalid %s device USB version in VM '%s'" % (device, backend_vm.name)) - - if frontend is None: - frontend = usb_find_unused_frontend(xs_trans, backend_vm.xid, vm.xid, usb_ver) +def usb_check_attached(device): + """Reread device attachment status""" + vm = device['vm'] + untrusted_connected_to = vm.qdb.read( + '{}/connected-to'.format(device['qdb_path'])) + if untrusted_connected_to: + if not usb_connected_to_re.match(untrusted_connected_to): + raise QubesException( + "Invalid %s device 'connected-to' in VM '%s'" % ( + device['device'], vm.name)) + connected_to = untrusted_connected_to else: - # Check if any device attached at this frontend - #if usb_check_frontend_busy(vm, frontend): - # raise QubesException("Frontend %s busy in VM %s, detach it first" % (frontend, vm.name)) - vmm.xs.transaction_end(xs_trans) - raise NotImplementedError("Explicit USB frontend specification is not implemented yet") + connected_to = None + return connected_to - # Check if this device is attached to some domain - attached_vm = usb_check_attached(xs_trans, backend_vm.xid, device) - vmm.xs.transaction_end(xs_trans) +def usb_attach(vm, device, auto_detach=False, wait=True): + if not vm.is_running(): + raise QubesException("VM {} not running".format(vm.name)) - if attached_vm: + if not device['vm'].is_running(): + raise QubesException("VM {} not running".format(device['vm'].name)) + + connected_to = usb_check_attached(device) + if connected_to: if auto_detach: - usb_detach(backend_vm, attached_vm) + usb_detach(device) else: - raise QubesException("Device %s from %s already connected to VM %s as %s" % (device, backend_vm.name, attached_vm['vm'], attached_vm['frontend'])) + raise QubesException("Device {} already connected, to {}".format( + device['name'], connected_to + )) - # Run helper script - xl_cmd = [ '/usr/lib/qubes/xl-qvm-usb-attach.py', str(vm.xid), device, frontend, str(backend_vm.xid) ] - subprocess.check_call(xl_cmd) + # set qrexec policy to allow this device + policy_line = '{} {} allow\n'.format(vm.name, device['vm'].name) + policy_path = '/etc/qubes-rpc/policy/qubes.USB+{}'.format(device['device']) + policy_exists = os.path.exists(policy_path) + if not policy_exists: + try: + fd = os.open(policy_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY) + with os.fdopen(fd, 'w') as f: + f.write(policy_line) + except OSError as e: + if e.errno == errno.EEXIST: + pass + else: + raise + else: + with open(policy_path, 'r+') as f: + policy = f.readlines() + policy.insert(0, policy_line) + f.truncate(0) + f.seek(0) + f.write(''.join(policy)) + try: + # and actual attach + p = vm.run_service('qubes.USBAttach', passio_popen=True, user='root') + (stdout, stderr) = p.communicate( + '{} {}\n'.format(device['vm'].name, device['device'])) + if p.returncode != 0: + # TODO: sanitize and include stdout + raise QubesException('Device attach failed') + finally: + # FIXME: there is a race condition here - some other process might + # modify the file in the meantime. This may result in unexpected + # denials, but will not allow too much + if not policy_exists: + os.unlink(policy_path) + else: + with open(policy_path, 'r+') as f: + policy = f.readlines() + policy.remove('{} {} allow\n'.format(vm.name, device['vm'].name)) + f.truncate(0) + f.seek(0) + f.write(''.join(policy)) -def usb_detach(backend_vm, attachment): - xl_cmd = [ '/usr/lib/qubes/xl-qvm-usb-detach.py', str(attachment['xid']), attachment['devid'], attachment['frontend'], str(backend_vm.xid) ] - subprocess.check_call(xl_cmd) +def usb_detach(vm, device): + connected_to = usb_check_attached(device) + # detect race conditions; there is still race here, but much smaller + if vm.name != connected_to: + raise QubesException( + "Device {} not connected to VM {}".format(device['name'], vm.name)) + + p = vm.run_service('qubes.USBDetach', passio_popen=True, user='root') + (stdout, stderr) = p.communicate( + '{} {}\n'.format(device['vm'].name, device['device'])) + if p.returncode != 0: + # TODO: sanitize and include stdout + raise QubesException('Device detach failed') def usb_detach_all(vm): raise NotImplementedError("Detaching all devices from a given VM is not implemented yet") diff --git a/qvm-tools/qvm-usb b/qvm-tools/qvm-usb index 31556379..3faa7b25 100755 --- a/qvm-tools/qvm-usb +++ b/qvm-tools/qvm-usb @@ -78,11 +78,10 @@ def main(): print >> sys.stderr, "Only one of -l -a -d is allowed!" exit (1) - if options.do_attach or options.do_detach: - qvm_collection = QubesVmCollection() - qvm_collection.lock_db_for_reading() - qvm_collection.load() - qvm_collection.unlock_db() + qvm_collection = QubesVmCollection() + qvm_collection.lock_db_for_reading() + qvm_collection.load() + qvm_collection.unlock_db() if options.do_attach: if (len (args) != 2): @@ -91,14 +90,17 @@ def main(): if vm is None: parser.error ("Invalid VM name: %s" % args[0]) - # FIXME: here we assume that device is always in form "domain:dev", which can be changed in the future + # FIXME: here we assume that device is always in form "domain:dev", + # which can be changed in the future if args[1].find(":") < 0: - parser.error ("Invalid device syntax: %s" % args[1]) - dev_list = usb_list() + parser.error("Invalid device syntax: %s" % args[1]) + backend_vm = qvm_collection.get_vm_by_name(args[1].split(":")[0]) + if backend_vm is None: + parser.error("No such VM: {}".format(args[1].split(":")[0])) + dev_list = usb_list(vm=backend_vm) if not args[1] in dev_list.keys(): - parser.error ("Invalid device name: %s" % args[1]) + parser.error("Invalid device name: %s" % args[1]) dev = dev_list[args[1]] - backend_vm = qvm_collection.get_vm_by_name(dev['vm']) assert backend_vm is not None kwargs = {} @@ -106,14 +108,14 @@ def main(): # kwargs['frontend'] = options.frontend kwargs['auto_detach'] = options.auto_detach try: - usb_attach(vm, backend_vm, dev['device'], **kwargs) + usb_attach(vm, dev, **kwargs) except QubesException as e: print >> sys.stderr, "ERROR: %s" % str(e) sys.exit(1) elif options.do_detach: if (len (args) < 1): parser.error ("You must provide device or vm name!") - if len(args) > 1: + if len(args) > 1: parser.error ("Too many parameters") # Check if provided name is VM vm = qvm_collection.get_vm_by_name(args[0]) @@ -127,32 +129,34 @@ def main(): else: # Maybe usbvm:device? - # FIXME: nasty copy-paste from attach code half a page above - # FIXME: here we assume that device is always in form "domain:dev", which can be changed in the future + # FIXME: nasty copy-paste from attach code half a page above + # FIXME: here we assume that device is always in form "domain:dev", + # which can be changed in the future if args[0].find(":") < 0: - parser.error ("Invalid device syntax: %s" % args[0]) - dev_list = usb_list() + parser.error("Invalid device syntax: %s" % args[0]) + backend_vm = qvm_collection.get_vm_by_name(args[0].split(":")[0]) + if backend_vm is None: + parser.error("No such VM: {}".format(args[0].split(":")[0])) + dev_list = usb_list(vm=backend_vm) if not args[0] in dev_list.keys(): - parser.error ("Invalid device name: %s" % args[0]) + parser.error("Invalid device name: %s" % args[0]) dev = dev_list[args[0]] - backend_vm = qvm_collection.get_vm_by_name(dev['vm']) - assert backend_vm is not None - - attached_to = usb_check_attached('', backend_vm.xid, dev['device']) + attached_to = usb_check_attached(dev) if attached_to is None: print >> sys.stderr, "WARNING: Device not connected to any VM" exit(0) - usb_detach(backend_vm, attached_to) + vm = qvm_collection.get_vm_by_name(attached_to) + usb_detach(vm, dev) else: - if len(args) > 0: - parser.error ("Too many parameters") + if len(args) > 0: + parser.error("Too many parameters") # do_list - for dev in usb_list().values(): - attached_to = usb_check_attached('', dev['xid'], dev['device']) + for dev in usb_list(qvmc=qvm_collection).values(): + attached_to = usb_check_attached(dev) attached_to_str = "" if attached_to: - attached_to_str = " (attached to %s:%s)" % (attached_to['vm'], attached_to['frontend']) - print "%s\t%s%s (USBv%s)" % (dev['name'], dev['desc'], attached_to_str, dev['usb_ver']) + attached_to_str = " (attached to %s)" % (attached_to) + print "%s\t%s%s" % (dev['name'], dev['desc'], attached_to_str) exit (0) main()