Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

new device API #32

Merged
merged 23 commits into from
Jun 21, 2024
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
d985cf1
q-dev: implement part of new API for DeviceInfo
piotrbartman Dec 5, 2023
ab08892
q-dev: events
piotrbartman Dec 28, 2023
30fa006
q-dev: DeviceInterface
piotrbartman Dec 30, 2023
b8513dc
q-dev: assignments -> get_assigned_devices
piotrbartman Jan 22, 2024
1f3aa0c
q-dev: fire device-attach on domain start
piotrbartman Jan 23, 2024
ece1fd0
q-dev: usb device full identity
piotrbartman Jan 30, 2024
daac099
q-dev: frontend_device -> attachment
piotrbartman Feb 1, 2024
d261a0f
q-dev: implementation of self_identity
piotrbartman Feb 17, 2024
785bd62
q-dev: port assignment
piotrbartman Feb 20, 2024
e7acdf4
q-dev: use ext/utils
piotrbartman Mar 19, 2024
f9f84be
q-dev: device protocol
piotrbartman Mar 20, 2024
0be9633
q-dev: minor optimization
piotrbartman Apr 24, 2024
fca4777
q-dev: update integ tests
piotrbartman May 14, 2024
9094561
q-dev: fix attaching usb devices on domain start
piotrbartman May 18, 2024
f6ad1c3
q-dev: small fix for unknown devices
piotrbartman May 29, 2024
5cb2a8f
q-dev: keep partial backward compatibility
piotrbartman Jun 1, 2024
cad623a
q-dev: keep partial backward compatibility for auto-attachment
piotrbartman Jun 5, 2024
227ec98
q-dev: keep partial backward compatibility in tests
piotrbartman Jun 7, 2024
7cbbb49
q-dev: do not use unicode_escape
piotrbartman Jun 17, 2024
54fca94
q-dev: cleanup
piotrbartman Jun 17, 2024
18abd26
q-dev: handle invalid values
piotrbartman Jun 17, 2024
29d32c5
q-dev: handle invalid values
piotrbartman Jun 18, 2024
f86dbf6
q-dev: fix loop
piotrbartman Jun 21, 2024
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
q-dev: keep partial backward compatibility in tests
  • Loading branch information
piotrbartman committed Jun 10, 2024
commit 227ec98664c6541cf7d49c006ea77220d71498b5
8 changes: 4 additions & 4 deletions qubesusbproxy/core3ext.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ def get_assigned_devices(devices):
class DescriptionOverrider:
@property
def description(self):
return self.vendor + " " + self.product
return self.name

class DeviceInfo(DescriptionOverrider, LegacyDeviceInfo):
def __init__(self, *args, **kwargs):
Expand All @@ -70,9 +70,9 @@ def __init__(self, *args, **kwargs):
# `_load_interfaces_from_qubesdb` will never be called
self._interfaces = "?******"

@property
def fronted_domain(self):
return self.attachment
@property
def frontend_domain(self):
return self.attachment

class DeviceInterface:
pass
Expand Down
78 changes: 46 additions & 32 deletions qubesusbproxy/tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,11 +27,35 @@

core2 = False
core3 = False
legacy = False
try:
import qubesusbproxy.core3ext
import qubes.devices
import asyncio

try:
from qubes.device_protocol import DeviceAssignment

def assign(test, collection, assignment):
test.loop.run_until_complete(collection.assign(assignment))

def unassign(test, collection, assignment):
test.loop.run_until_complete(collection.unassign(assignment))

AUTO_ATTACH = {"attach_automatically": True, "required": True}
except ImportError:
# This extension supports both the legacy and new device API.
# In the case of the legacy backend, functionality is limited.
from qubes.devices import DeviceAssignment

def assign(test, collection, assignment):
test.loop.run_until_complete(collection.attach(assignment))

def unassign(test, collection, assignment):
test.loop.run_until_complete(collection.detach(assignment))

legacy = True
AUTO_ATTACH = {"persistent": True}

core3 = True
except ImportError:
pass
Expand Down Expand Up @@ -380,11 +404,9 @@ def test_000_list(self):

def test_010_assign(self):
usb_dev = self.backend.devices['usb'][self.usbdev_ident]
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident,
attach_automatically=True,
required=True)
self.loop.run_until_complete(
self.frontend.devices['usb'].assign(ass))
ass = DeviceAssignment(
self.backend, self.usbdev_ident, **AUTO_ATTACH)
assign(self, self.frontend.devices['usb'], ass)
self.assertIsNone(usb_dev.attachment)
try:
self.frontend.start()
Expand All @@ -395,13 +417,12 @@ def test_010_assign(self):
wait=True), 0,
"Device connection failed")

