Add support for conditional credential preservation

This commit is contained in:
codeking 2025-09-09 02:09:57 +02:00
parent 3f1f5d5633
commit 5e4bce542a
2 changed files with 40 additions and 5 deletions

View file

@ -6,6 +6,7 @@ from core.models.session.ProxyConfiguration import ProxyConfiguration
from core.models.session.SessionConnection import SessionConnection from core.models.session.SessionConnection import SessionConnection
from dataclasses import dataclass from dataclasses import dataclass
from json import JSONDecodeError from json import JSONDecodeError
from pathlib import Path
from typing import Optional from typing import Optional
import json import json
import os import os
@ -31,6 +32,11 @@ class SessionProfile(BaseProfile):
if os.path.isdir(persistent_state_path): if os.path.isdir(persistent_state_path):
shutil.rmtree(persistent_state_path, ignore_errors=True) shutil.rmtree(persistent_state_path, ignore_errors=True)
if 'location' in self._get_dirty_keys():
self.__delete_proxy_configuration()
self.__delete_wireguard_configuration()
super().save() super().save()
def attach_proxy_configuration(self, proxy_configuration): def attach_proxy_configuration(self, proxy_configuration):
@ -122,3 +128,9 @@ class SessionProfile(BaseProfile):
if match: if match:
return re.sub(r'[^a-zA-Z0-9+\-_ /]', '', match.group(1).strip()) return re.sub(r'[^a-zA-Z0-9+\-_ /]', '', match.group(1).strip())
def __delete_proxy_configuration(self):
Path(self.get_proxy_configuration_path()).unlink(missing_ok=True)
def __delete_wireguard_configuration(self):
Path(self.get_wireguard_configuration_path()).unlink(missing_ok=True)

View file

@ -16,6 +16,13 @@ class SystemProfile(BaseProfile):
def get_system_config_path(self): def get_system_config_path(self):
return self.__get_system_config_path(self.id) return self.__get_system_config_path(self.id)
def save(self):
if 'location' in self._get_dirty_keys():
self.__delete_wireguard_configuration()
super().save()
def attach_wireguard_configuration(self, wireguard_configuration): def attach_wireguard_configuration(self, wireguard_configuration):
if shutil.which('pkexec') is None: if shutil.which('pkexec') is None:
@ -40,7 +47,7 @@ class SystemProfile(BaseProfile):
failed_attempt_count += 1 failed_attempt_count += 1
if not wireguard_configuration_is_attached: if not wireguard_configuration_is_attached:
raise ProfileModificationError('WireGuard configuration could not be attached.') raise ProfileModificationError('The WireGuard configuration could not be attached.')
def get_wireguard_configuration_path(self): def get_wireguard_configuration_path(self):
return f'{self.get_system_config_path()}/wg.conf' return f'{self.get_system_config_path()}/wg.conf'
@ -50,18 +57,34 @@ class SystemProfile(BaseProfile):
def delete(self): def delete(self):
try:
self.__delete_wireguard_configuration()
except ProfileModificationError:
raise ProfileDeletionError('The WireGuard configuration could not be deleted.')
if shutil.which('pkexec') is None:
raise CommandNotFoundError('pkexec')
process = subprocess.Popen(('pkexec', 'rm', '-d', self.get_system_config_path()))
completed_successfully = not bool(os.waitpid(process.pid, 0)[1] >> 8)
if not completed_successfully:
raise ProfileDeletionError('The profile could not be deleted.')
super().delete()
def __delete_wireguard_configuration(self):
if self.has_wireguard_configuration(): if self.has_wireguard_configuration():
if shutil.which('pkexec') is None: if shutil.which('pkexec') is None:
raise CommandNotFoundError('pkexec') raise CommandNotFoundError('pkexec')
process = subprocess.Popen(('pkexec', 'rm', '-d', self.get_wireguard_configuration_path(), self.get_system_config_path())) process = subprocess.Popen(('pkexec', 'rm', '-d', self.get_wireguard_configuration_path()))
completed_successfully = not bool(os.waitpid(process.pid, 0)[1] >> 8) completed_successfully = not bool(os.waitpid(process.pid, 0)[1] >> 8)
if not completed_successfully: if not completed_successfully:
raise ProfileDeletionError('The profile could not be deleted.') raise ProfileModificationError('The WireGuard configuration could not be deleted.')
super().delete()
@staticmethod @staticmethod
def __get_system_config_path(id: int): def __get_system_config_path(id: int):