Allow named disposables in OpenInVM config

Author: Jayant-kernel

qubes_config/global_config/disposables.py

@@ -20,6 +20,7 @@
 """
 Disposables page handler
 """
+
 import subprocess
 from enum import Enum
 
@@ -442,6 +443,7 @@ def __init__(
             policy_file_name="50-config-openinvm",
             prefix="openinvm",
             policy_manager=self.policy_manager,
+            target_filter=self._permanent_dispvm_and_dispvm_template_filter,
         )
         self.openurl_handler = DispvmExceptionHandler(
             gtk_builder=gtk_builder,
@@ -450,6 +452,7 @@ def __init__(
             policy_file_name="50-config-openurl",
             prefix="url",
             policy_manager=self.policy_manager,
+            target_filter=self._permanent_dispvm_and_dispvm_template_filter,
         )
         self.handlers: list[PageHandler | PropertyHandler] = [
             self.defdispvm_handler,
@@ -463,6 +466,16 @@ def __init__(
     def _default_dispvm_filter(vm) -> bool:
         return getattr(vm, "template_for_dispvms", False)
 
+    @staticmethod
+    def _permanent_dispvm_and_dispvm_template_filter(vm) -> bool:
+        if getattr(vm, "template_for_dispvms", False):
+            return True
+
+        return (
+            getattr(vm, "klass", None) == "DispVM"
+            and getattr(vm, "auto_cleanup", False) is not True
+        )
+
     def get_unsaved(self) -> str:
         """Get list of unsaved changes."""
         unsaved = []

qubes_config/global_config/policy_exceptions_handler.py

@@ -218,6 +218,7 @@ def __init__(
         prefix: str,
         policy_manager,
         policy_file_name: str,
+        target_filter: Optional[Callable[[qubesadmin.vm.QubesVM], bool]] = None,
     ):
         """
         Handler for various dispvm-related exception lists
@@ -239,6 +240,7 @@ def __init__(
         self.service_name = service_name
         self.policy_manager = policy_manager
         self.policy_file_name = policy_file_name
+        self.target_filter = target_filter
 
         self.list_handler = PolicyExceptionsHandler(
             gtk_builder=gtk_builder,
@@ -282,6 +284,7 @@ def _get_row(self, rule: Rule, new: bool = False):
                 }
             ),
             is_new_row=new,
+            filter_target=self.target_filter,
         )
 
     def _new_rule(self) -> Rule:

qubes_config/global_config/rule_list_widgets.py

@@ -18,6 +18,7 @@
 # You should have received a copy of the GNU Lesser General Public License along
 # with this program; if not, see <http://www.gnu.org/licenses/>.
 """Widgets used by various list of policy rules."""
+
 from typing import Optional, Dict, Callable, Any
 
 from ..widgets.gtk_widgets import (
@@ -667,14 +668,17 @@ def __init__(
         qapp: qubesadmin.Qubes,
         verb_description: Optional[AbstractVerbDescription] = None,
         is_new_row: bool = False,
+        filter_target: Optional[Callable] = None,
     ):
         super().__init__(
             parent_handler=parent_handler,
             rule=rule,
             qapp=qapp,
             verb_description=verb_description,
             initial_verb=_("will"),
-            filter_target=self._dvm_template_filter,
+            filter_target=(
+                filter_target or self._permanent_dispvm_and_dispvm_template_filter
+            ),
             is_new_row=is_new_row,
             source_categories=SOURCE_CATEGORIES,
             target_categories=DISPVM_CATEGORIES,
@@ -685,8 +689,14 @@ def __init__(
             self.target_widget.hide_selectors()
 
     @staticmethod
-    def _dvm_template_filter(vm: qubesadmin.vm.QubesVM):
-        return getattr(vm, "template_for_dispvms", False)
+    def _permanent_dispvm_and_dispvm_template_filter(vm: qubesadmin.vm.QubesVM):
+        if getattr(vm, "template_for_dispvms", False):
+            return True
+
+        return (
+            getattr(vm, "klass", None) == "DispVM"
+            and getattr(vm, "auto_cleanup", False) is not True
+        )
 
     def _hide_target_on_deny(self):
         if self.action_widget.get_selected() == "deny":

qubes_config/tests/test_disposables.py

@@ -202,6 +202,43 @@ def test_dvm_list(real_builder, test_qapp, test_policy_manager):
     assert found_default
 
 
+def test_disposables_exception_rows_include_named_disposables(
+    real_builder, test_qapp, test_policy_manager
+):
+    test_qapp._qubes["volatile-disp"] = MockQube(
+        name="volatile-disp",
+        qapp=test_qapp,
+        klass="DispVM",
+        template="default-dvm",
+    )
+    test_qapp._qubes["volatile-disp"].properties["auto_cleanup"].value = True
+    test_qapp.update_vm_calls()
+
+    disposables_handler = DisposablesHandler(
+        test_qapp, test_policy_manager, real_builder
+    )
+
+    default_dvm = test_qapp.domains["default-dvm"]
+    named_dispvm = test_qapp.domains["test-disp"]
+    volatile_dispvm = test_qapp.domains["volatile-disp"]
+    regular_appvm = test_qapp.domains["test-vm"]
+
+    assert disposables_handler.defdispvm_handler.model.is_vm_available(default_dvm)
+    assert not disposables_handler.defdispvm_handler.model.is_vm_available(named_dispvm)
+
+    for handler in (
+        disposables_handler.open_in_dvm_handler,
+        disposables_handler.openurl_handler,
+    ):
+        handler.list_handler.add_button.clicked()
+        row = next(row for row in handler.list_handler.current_rows if row.editing)
+
+        assert row.target_widget.model.is_vm_available(default_dvm)
+        assert row.target_widget.model.is_vm_available(named_dispvm)
+        assert not row.target_widget.model.is_vm_available(volatile_dispvm)
+        assert not row.target_widget.model.is_vm_available(regular_appvm)
+
+
 def test_dispvm_change(real_builder, test_qapp, test_policy_manager):
     disposables_handler = DisposablesHandler(
         test_qapp, test_policy_manager, real_builder

qubes_config/tests/test_rule_list_widgets.py

@@ -21,15 +21,21 @@
 # pylint: disable=missing-function-docstring
 # pylint: disable=missing-class-docstring
 from unittest.mock import Mock, patch
+from qubesadmin.tests.mock_app import MockQube
 from qrexec.policy.parser import Rule
 from ..global_config.policy_handler import PolicyHandler
-from ..global_config.policy_rules import RuleSimple, SimpleVerbDescription
+from ..global_config.policy_rules import (
+    RuleDispVM,
+    RuleSimple,
+    SimpleVerbDescription,
+)
 from ..global_config.rule_list_widgets import (
     VMWidget,
     ActionWidget,
     RuleListBoxRow,
     NoActionListBoxRow,
     FilteredListBoxRow,
+    DispvmRuleRow,
 )
 
 import gi
@@ -298,6 +304,56 @@ def test_limited_selection_row(test_qapp):
     assert not rule_row.target_widget.model.is_vm_available(vm_vault)
 
 
+def test_dispvm_rule_row_includes_named_disposables(test_qapp):
+    mock_handler = Mock(spec=PolicyHandler)
+    mock_handler.verify_new_rule.return_value = None
+    rule = RuleDispVM(
+        Rule.from_line(
+            None,
+            "Service * test-blue @dispvm allow target=@dispvm",
+            filepath=None,
+            lineno=0,
+        )
+    )
+
+    rule_row = DispvmRuleRow(parent_handler=mock_handler, rule=rule, qapp=test_qapp)
+
+    default_dvm = test_qapp.domains["default-dvm"]
+    named_dispvm = test_qapp.domains["test-disp"]
+    regular_appvm = test_qapp.domains["test-vm"]
+
+    assert rule_row.target_widget.model.is_vm_available(default_dvm)
+    assert rule_row.target_widget.model.is_vm_available(named_dispvm)
+    assert not rule_row.target_widget.model.is_vm_available(regular_appvm)
+
+
+def test_dispvm_rule_row_excludes_volatile_disposables(test_qapp):
+    mock_handler = Mock(spec=PolicyHandler)
+    mock_handler.verify_new_rule.return_value = None
+    test_qapp._qubes["volatile-disp"] = MockQube(
+        name="volatile-disp",
+        qapp=test_qapp,
+        klass="DispVM",
+        template="default-dvm",
+        auto_cleanup=True,
+    )
+    test_qapp.update_vm_calls()
+    rule = RuleDispVM(
+        Rule.from_line(
+            None,
+            "Service * test-blue @dispvm allow target=@dispvm",
+            filepath=None,
+            lineno=0,
+        )
+    )
+
+    rule_row = DispvmRuleRow(parent_handler=mock_handler, rule=rule, qapp=test_qapp)
+
+    assert not rule_row.target_widget.model.is_vm_available(
+        test_qapp.domains["volatile-disp"]
+    )
+
+
 def test_rule_row_init_delete(test_qapp):
     mock_handler = Mock(spec=PolicyHandler)
     mock_handler.verify_new_rule.return_value = None