self.assertEqual(usb_dev.attachment,
self.frontend)
self.assertEqual(usb_dev.attachment, self.frontend)

def test_020_attach(self):
self.frontend.start()
usb_dev = self.backend.devices['usb'][self.usbdev_ident]
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
try:
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))
Expand All @@ -412,13 +433,12 @@ def test_020_attach(self):
wait=True), 0,
"Device connection failed")

self.assertEquals(usb_dev.attachment,
self.frontend)
self.assertEquals(usb_dev.attachment, self.frontend)

def test_030_detach(self):
self.frontend.start()
usb_dev = self.backend.devices['usb'][self.usbdev_ident]
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
try:
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))
Expand All @@ -438,14 +458,11 @@ def test_030_detach(self):

def test_040_unassign(self):
usb_dev = self.backend.devices['usb'][self.usbdev_ident]
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident,
attach_automatically=True,
required=True)
self.loop.run_until_complete(
self.frontend.devices['usb'].assign(ass))
ass = DeviceAssignment(
self.backend, self.usbdev_ident, **AUTO_ATTACH)
assign(self, self.frontend.devices['usb'], ass)
self.assertIsNone(usb_dev.attachment)
self.loop.run_until_complete(
self.frontend.devices['usb'].unassign(ass))
unassign(self, self.frontend.devices['usb'], ass)
self.assertIsNone(usb_dev.attachment)

def test_050_list_attached(self):
Expand All @@ -454,7 +471,7 @@ def test_050_list_attached(self):
usb_list = self.backend.devices['usb']

usb_list_front_pre = list(self.frontend.devices['usb'])
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)

try:
self.loop.run_until_complete(
Expand All @@ -466,8 +483,7 @@ def test_050_list_attached(self):
wait=True), 0,
"Device connection failed")

self.assertEquals(usb_list[self.usbdev_ident].attachment,
self.frontend)
self.assertEquals(usb_list[self.usbdev_ident].attachment, self.frontend)

usb_list_front_post = list(self.frontend.devices['usb'])

Expand All @@ -476,7 +492,7 @@ def test_050_list_attached(self):
def test_060_auto_detach_on_remove(self):
self.frontend.start()
usb_list = self.backend.devices['usb']
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
try:
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))
Expand All @@ -495,12 +511,10 @@ def test_060_auto_detach_on_remove(self):
def test_061_auto_attach_on_reconnect(self):
self.frontend.start()
usb_list = self.backend.devices['usb']
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident,
attach_automatically=True,
required=True)
ass = DeviceAssignment(
self.backend, self.usbdev_ident, **AUTO_ATTACH)
try:
self.loop.run_until_complete(
self.frontend.devices['usb'].assign(ass))
assign(self, self.frontend.devices['usb'], ass)
except qubesusbproxy.core3ext.USBProxyNotInstalled as e:
self.skipTest(str(e))

Expand Down Expand Up @@ -530,7 +544,7 @@ def test_070_attach_not_installed_front(self):
user="root", wait=True)
if retcode != 0:
raise RuntimeError("Failed to simulate not installed package")
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
with self.assertRaises(qubesusbproxy.core3ext.USBProxyNotInstalled):
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))
Expand All @@ -543,7 +557,7 @@ def test_075_attach_not_installed_back(self):
user="root", wait=True)
if retcode != 0:
raise RuntimeError("Failed to simulate not installed package")
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
try:
with self.assertRaises(qubesusbproxy.core3ext.USBProxyNotInstalled):
self.loop.run_until_complete(
Expand All @@ -560,7 +574,7 @@ def test_080_attach_existing_policy(self):
'/etc/qubes-rpc/policy/qubes.USB+{}'.format(self.usbdev_ident),
'w+') as policy_file:
policy_file.write('# empty policy\n')
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))

Expand All @@ -569,7 +583,7 @@ def test_090_attach_stubdom(self):
self.frontend.virt_mode = 'hvm'
self.frontend.features['stubdom-qrexec'] = True
self.frontend.start()
ass = qubes.devices.DeviceAssignment(self.backend, self.usbdev_ident)
ass = DeviceAssignment(self.backend, self.usbdev_ident)
try:
self.loop.run_until_complete(
self.frontend.devices['usb'].attach(ass))
Expand Down