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,18 @@
 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:
+    PolicyAdminException = subprocess.CalledProcessError
+    PolicyAdminFileNotFoundException = subprocess.CalledProcessError
 
 import gettext
 
@@ -51,7 +60,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 +92,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

@@ -31,6 +31,15 @@
 from qrexec.policy.parser import StringPolicy
 from qrexec.exc import PolicySyntaxError
 
+try:
+    from qrexec.policy.admin import (
+        PolicyAdminException,
+        PolicyAdminFileNotFoundException,
+    )
+except ImportError:
+    PolicyAdminException = subprocess.CalledProcessError
+    PolicyAdminFileNotFoundException = subprocess.CalledProcessError
+
 from qubes_config.widgets.gtk_utils import (
     load_theme,
     show_error,
@@ -450,9 +459,9 @@ def _save(self, *_args):
             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}: "
+            if hasattr(ex, "stdout"):
                 err_msg += ex.stdout.decode()
             else:
                 err_msg += str(ex)
@@ -533,8 +542,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,16 @@
     MockDevice,
 )
 
+try:
+    from qrexec.policy.admin import (
+        PolicyAdminFileNotFoundException,
+        PolicyAdminTokenException,
+    )
+
+    admin_exc = True
+except ImportError:
+    admin_exc = False
+
 
 @pytest.fixture
 def test_qapp():
@@ -291,6 +301,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 admin_exc:
+            raise PolicyAdminFileNotFoundException
         raise subprocess.CalledProcessError(2, "test")
 
     def policy_include_get(self, file_name):
@@ -300,15 +312,21 @@ def policy_include_get(self, file_name):
                 self.include_files[file_name],
                 self.include_file_tokens[file_name],
             )
+        if admin_exc:
+            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 admin_exc:
+                    raise PolicyAdminTokenException
                 raise subprocess.CalledProcessError(2, "test")
         elif token != "any":
             if token != self.file_tokens.get(filename, ""):
+                if admin_exc:
+                    raise PolicyAdminTokenException
                 raise subprocess.CalledProcessError(2, "test")
         self.files[filename] = policy_text
         self.file_tokens[filename] = str(len(policy_text))
@@ -317,6 +335,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 admin_exc:
+                    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

@@ -25,6 +25,13 @@
 from ..global_config.policy_manager import PolicyManager
 from qrexec.policy.parser import Rule
 
+try:
+    from qrexec.policy.admin import PolicyAdminFileNotFoundException
+
+    admin_exc = True
+except ImportError:
+    admin_exc = False
+
 
 def test_conflict_files():
     def return_files(service_name):
@@ -58,10 +65,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 admin_exc:
+        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 +82,8 @@ def __init__(self):
         def policy_get(self, filename):
             if filename in self.files:
                 return self.files[filename], filename
+            if admin_exc:
+                raise PolicyAdminFileNotFoundException
             raise subprocess.CalledProcessError(2, "test")
 
         def policy_replace(self, filename, text):