Add initial support for privilege policies

core/Constants.py

@@ -37,6 +37,8 @@ class Constants:
     HV_INCIDENT_DATA_HOME: Final[str] = f'{HV_DATA_HOME}/incidents'
     HV_RUNTIME_DATA_HOME: Final[str] = f'{HV_DATA_HOME}/runtime'
 
-    HV_SESSION_STATE_HOME: Final[str] = f'{HV_STATE_HOME}/sessions'
-
     HV_STORAGE_DATABASE_PATH: Final[str] = f'{HV_DATA_HOME}/storage.db'
+
+    HV_PRIVILEGE_POLICY_PATH: Final[str] = f'{SYSTEM_CONFIG_PATH}/sudoers.d/hydra-veil'
+
+    HV_SESSION_STATE_HOME: Final[str] = f'{HV_STATE_HOME}/sessions'

core/Errors.py

@@ -26,6 +26,18 @@ class ConnectionTerminationError(Exception):
     pass
 
 
+class PolicyAssignmentError(Exception):
+    pass
+
+
+class PolicyInstatementError(Exception):
+    pass
+
+
+class PolicyRevocationError(Exception):
+    pass
+
+
 class ProfileDeletionError(Exception):
     pass
 

core/controllers/ConnectionController.py

@@ -2,6 +2,7 @@ from collections.abc import Callable
 from core.Constants import Constants
 from core.Errors import InvalidSubscriptionError, MissingSubscriptionError, ConnectionUnprotectedError, ConnectionTerminationError, CommandNotFoundError
 from core.controllers.ConfigurationController import ConfigurationController
+from core.controllers.PrivilegePolicyController import PrivilegePolicyController
 from core.controllers.ProfileController import ProfileController
 from core.controllers.SessionStateController import SessionStateController
 from core.controllers.SystemStateController import SystemStateController
@@ -11,6 +12,7 @@ from core.models.system.SystemState import SystemState
 from core.observers import ConnectionObserver
 from core.services.WebServiceApiService import WebServiceApiService
 from pathlib import Path
+from subprocess import CalledProcessError
 from typing import Union, Optional, Any
 import os
 import re
@@ -167,50 +169,38 @@ class ConnectionController:
     @staticmethod
     def establish_system_connection(profile: SystemProfile, ignore: tuple[type[Exception]] = (), connection_observer: Optional[ConnectionObserver] = None):
 
-        if shutil.which('pkexec') is None:
-            raise CommandNotFoundError('pkexec')
-
-        if shutil.which('wg-quick') is None:
-            raise CommandNotFoundError('wg-quick')
-
         if ConfigurationController.get_endpoint_verification_enabled():
             ProfileController.verify_wireguard_endpoint(profile, ignore=ignore)
 
-        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)
-
-        if completed_successfully:
-
-            SystemStateController.update_or_create(SystemState(profile.id))
-
         try:
-                ConnectionController.await_connection(connection_observer=connection_observer)
+            ConnectionController.__establish_system_connection(profile, connection_observer)
 
         except ConnectionError:
 
+            try:
                 ConnectionController.terminate_system_connection(profile)
+            except ConnectionTerminationError:
+                pass
+
             raise ConnectionError('The connection could not be established.')
 
-        else:
-
-            ConnectionController.terminate_system_connection(profile)
-
-            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)
-
-            if completed_successfully:
-
-                SystemStateController.update_or_create(SystemState(profile.id))
+        except CalledProcessError:
 
             try:
-                    ConnectionController.await_connection(connection_observer=connection_observer)
-
-                except ConnectionError:
-
                 ConnectionController.terminate_system_connection(profile)
-                    raise ConnectionError('The connection could not be established.')
+            except ConnectionTerminationError:
+                pass
+
+            try:
+                ConnectionController.__establish_system_connection(profile, connection_observer)
+
+            except (ConnectionError, CalledProcessError):
+
+                try:
+                    ConnectionController.terminate_system_connection(profile)
+                except ConnectionTerminationError:
+                    pass
 
-            else:
                 raise ConnectionError('The connection could not be established.')
 
         time.sleep(1.0)
@@ -386,6 +376,46 @@ class ConnectionController:
 
         return bool(re.search('dev wg', str(process_output)))
 
