Skip to content

Add method to solely create disposable #401

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.

Sign up for GitHub

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

Open
ben-grande wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Add method to solely create disposable #401

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
963532c
Add method to solely create disposable
ben-grande Dec 10, 2025
96e818a
Use newer disposable token
ben-grande Dec 10, 2025
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
50 changes: 25 additions & 25 deletions qubesadmin/tests/tools/qvm_run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -668,15 +668,15 @@ def test_008_dispvm_remote(self):
self.app.service_calls,
[
(
"$dispvm",
"@dispvm",
"test.service",
{
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"user": None,
},
),
("$dispvm", "test.service", b""),
("@dispvm", "test.service", b""),
],
)
self.assertAllCalled()
Expand All @@ -699,15 +699,15 @@ def test_009_dispvm_remote_specific(self):
self.app.service_calls,
[
(
"$dispvm:test-vm",
"@dispvm:test-vm",
"test.service",
{
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"user": None,
},
),
("$dispvm:test-vm", "test.service", b""),
("@dispvm:test-vm", "test.service", b""),
],
)
self.assertAllCalled()
Expand Down Expand Up @@ -1150,7 +1150,7 @@ def test_022_no_shell(self):
def test_023_dispvm_no_shell(self):
self.app.expected_calls[
(
"$dispvm:test-vm",
"@dispvm:test-vm",
"admin.vm.feature.CheckWithTemplate",
"vmexec",
None,
Expand All @@ -1165,15 +1165,15 @@ def test_023_dispvm_no_shell(self):
self.app.service_calls,
[
(
"$dispvm:test-vm",
"@dispvm:test-vm",
"qubes.VMExec+command",
{
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"user": None,
},
),
("$dispvm:test-vm", "qubes.VMExec+command", b""),
("@dispvm:test-vm", "qubes.VMExec+command", b""),
],
)
self.assertAllCalled()
Expand Down Expand Up @@ -1277,7 +1277,7 @@ def test_026_no_shell_double_dashdash(self):
def test_027_no_shell_dispvm(self):
self.app.expected_calls[
(
"$dispvm:test-vm",
"@dispvm:test-vm",
"admin.vm.feature.CheckWithTemplate",
"vmexec",
None,
Expand All @@ -1292,23 +1292,23 @@ def test_027_no_shell_dispvm(self):
self.app.service_calls,
[
(
"$dispvm:test-vm",
"@dispvm:test-vm",
"qubes.VMExec+command+arg",
{
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"user": None,
},
),
("$dispvm:test-vm", "qubes.VMExec+command+arg", b""),
("@dispvm:test-vm", "qubes.VMExec+command+arg", b""),
],
)
self.assertAllCalled()

def test_028_argparse_bug_workaround(self):
self.app.expected_calls[
(
"$dispvm:test-vm",
"@dispvm:test-vm",
"admin.vm.feature.CheckWithTemplate",
"vmexec",
None,
Expand All @@ -1323,23 +1323,23 @@ def test_028_argparse_bug_workaround(self):
self.app.service_calls,
[
(
"$dispvm:test-vm",
"@dispvm:test-vm",
"qubes.VMExec+command+----",
{
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"user": None,
},
),
("$dispvm:test-vm", "qubes.VMExec+command+----", b""),
("@dispvm:test-vm", "qubes.VMExec+command+----", b""),
],
)
self.assertAllCalled()

def test_029_command_is_dashdash(self):
self.app.expected_calls[
(
"$dispvm:test-vm",
"@dispvm:test-vm",
"admin.vm.feature.CheckWithTemplate",
"vmexec",
None,
Expand All @@ -1354,22 +1354,22 @@ def test_029_command_is_dashdash(self):
self.app.service_calls,
[
(
"$dispvm:test-vm",
"@dispvm:test-vm",
"qubes.VMExec+----",
{
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"user": None,
},
),
("$dispvm:test-vm", "qubes.VMExec+----", b""),
("@dispvm:test-vm", "qubes.VMExec+----", b""),
],
)
self.assertAllCalled()

def test_030_no_shell_dispvm(self):
self.app.expected_calls[
("$dispvm", "admin.vm.feature.CheckWithTemplate", "vmexec", None)
("@dispvm", "admin.vm.feature.CheckWithTemplate", "vmexec", None)
] = b"0\x001"
ret = qubesadmin.tools.qvm_run.main(
["--no-gui", "--dispvm", "--", "test-vm", "command", "arg"],
Expand All @@ -1380,22 +1380,22 @@ def test_030_no_shell_dispvm(self):
self.app.service_calls,
[
(
"$dispvm",
"@dispvm",
"qubes.VMExec+test--vm+command+arg",
{
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"user": None,
},
),
("$dispvm", "qubes.VMExec+test--vm+command+arg", b""),
("@dispvm", "qubes.VMExec+test--vm+command+arg", b""),
],
)
self.assertAllCalled()

def test_031_argparse_bug_workaround(self):
self.app.expected_calls[
("$dispvm", "admin.vm.feature.CheckWithTemplate", "vmexec", None)
("@dispvm", "admin.vm.feature.CheckWithTemplate", "vmexec", None)
] = b"0\x001"
ret = qubesadmin.tools.qvm_run.main(
["--no-gui", "--dispvm", "--", "test-vm", "command", "--"],
Expand All @@ -1406,23 +1406,23 @@ def test_031_argparse_bug_workaround(self):
self.app.service_calls,
[
(
"$dispvm",
"@dispvm",
"qubes.VMExec+test--vm+command+----",
{
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"user": None,
},
),
("$dispvm", "qubes.VMExec+test--vm+command+----", b""),
("@dispvm", "qubes.VMExec+test--vm+command+----", b""),
],
)
self.assertAllCalled()

@unittest.expectedFailure
def test_032_argparse_bug_workaround_unnamed_dispvm(self):
self.app.expected_calls[
("$dispvm", "admin.vm.feature.CheckWithTemplate", "vmexec", None)
("@dispvm", "admin.vm.feature.CheckWithTemplate", "vmexec", None)
] = b"0\x001"
ret = qubesadmin.tools.qvm_run.main(
["--no-gui", "--dispvm", "test-vm", "command", "--"], app=self.app
Expand All @@ -1432,15 +1432,15 @@ def test_032_argparse_bug_workaround_unnamed_dispvm(self):
self.app.service_calls,
[
(
"$dispvm",
"@dispvm",
"qubes.VMExec+test--vm+command+----",
{
"stdout": subprocess.DEVNULL,
"stderr": subprocess.DEVNULL,
"user": None,
},
),
("$dispvm", "qubes.VMExec+test--vm+command+----", b""),
("@dispvm", "qubes.VMExec+test--vm+command+----", b""),
],
)
self.assertAllCalled()
8 changes: 4 additions & 4 deletions qubesadmin/tests/vm/dispvm.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,8 +73,8 @@ def test_010_remote_create_default(self):
vm.run_service_for_stdio('test.service')
vm.cleanup()
self.assertEqual(self.app.service_calls, [
('$dispvm', 'test.service', {}),
('$dispvm', 'test.service', b''),
('@dispvm', 'test.service', {}),
('@dispvm', 'test.service', b''),
])
self.assertAllCalled()

Expand All @@ -83,8 +83,8 @@ def test_011_remote_create_specific(self):
vm.run_service_for_stdio('test.service')
vm.cleanup()
self.assertEqual(self.app.service_calls, [
('$dispvm:test-vm', 'test.service', {}),
('$dispvm:test-vm', 'test.service', b''),
('@dispvm:test-vm', 'test.service', {}),
('@dispvm:test-vm', 'test.service', b''),
])
self.assertAllCalled()

Expand Down
62 changes: 32 additions & 30 deletions qubesadmin/vm/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -454,57 +454,59 @@ def set_notes(self, notes: str):


class DispVMWrapper(QubesVM):
"""Wrapper class for new DispVM, supporting only service call
"""
Wrapper class for new disposable, supporting only service call.

Note that when running in dom0, one need to manually kill the DispVM after
service call ends.
Note that when running in dom0, one need to manually kill the disposable
after service call ends.
"""

def run_service(self, service, **kwargs):
if self.app.qubesd_connection_type == "socket":
# create dispvm at service call
if self._method_dest.startswith("$dispvm"):
if self._method_dest.startswith("$dispvm:"):
method_dest = self._method_dest[len("$dispvm:") :]
else:
method_dest = "dom0"
dispvm = self.app.qubesd_call(
method_dest, "admin.vm.CreateDisposable"
)
dispvm = dispvm.decode("ascii")
self._method_dest = dispvm
# Service call may wait for session start, give it more time
# than default 5s
kwargs["connect_timeout"] = self.qrexec_timeout
"""Create disposable if absent and run service."""
if (
self.app.qubesd_connection_type == "socket"
and self._method_dest.startswith("@dispvm")
):
self.create_disposable()
# Service call may wait for session start, give it more time
# than default 5s
kwargs["connect_timeout"] = self.qrexec_timeout
return super().run_service(service, **kwargs)

def cleanup(self):
"""Cleanup after DispVM usage"""
# in 'remote' case nothing is needed, as DispVM is cleaned up
# automatically
"""
Cleanup after disposable usage.

Disposable is cleaned up automatically in 'remote' case.
"""
if (
self.app.qubesd_connection_type == "socket"
and not self._method_dest.startswith("$dispvm")
and not self._method_dest.startswith("@dispvm")
):
try:
self.kill()
except qubesadmin.exc.QubesVMNotRunningError:
pass

def start(self):
"""Start this DispVM"""
# create dispvm
if self._method_dest.startswith("$dispvm"):
if self._method_dest.startswith("$dispvm:"):
method_dest = self._method_dest[len("$dispvm:") :]
"""Create disposable if absent and start it."""
if self._method_dest.startswith("@dispvm"):
self.create_disposable()
super().start()

def create_disposable(self):
"""Create disposable."""
if self._method_dest.startswith("@dispvm"):
if self._method_dest.startswith("@dispvm:"):
method_dest = self._method_dest[len("@dispvm:") :]
else:
method_dest = "dom0"
dispvm = self.app.qubesd_call(
method_dest, "admin.vm.CreateDisposable"
)
dispvm = dispvm.decode("ascii")
self._method_dest = dispvm
super().start()
return self


class DispVM(QubesVM):
Expand All @@ -516,9 +518,9 @@ def from_appvm(cls, app, appvm):
AppVM. If *appvm* is none, use default DispVM template"""

if appvm:
method_dest = "$dispvm:" + str(appvm)
method_dest = "@dispvm:" + str(appvm)
else:
method_dest = "$dispvm"
method_dest = "@dispvm"

wrapper = DispVMWrapper(app, method_dest)
return wrapper