Improve exception handling-related logic

core/Errors.py

@@ -1,3 +1,11 @@
+class CommandNotFoundError(OSError):
+
+    def __init__(self, subject):
+
+        self.subject = subject
+        super().__init__(f"Command '{subject}' could not be found.")
+
+
 class UnknownClientVersionError(Exception):
     pass
 

core/controllers/ApplicationController.py

@@ -1,4 +1,5 @@
 from core.Constants import Constants
+from core.Errors import CommandNotFoundError
 from core.controllers.SessionStateController import SessionStateController
 from core.models.session.Application import Application
 from core.models.session.ApplicationVersion import ApplicationVersion
@@ -111,7 +112,10 @@ class ApplicationController:
     @staticmethod
     def __run_process(initialization_file_path, profile, display, session_state):
 
-        virtual_display_process = subprocess.Popen(('/usr/bin/Xephyr', '-ac', '-br', '-title', f'Hydra Veil - {profile.name or "Unnamed Profile"}', '-screen', profile.resolution, '-no-host-grab', display), stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
+        if shutil.which('Xephyr') is None:
+            raise CommandNotFoundError('Xephyr')
+
+        virtual_display_process = subprocess.Popen(('Xephyr', '-ac', '-br', '-title', f'Hydra Veil - {profile.name or "Unnamed Profile"}', '-screen', profile.resolution, '-no-host-grab', display), stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
 
         start_time = time.time()
         timeout = float(10)

core/controllers/ConnectionController.py

@@ -1,5 +1,5 @@
 from core.Constants import Constants
-from core.Errors import InvalidSubscriptionError, MissingSubscriptionError, ConnectionUnprotectedError, ConnectionTerminationError
+from core.Errors import InvalidSubscriptionError, MissingSubscriptionError, ConnectionUnprotectedError, ConnectionTerminationError, CommandNotFoundError
 from core.controllers.ConfigurationController import ConfigurationController
 from core.controllers.ProfileController import ProfileController
 from core.controllers.SessionStateController import SessionStateController
@@ -14,6 +14,7 @@ from typing import Union, Optional
 import os
 import re
 import requests
+import shutil
 import socket
 import subprocess
 import tempfile
@@ -157,11 +158,11 @@ class ConnectionController:
     @staticmethod
     def establish_system_connection(profile: SystemProfile, connection_observer: Optional[ConnectionObserver] = None):
 
-        if subprocess.getstatusoutput('pkexec --help')[0] == 127:
+        if shutil.which('pkexec') is None:
-            raise OSError('The polkit toolkit does not appear to be installed.')
+            raise CommandNotFoundError('pkexec')
 
-        if subprocess.getstatusoutput('wg-quick --help')[0] == 127:
+        if shutil.which('wg-quick') is None:
-            raise OSError('WireGuard tools does not appear to be installed.')
+            raise CommandNotFoundError('wg-quick')
 
         process = subprocess.Popen(('pkexec', 'wg-quick', 'up', profile.get_wireguard_configuration_path()), stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
         completed_successfully = not bool(os.waitpid(process.pid, 0)[1] >> 8)
@@ -205,8 +206,8 @@ class ConnectionController:
     @staticmethod
     def establish_tor_session_connection(session_directory: str, port_number: int):
 
-        if subprocess.getstatusoutput('tor --help')[0] == 127:
+        if shutil.which('tor') is None:
-            raise OSError('Tor does not appear to be installed.')
+            raise CommandNotFoundError('tor')
 
         tor_session_directory = f'{session_directory}/tor'
         Path(tor_session_directory).mkdir(exist_ok=True, mode=0o700)
@@ -236,11 +237,11 @@ class ConnectionController:
     @staticmethod
     def establish_proxy_session_connection(profile: SessionProfile, session_directory: str, port_number: int, proxy_port_number: int):
 
-        if subprocess.getstatusoutput('proxychains4 --help')[0] == 127:
+        if shutil.which('proxychains4') is None:
-            raise OSError('ProxyChains does not appear to be installed.')
+            raise CommandNotFoundError('proxychains4')
 
-        if subprocess.getstatusoutput('microsocks --help')[0] == 127:
+        if shutil.which('microsocks') is None:
-            raise OSError('MicroSocks does not appear to be installed.')
+            raise CommandNotFoundError('microsocks')
 
         if profile.has_proxy_configuration():
             proxy_configuration = profile.get_proxy_configuration()
@@ -275,11 +276,11 @@ class ConnectionController:
     @staticmethod
     def terminate_system_connection(profile: SystemProfile):
 
-        if subprocess.getstatusoutput('pkexec --help')[0] == 127:
+        if shutil.which('pkexec') is None:
-            raise OSError('The polkit toolkit does not appear to be installed.')
+            raise CommandNotFoundError('pkexec')
 
-        if subprocess.getstatusoutput('wg-quick --help')[0] == 127:
+        if shutil.which('wg-quick') is None:
-            raise OSError('WireGuard tools does not appear to be installed.')
+            raise CommandNotFoundError('wg-quick')
 
         process = subprocess.Popen(('pkexec', 'wg-quick', 'down', profile.get_wireguard_configuration_path()), stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
         completed_successfully = not bool(os.waitpid(process.pid, 0)[1] >> 8)
@@ -365,8 +366,8 @@ class ConnectionController:
     @staticmethod
     def system_uses_wireguard_interface():
 
-        if subprocess.getstatusoutput('ip --help')[0] == 127:
+        if shutil.which('ip') is None:
-            raise OSError('The iproute2 utility package does not appear to be installed.')
+            raise CommandNotFoundError('ip')
 
         process = subprocess.Popen(('ip', 'route', 'get', '192.0.2.1'), stdout=subprocess.PIPE)
         process_output = str(process.stdout.read())

core/controllers/ProfileController.py

@@ -216,7 +216,7 @@ class ProfileController:
         from core.controllers.ConnectionController import ConnectionController
 
         if profile.subscription is None:
-            raise InvalidSubscriptionError()
+            raise MissingSubscriptionError()
 
         if profile.location is None:
             raise MissingLocationError()

core/models/system/SystemProfile.py

@@ -1,10 +1,11 @@
 from core.Constants import Constants
-from core.Errors import ProfileDeletionError, ProfileModificationError
+from core.Errors import ProfileDeletionError, ProfileModificationError, CommandNotFoundError
 from core.models.BaseProfile import BaseProfile
 from core.models.system.SystemConnection import SystemConnection
 from dataclasses import dataclass
 from typing import Optional
 import os
+import shutil
 import subprocess
 
 
@@ -17,8 +18,8 @@ class SystemProfile(BaseProfile):
 
     def attach_wireguard_configuration(self, wireguard_configuration):
 
-        if subprocess.getstatusoutput('pkexec --help')[0] == 127:
+        if shutil.which('pkexec') is None:
-            raise OSError('The polkit toolkit does not appear to be installed.')
+            raise CommandNotFoundError('pkexec')
 
         wireguard_configuration_file_backup_path = f'{self.get_config_path()}/wg.conf.bak'
 
@@ -51,8 +52,8 @@ class SystemProfile(BaseProfile):
 
         if self.has_wireguard_configuration():
 
-            if subprocess.getstatusoutput('pkexec --help')[0] == 127:
+            if shutil.which('pkexec') is None:
-                raise OSError('The polkit toolkit does not appear to be installed.')
+                raise CommandNotFoundError('pkexec')
 
             process = subprocess.Popen(('pkexec', 'rm', '-d', self.get_wireguard_configuration_path(), self.get_system_config_path()))
             completed_successfully = not bool(os.waitpid(process.pid, 0)[1] >> 8)