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

Singbox Protocols can be enabled via the main ProfileController

Browse source
This commit is contained in:
SimplifiedPrivacy 2026-08-10 18:55:23 -04:00
parent c3b50460cd
commit c291517b7f
5 changed files with 26 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
Untitled Document
View file

@ -1,20 +0,0 @@
from core.observers.ConnectionObserver import ConnectionObserver
from core.services.networking.systemwide.encrypted_proxy.get_proxy_data import post_operator_proxy
import json
observer = ConnectionObserver()
observer.subscribe("connecting", lambda msg: print(f"[CONNECT] {msg}"))
observer.subscribe("tor_bootstrapping", lambda msg: print(f"[BOOTSTRAP] {msg}"))
observer.subscribe("tor_bootstrap_progressing", lambda msg: print(f"[PROGRESS] {msg}"))
observer.subscribe("tor_bootstrapped", lambda msg: print(f"[SUCCESS] {msg}"))
observer.subscribe("custom_message", lambda msg: print(f"[INFO] {msg}"))
results = post_operator_proxy(
billing_code="2UEN-SXAJ-Z1YO-93EJ",
location_id=4,
protocol="hysteria2",
connection_observer=observer
)
print(results)

20
core/controllers/ProfileController.py
View file

@ -2,6 +2,7 @@ from core.services.networking.systemwide.systemwide_wireguard import terminate_s
from core.services.networking.general_connection_tools.testing_evaluating import system_uses_wireguard_interface
from core.services.networking.general_connection_tools.connection_enable import establish_connection
from core.services.networking.systemwide.systemwide_utils import get_firewall_setting, get_dns_setting
from core.services.networking.systemwide.encrypted_proxy.singbox_runner import end_singbox
from core.services.networking.systemwide import killswitch
from core.services.subscriptions import subscriptions
from core.errors.exceptions import FirewallError
@ -156,10 +157,15 @@ class ProfileController:
firewall_setting = get_firewall_setting()
dns_setting = get_dns_setting()
# ======= KILL SYSTEMWIDE =============
terminate_system_connection(
firewall_setting=firewall_setting,
dns_setting=dns_setting
)
if profile.connection.code == "wireguard":
terminate_system_connection(
firewall_setting=firewall_setting,
dns_setting=dns_setting
)
elif profile.connection.code in ("hysteria2", "vless"):
end_singbox()
else:
raise ProfileDeactivationError('Unsupported protocol.')
# ================= UPDATE UI ================
# if it made it this far, it worked in theory.
if profile_observer is not None:
@ -173,12 +179,6 @@ class ProfileController:
except FirewallError:
raise
# if profile_observer is not None:
# profile_observer.notify('disabled', profile, dict(
# explicitly=explicitly,
# ))
time.sleep(1.0)
@staticmethod

4
core/services/networking/general_connection_tools/connection_enable.py
View file

@ -3,6 +3,7 @@ from core.services.networking.systemwide.systemwide_wireguard import establish_s
from core.services.keys_and_verifications.wireguard_keys import register_wireguard_session
from core.services.subscriptions.subscriptions import activate_subscription, get_encrypted_proxy_billing_code
from core.services.networking.systemwide.encrypted_proxy.configure_singbox import configure_singbox
from core.services.networking.systemwide.encrypted_proxy.singbox_runner import start_singbox
from core.models.Result import Result, ResultError
from core.utils.basic_operations.write_or_read_from_json import get_value_from_json_file
@ -86,9 +87,10 @@ def launch_encrypted_proxy(
connection_observer=connection_observer
)
logger.info(f"We have a server IP now of {server_ip}")
return start_singbox(
profile_id=profile.id,
config_path=config_path,
server_ip=server_ip,
connection_observer=connection_observer
)

16
core/services/networking/systemwide/encrypted_proxy/singbox_runner.py
View file

@ -99,7 +99,11 @@ def _attempt_start_with_retry(profile_id: int, quantity_of_attempts: int = 2) ->
# end of loop / make sure it's not running already..
error_msg = f"Singbox failed to launch after {quantity_of_attempts} attempts"
logger.error(f"{error_msg}, let's double check it's not running already:")
double_check = pid_tools.get_pid_by_app_name(exact_app_name="sing-box")
try:
double_check = pid_tools.get_pid_by_app_name(exact_app_name="sing-box")
except Exception as e:
return Result(valid=False, error_type=PROCESS_WONT_START, message=str(e))
if double_check.valid:
return double_check
else:
@ -107,9 +111,7 @@ def _attempt_start_with_retry(profile_id: int, quantity_of_attempts: int = 2) ->
return Result(valid=False, error_type=PROCESS_WONT_START, message=error_msg)
def end_singbox(
connection_observer: Optional[ConnectionObserver] = None
) -> Result:
def end_singbox() -> Result:
# get by systemstate:
current_state = SystemState.get()
@ -196,11 +198,13 @@ def start_singbox(
for each_requirement in requirements:
if each_requirement is None:
return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=f"Invalid inputs into {function_name}")
logger.info("All requirements met.")
# ========== KILL IT IF ALREADY UP ==========
killed_pre_existing = end_singbox(connection_observer)
killed_pre_existing = end_singbox()
if not killed_pre_existing.valid:
return killed_pre_existing
logger.info("Pre-existing process ended.")
# ============= START BINARY =============
launched = _attempt_start_with_retry(profile_id=profile_id, quantity_of_attempts=2)
@ -208,6 +212,7 @@ def start_singbox(
return launched
process_id = launched.data
logger.info(f"Binary started with process id {process_id}.")
# ============= CHECK INTERFACE =============
interface_result = interface_tools.get_output(Constants.SINGBOX_TUN_IF)
@ -237,6 +242,7 @@ def start_singbox(
dns_set=True, # intended setting, not result yet
process_id=process_id
)
logger.info(f"State is created.")
# ============= CONFIRM IP ADDRESS =============
ip_result = ip.test(connection_observer)

2
core/utils/basic_operations/pid_tools.py
View file

@ -29,7 +29,7 @@ def is_running(pid: int, process_name: str) -> bool:
# Function is public because it works for any app
def get_pid_by_app_name(exact_app_name: str) -> Result:
command = ['pgrep', '-x', exact_app_name]
human_readable_goal = "Get a pid by an exact match"
human_readable_goal = "Get a pid by an exact match get_pid_by_app_name"
result = run_generic_command(command, human_readable_goal, timeout=7)
if result.valid: # equivalent: (returncode == 0):