diff --git a/qubes_config/global_config/policy_manager.py b/qubes_config/global_config/policy_manager.py index f25adf76..47fb1b6f 100644 --- a/qubes_config/global_config/policy_manager.py +++ b/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 diff --git a/qubes_config/policy_editor/policy_editor.py b/qubes_config/policy_editor/policy_editor.py index cd91bad4..d59f67bd 100644 --- a/qubes_config/policy_editor/policy_editor.py +++ b/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,11 @@ 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", diff --git a/qubes_config/tests/conftest.py b/qubes_config/tests/conftest.py index 507b94df..fb5bb019 100644 --- a/qubes_config/tests/conftest.py +++ b/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)) diff --git a/qubes_config/tests/test_policy_manager.py b/qubes_config/tests/test_policy_manager.py index 52875ea3..f15d7d7f 100644 --- a/qubes_config/tests/test_policy_manager.py +++ b/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):