Support/sp-hydra-veil-core
Support/sp-hydra-veil-core
2
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Wireguard now checks if it needs to disable singbox based on systemstate. Also eliminated content from profilecontroller that was no longer being used for subscriptions, and causing circular import issues

Browse source
This commit is contained in:
SimplifiedPrivacy 2026-10-15 19:30:34 -04:00
parent 0931200c79
commit 5dee318aa4
2 changed files with 39 additions and 36 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

57
core/controllers/ProfileController.py
View file

@ -1,5 +1,5 @@
from core.services.profiles.ProfileRepository import ProfileRepository
from core.services.subscriptions import subscriptions
# from core.services.subscriptions import subscriptions
from core.models.Result import Result, ResultError
from core.errors.logger import logger
@ -69,47 +69,44 @@ class ProfileController:
ProfileController._repo.invalidate(profile_id=profile_id)
@staticmethod
def attach_subscription(profile: Union[SessionProfile, SystemProfile], subscription: Subscription):
# @staticmethod
# def attach_subscription(profile: Union[SessionProfile, SystemProfile], subscription: Subscription):
profile.subscription = subscription
profile.save()
# profile.subscription = subscription
# profile.save()
@staticmethod
def activate_subscription(profile: Union[SessionProfile, SystemProfile], connection_observer: Optional[ConnectionObserver] = None):
# @staticmethod
# def activate_subscription(profile: Union[SessionProfile, SystemProfile], connection_observer: Optional[ConnectionObserver] = None):
from core.controllers.ConnectionController import ConnectionController
# from core.controllers.ConnectionController import ConnectionController
if profile.has_subscription():
# if profile.has_subscription():
subscription = subscriptions.get_subscription(
billing_code=profile.subscription.billing_code,
connection_observer=connection_observer
)
# subscription = subscriptions.get_subscription(
# billing_code=profile.subscription.billing_code,
# connection_observer=connection_observer
# )
# legacy:
# subscription = ConnectionController.with_preferred_connection(profile.subscription.billing_code, task=WebServiceApiService.get_subscription, connection_observer=connection_observer)
# if subscription is not None:
if subscription is not None:
# profile.subscription = subscription
# profile.save()
profile.subscription = subscription
profile.save()
# else:
# raise InvalidSubscriptionError()
else:
raise InvalidSubscriptionError()
else:
raise MissingSubscriptionError()
# else:
# raise MissingSubscriptionError()
@staticmethod
def get_invoice(profile: Union[SessionProfile, SystemProfile]):
# @staticmethod
# def get_invoice(profile: Union[SessionProfile, SystemProfile]):
if profile.has_subscription():
# return subscriptions.get_invoice(billing_code=profile.subscription.billing_code)
return WebServiceApiService.get_invoice(profile.subscription.billing_code)
else:
return None
# if profile.has_subscription():
# # return subscriptions.get_invoice(billing_code=profile.subscription.billing_code)
# return WebServiceApiService.get_invoice(profile.subscription.billing_code)
# else:
# return None
@staticmethod
def attach_proxy_configuration(profile: Union[SessionProfile, SystemProfile]):

18
core/services/networking/systemwide/systemwide_wireguard.py
View file

@ -1,4 +1,5 @@
from core.models.enum_models import ProtocolChoice
from core.models.enum_models import ProtocolChoice, SINGBOX_PROTOCOLS
from core.controllers.profile_state.get_profile_data import get_protocol_enum
from core.services.networking.general_connection_tools.testing_evaluating import await_connection, system_uses_wireguard_interface, terminate_tor_connection
from core.services.keys_and_verifications.endpoint_verification import verify_wireguard_endpoint
from core.models.Result import Result, ResultError
@ -57,15 +58,20 @@ def terminate_system_connection(
# ====== NMCLI =======
if SystemStateController.exists():
# ====== SINGBOX =======
# Check if the user is switching from Singbox to Wireguard:
current_state = SystemState.get()
if get_protocol_enum(current_state.profile_id) in SINGBOX_PROTOCOLS:
logger.info("Ending Singbox protocols first..")
from core.services.networking.systemwide.encrypted_proxy.singbox_runner import end_singbox
return end_singbox()
# ====== FINISH WIREGUARD =======
command = ['nmcli', 'connection', 'delete', 'wg']
human_readable_goal = "Deleting NMCLI wg interface"
delete_completed = run_generic_command(command, human_readable_goal, timeout=8)
# legacy version:
# process = subprocess.Popen((), stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
# completed_successfully = not bool(os.waitpid(process.pid, 0)[1] >> 8)
# if completed_successfully or not system_uses_wireguard_interface():
if delete_completed.valid or not system_uses_wireguard_interface():
command = ['nmcli', 'connection', 'delete', 'hv-ipv6-sink']
human_readable_goal = "Deleting NMCLI IPv6 Sink"