+    @staticmethod
+    def __establish_system_connection(profile: SystemProfile, connection_observer: Optional[ConnectionObserver] = None):
+
+        if shutil.which('wg-quick') is None:
+            raise CommandNotFoundError('wg-quick')
+
+        permission_denied = False
+        return_code = None
+
+        if PrivilegePolicyController.is_instated():
+
+            process = subprocess.Popen(('sudo', '-n', 'wg-quick', 'up', profile.get_wireguard_configuration_path()), stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
+            process.wait()
+
+            return_code = process.returncode
+            permission_denied = return_code != 0 and b'sudo:' in process.stderr.read()
+
+        if not PrivilegePolicyController.is_instated() or permission_denied:
+
+            if shutil.which('pkexec') is None:
+                raise CommandNotFoundError('pkexec')
+
+            process = subprocess.Popen(('pkexec', 'wg-quick', 'up', profile.get_wireguard_configuration_path()), stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
+            process.wait()
+
+            return_code = process.returncode
+
+        if return_code == 0:
+
+            SystemStateController.update_or_create(SystemState(profile.id))
+
+            try:
+                ConnectionController.await_connection(connection_observer=connection_observer)
+
+            except ConnectionError:
+                raise ConnectionError('The connection could not be established.')
+
+        else:
+            raise CalledProcessError(return_code, 'wg-quick')
+
     @staticmethod
     def __with_tor_connection(*args, task: Callable[..., Any], connection_observer: Optional[ConnectionObserver] = None, **kwargs):
 

core/controllers/PrivilegePolicyController.py

@@ -0,0 +1,110 @@
+from core.Constants import Constants
+from core.Errors import CommandNotFoundError, PolicyAssignmentError, PolicyInstatementError, PolicyRevocationError
+from packaging import version
+from packaging.version import InvalidVersion
+from subprocess import CalledProcessError
+import os
+import pwd
+import re
+import shutil
+import subprocess
+
+
+class PrivilegePolicyController:
+
+    @staticmethod
+    def preview():
+
+        username = PrivilegePolicyController.__determine_username()
+        return PrivilegePolicyController.__generate(username)
+
+    @staticmethod
+    def instate():
+
+        if shutil.which('pkexec') is None:
+            raise CommandNotFoundError('pkexec')
+
+        if not PrivilegePolicyController.__is_compatible():
+            raise PolicyInstatementError('The privilege policy is not compatible.')
+
+        username = PrivilegePolicyController.__determine_username()
+        privilege_policy = PrivilegePolicyController.__generate(username)
+
+        completed_successfully = False
+        failed_attempt_count = 0
+
+        while not completed_successfully and failed_attempt_count < 3:
+
+            process = subprocess.Popen([
+                'pkexec', 'install', '/dev/stdin', Constants.HV_PRIVILEGE_POLICY_PATH, '-o', 'root', '-m', '440'
+            ], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+
+            process.communicate(f'{privilege_policy}\n')
+            completed_successfully = (process.returncode == 0)
+
+            if not completed_successfully:
+                failed_attempt_count += 1
+
+        if not completed_successfully:
+            raise PolicyInstatementError('The privilege policy could not be instated.')
+
+    @staticmethod
+    def revoke():
+
+        if shutil.which('pkexec') is None:
+            raise CommandNotFoundError('pkexec')
+
+        process = subprocess.Popen(('pkexec', 'rm', Constants.HV_PRIVILEGE_POLICY_PATH))
+        completed_successfully = not bool(os.waitpid(process.pid, 0)[1] >> 8)
+
+        if not completed_successfully:
+            raise PolicyRevocationError('The privilege policy could not be revoked.')
+
+    @staticmethod
+    def is_instated():
+        return os.path.exists(Constants.HV_PRIVILEGE_POLICY_PATH)
+
+    @staticmethod
+    def __determine_username():
+
+        try:
+            password_database_entry = pwd.getpwuid(os.geteuid())
+        except (OSError, KeyError):
+            raise PolicyAssignmentError('The privilege policy could not be assigned to the current user.')
+
+        if password_database_entry.pw_uid == 0:
+            raise PolicyAssignmentError('The privilege policy could not be assigned to the current user.')
+
+        return password_database_entry.pw_name
+
+    @staticmethod
+    def __generate(username: str):
+
+        return '\n'.join([
+            f'{username} ALL=(root) NOPASSWD: /usr/bin/wg-quick ^up {Constants.HV_SYSTEM_PROFILE_CONFIG_PATH}/[0-9]+/wg.conf$'
+        ])
+
+    @staticmethod
+    def __is_compatible():
+
+        try:
+            process_output = subprocess.check_output(['sudo', '-V'], text=True)
+        except CalledProcessError:
+            return False
+
+        if process_output.splitlines():
+            sudo_version_details = process_output.splitlines()[0].strip()
+        else:
+            return False
+
+        sudo_version_number = (m := re.search(r'(\d[0-9.]+?)(?=p|$)', sudo_version_details)) and m.group(1)
+
+        if not sudo_version_number:
+            return False
+
+        try:
+            sudo_version = version.parse(sudo_version_number)
+        except InvalidVersion:
+            return False
+
+        return sudo_version >= version.parse('1.9.10') and os.path.isfile('/usr/bin/wg-quick')