Vless is now working. Config Parsing and setup is stable, and increased the wait time for connection testing. But this is a temporary solution. The real answer is not a static time check, but a dynamic result reading for when to test. Which then has it's own timeout time

This commit is contained in:
SimplifiedPrivacy 2026-08-20 10:18:16 -04:00
parent c8811c93bb
commit 5923cc8a7f
5 changed files with 31 additions and 13 deletions

View file

@ -14,13 +14,12 @@ def save_to_sudo_folder(model: BaseModel, filepath: str) -> bool:
Save model to a sudo-protected file using pkexec. Save model to a sudo-protected file using pkexec.
""" """
try: try:
Path(filepath).parent.mkdir(parents=True, exist_ok=True)
json_data = model.model_dump_json(indent=2) json_data = model.model_dump_json(indent=2)
parent_dir = str(Path(filepath).parent)
# Use pkexec + tee to write with elevated privileges # Single pkexec call: mkdir + write
process = subprocess.Popen( process = subprocess.Popen(
('pkexec', 'tee', filepath), ('pkexec', 'bash', '-c', f'mkdir -p "{parent_dir}" && tee "{filepath}"'),
stdin=subprocess.PIPE, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, stderr=subprocess.PIPE,

View file

@ -98,12 +98,16 @@ def configure_singbox(
# parse the link: # parse the link:
vless = parse_vless_link(validated_data.links) vless = parse_vless_link(validated_data.links)
logger.info("parsed vlink link!")
# use the parsed link: # use the parsed link:
real_config = build_vless_config( real_config = build_vless_config(
vless=vless, vless=vless,
socks5_port=random_port, socks5_port=random_port,
server_ip=validated_data.server_ip server_ip=validated_data.server_ip
) )
if real_config:
logger.info("Got the real vless config")
# UNKNOWN # UNKNOWN
else: else:
@ -114,6 +118,7 @@ def configure_singbox(
# SAVE REAL CONFIG # SAVE REAL CONFIG
################################### ###################################
# goes in a sudo protected folder & prompts for password: # goes in a sudo protected folder & prompts for password:
logger.info("Saving to sudo folder.")
saved_it = attach_config_to_sudo_folder( saved_it = attach_config_to_sudo_folder(
config_data=real_config, config_data=real_config,
sudo_filepath=profile_sudo_filepath, sudo_filepath=profile_sudo_filepath,
@ -169,18 +174,27 @@ def attach_config_to_sudo_folder(
if shutil.which('pkexec') is None: if shutil.which('pkexec') is None:
raise CommandNotFoundError('pkexec') raise CommandNotFoundError('pkexec')
logger.info("About to save to backup config path")
backup_path = f'{backup_folder}/backup.conf.bak' backup_path = f'{backup_folder}/backup.conf.bak'
with open(backup_path, "w") as configuration_file: try:
json.dump(config_data, configuration_file, indent=4) with open(backup_path, "w") as configuration_file:
json.dump(config_data, configuration_file, indent=4)
logger.info("Saved to backup config path!")
except Exception as e:
logger.error(e)
return False
configuration_is_attached = False configuration_is_attached = False
failed_attempt_count = 0 failed_attempt_count = 0
while not configuration_is_attached and failed_attempt_count < 3: while not configuration_is_attached and failed_attempt_count < 3:
logger.info(f"Trying attempt {failed_attempt_count} to write to the sudo folder..")
process = subprocess.Popen(('pkexec', 'install', '-D', backup_path, sudo_filepath, '-o', 'root', '-m', '744')) try:
configuration_is_attached = not bool(os.waitpid(process.pid, 0)[1] >> 8) process = subprocess.Popen(('pkexec', 'install', '-D', backup_path, sudo_filepath, '-o', 'root', '-m', '744'))
configuration_is_attached = not bool(os.waitpid(process.pid, 0)[1] >> 8)
except Exception as e:
logger.error(e)
if not configuration_is_attached: if not configuration_is_attached:
failed_attempt_count += 1 failed_attempt_count += 1

View file

@ -61,8 +61,8 @@ def launch_singbox_binary(profile_id: int) -> Result:
process_id = int(activation_result.data) process_id = int(activation_result.data)
logger.info(f"Waiting 2 seconds to see if the process id {process_id} is still alive..") logger.info(f"Waiting 6 seconds to see if the process id {process_id} is still alive..")
time.sleep(2) time.sleep(6)
# Evaluate if running by that exact pid: # Evaluate if running by that exact pid:
active = pid_tools.is_running(pid=process_id, process_name="sing-box") active = pid_tools.is_running(pid=process_id, process_name="sing-box")

View file

@ -57,7 +57,7 @@ def build_vless_config(vless: dict, socks5_port: int, server_ip: str) -> dict:
{ {
"type": "vless", "type": "vless",
"tag": "proxy", "tag": "proxy",
"server": server_ip, "server": str(server_ip),
"server_port": vless["port"], "server_port": vless["port"],
"uuid": vless["uuid"], "uuid": vless["uuid"],
"tls": { "tls": {

View file

@ -107,6 +107,11 @@ def get_value_from_json_file(filepath: str, key: str, category: str | None = Non
data = read_entire_json(filepath) data = read_entire_json(filepath)
if category: if category:
if isinstance(category, bool):
error_msg = f"Error in get_value_from_json_file, category should not be a boolean: {category}"
logger.error(error_msg)
raise ValueError(error_msg)
if category not in data: if category not in data:
raise KeyError(f"Category '{category}' not found in JSON file") raise KeyError(f"Category '{category}' not found in JSON file")
subcategory = data[category] subcategory = data[category]