Use new policy admin exceptions

For: QubesOS/qubes-issues#10746
Requires: QubesOS/qubes-core-qrexec#223

Author: ben-grande

qubes_config/global_config/policy_manager.py

@@ -21,9 +21,16 @@
 import subprocess
 from typing import Optional, List, Tuple
 
+from qubes_config.widgets.utils import compare_rule_lists
 from qrexec.policy.admin_client import PolicyClient
 from qrexec.policy.parser import StringPolicy, Rule
-from qubes_config.widgets.utils import compare_rule_lists
+try:
+    from qrexec.policy.admin import (
+        PolicyAdminException,
+        PolicyAdminFileNotFoundException,
+    )
+except ImportError:
+    pass
 
 import gettext
 
@@ -51,7 +58,7 @@ def get_all_policy_files(self, service: str) -> List[str]:
         """Just get a straightforward list of all relevant policy files."""
         try:
             return self.policy_client.policy_get_files(service)
-        except subprocess.CalledProcessError:
+        except (PolicyAdminException, subprocess.CalledProcessError):
             return []
 
     def get_conflicting_policy_files(self, service: str, own_file: str) -> List[str]:
@@ -83,7 +90,7 @@ def get_rules_from_filename(
         for the file."""
         try:
             rules_text, token = self.policy_client.policy_get(filename)
-        except subprocess.CalledProcessError:
+        except (PolicyAdminFileNotFoundException, subprocess.CalledProcessError):
             if not default_policy:
                 return [], None
             rules_text, token = default_policy, None

qubes_config/policy_editor/policy_editor.py

@@ -30,6 +30,13 @@
 from qrexec.policy.admin_client import PolicyClient
 from qrexec.policy.parser import StringPolicy
 from qrexec.exc import PolicySyntaxError
+try:
+    from qrexec.policy.admin import (
+        PolicyAdminException,
+        PolicyAdminFileNotFoundException,
+    )
+except ImportError:
+    pass
 
 from qubes_config.widgets.gtk_utils import (
     load_theme,
@@ -446,13 +453,14 @@ def _new(self, *_args):
 
     def _save(self, *_args):
         """Save changes. If successful, return True."""
+        err_msg = None
         try:
             self.policy_client.policy_replace(
                 self.filename, self.policy_text, self.token
             )
-        except subprocess.CalledProcessError as ex:
-            err_msg = "An error occurred while trying to save the policy file:\n"
-            if ex.stdout:
+        except (PolicyAdminException, subprocess.CalledProcessError) as ex:
+            err_msg = f"Failed to replace policy {self.filename!r}: {ex}"
+            if hasattr(ex, "stdout"):
                 err_msg += ex.stdout.decode()
             else:
                 err_msg += str(ex)
@@ -533,8 +541,8 @@ def open_policy_file(self, name: Optional[str]):
         else:
             try:
                 text, self.token = self.policy_client.policy_get(name)
-            except subprocess.CalledProcessError as ex:
-                if ex.returncode == 126:
+            except (PolicyAdminFileNotFoundException, subprocess.CalledProcessError) as ex:
+                if getattr(ex, "returncode", None) == 126:
                     show_error(
                         self.main_window,
                         "Access denied",

qubes_config/tests/conftest.py

@@ -46,6 +46,14 @@
     MockDevice,
 )
 
+try:
+    from qrexec.policy.admin import (
+        PolicyAdminFileNotFoundException,
+        PolicyAdminTokenException,
+    )
+except ImportError:
+    pass
+
 
 @pytest.fixture
 def test_qapp():
@@ -291,6 +299,8 @@ def policy_get(self, file_name):
         """Get file contents; takes into account policy_replace."""
         if file_name in self.files:
             return self.files[file_name], self.file_tokens[file_name]
+        if "PolicyAdminFileNotFoundException" in globals():
+            raise PolicyAdminFileNotFoundException
         raise subprocess.CalledProcessError(2, "test")
 
     def policy_include_get(self, file_name):
@@ -300,15 +310,21 @@ def policy_include_get(self, file_name):
                 self.include_files[file_name],
                 self.include_file_tokens[file_name],
             )
+        if "PolicyAdminFileNotFoundException" in globals():
+            raise PolicyAdminFileNotFoundException
         raise subprocess.CalledProcessError(2, "test")
 
     def policy_replace(self, filename, policy_text, token="any"):
         """Replace file contents with provided contents."""
         if token == "new":
             if filename in self.file_tokens:
+                if "PolicyAdminTokenException" in globals():
+                    raise PolicyAdminTokenException
                 raise subprocess.CalledProcessError(2, "test")
         elif token != "any":
             if token != self.file_tokens.get(filename, ""):
+                if "PolicyAdminTokenException" in globals():
+                    raise PolicyAdminTokenException
                 raise subprocess.CalledProcessError(2, "test")
         self.files[filename] = policy_text
         self.file_tokens[filename] = str(len(policy_text))
@@ -317,6 +333,8 @@ def policy_include_replace(self, filename, policy_text, token="any"):
         """Replace file contents with provided contents."""
         if token != "any":
             if token != self.include_file_tokens.get(filename, ""):
+                if "PolicyAdminTokenException" in globals():
+                    raise PolicyAdminTokenException
                 raise subprocess.CalledProcessError(2, "test")
         self.include_files[filename] = policy_text
         self.include_file_tokens[filename] = str(len(policy_text))

qubes_config/tests/test_policy_manager.py

@@ -24,6 +24,10 @@
 
 from ..global_config.policy_manager import PolicyManager
 from qrexec.policy.parser import Rule
+try:
+    from qrexec.policy.admin import PolicyAdminFileNotFoundException
+except ImportError:
+    pass
 
 
 def test_conflict_files():
@@ -58,10 +62,14 @@ def test_get_policy_from_file_new_no_default(mock_replace, mock_get):
     manager = PolicyManager()
 
     mock_get.side_effect = subprocess.CalledProcessError(2, "test")
-
     assert manager.get_rules_from_filename("test", "") == ([], None)
     assert not mock_replace.mock_calls
 
+    if "PolicyAdminFileNotFoundException" in globals():
+        mock_get.side_effect = PolicyAdminFileNotFoundException
+        assert manager.get_rules_from_filename("test", "") == ([], None)
+        assert not mock_replace.mock_calls
+
 
 def test_get_policy_from_file_new():
     class MockPolicy:
@@ -71,6 +79,8 @@ def __init__(self):
         def policy_get(self, filename):
             if filename in self.files:
                 return self.files[filename], filename
+            if "PolicyAdminFileNotFoundException" in globals():
+                raise PolicyAdminFileNotFoundException
             raise subprocess.CalledProcessError(2, "test")
 
         def policy_replace(self, filename, text):