bivashy/Blitz-Proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace " with '

Browse Source 
ODC
This commit is contained in:
Iam54r1n4
2025-01-26 08:12:56 +00:00
parent 5a999e54fb
commit bcb985c32c
1 changed files with 75 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

146
core/cli_api.py
View File

@ -44,27 +44,27 @@ class Command(Enum):
class HysteriaError(Exception): class HysteriaError(Exception):
"""Base class for Hysteria-related exceptions.""" '''Base class for Hysteria-related exceptions.'''
pass pass
class CommandExecutionError(HysteriaError): class CommandExecutionError(HysteriaError):
"""Raised when a command execution fails.""" '''Raised when a command execution fails.'''
pass pass
class InvalidInputError(HysteriaError): class InvalidInputError(HysteriaError):
"""Raised when the provided input is invalid.""" '''Raised when the provided input is invalid.'''
pass pass
class PasswordGenerationError(HysteriaError): class PasswordGenerationError(HysteriaError):
"""Raised when password generation fails.""" '''Raised when password generation fails.'''
pass pass
class ScriptNotFoundError(HysteriaError): class ScriptNotFoundError(HysteriaError):
"""Raised when a required script is not found.""" '''Raised when a required script is not found.'''
pass pass
# region Utils # region Utils
@ -84,7 +84,7 @@ def run_cmd(command: list[str]) -> str | None:
return result return result
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
if DEBUG: if DEBUG:
raise CommandExecutionError(f"Command execution failed: {e}\nOutput: {e.output.decode()}") raise CommandExecutionError(f'Command execution failed: {e}\nOutput: {e.output.decode()}')
else: else:
return None return None
return None return None
@ -98,7 +98,7 @@ def generate_password() -> str:
try: try:
return subprocess.check_output(['pwgen', '-s', '32', '1'], shell=False).decode().strip() return subprocess.check_output(['pwgen', '-s', '32', '1'], shell=False).decode().strip()
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
raise PasswordGenerationError(f"Failed to generate password: {e}") raise PasswordGenerationError(f'Failed to generate password: {e}')
# endregion # endregion
@ -108,43 +108,43 @@ def generate_password() -> str:
def install_hysteria2(port: int, sni: str): def install_hysteria2(port: int, sni: str):
""" '''
Installs Hysteria2 on the given port and uses the provided or default SNI value. Installs Hysteria2 on the given port and uses the provided or default SNI value.
""" '''
run_cmd(['bash', Command.INSTALL_HYSTERIA2.value, str(port), sni]) run_cmd(['bash', Command.INSTALL_HYSTERIA2.value, str(port), sni])
def uninstall_hysteria2(): def uninstall_hysteria2():
"""Uninstalls Hysteria2.""" '''Uninstalls Hysteria2.'''
run_cmd(['bash', Command.UNINSTALL_HYSTERIA2.value]) run_cmd(['bash', Command.UNINSTALL_HYSTERIA2.value])
def update_hysteria2(): def update_hysteria2():
"""Updates Hysteria2.""" '''Updates Hysteria2.'''
run_cmd(['bash', Command.UPDATE_HYSTERIA2.value]) run_cmd(['bash', Command.UPDATE_HYSTERIA2.value])
def restart_hysteria2(): def restart_hysteria2():
"""Restarts Hysteria2.""" '''Restarts Hysteria2.'''
run_cmd(['bash', Command.RESTART_HYSTERIA2.value]) run_cmd(['bash', Command.RESTART_HYSTERIA2.value])
def change_hysteria2_port(port: int): def change_hysteria2_port(port: int):
""" '''
Changes the port for Hysteria2. Changes the port for Hysteria2.
""" '''
run_cmd(['bash', Command.CHANGE_PORT_HYSTERIA2.value, str(port)]) run_cmd(['bash', Command.CHANGE_PORT_HYSTERIA2.value, str(port)])
def change_hysteria2_sni(sni: str): def change_hysteria2_sni(sni: str):
""" '''
Changes the SNI for Hysteria2. Changes the SNI for Hysteria2.
""" '''
run_cmd(['bash', Command.CHANGE_SNI_HYSTERIA2.value, sni]) run_cmd(['bash', Command.CHANGE_SNI_HYSTERIA2.value, sni])
def backup_hysteria(): def backup_hysteria():
"""Backups Hysteria configuration.""" '''Backups Hysteria configuration.'''
run_cmd(['bash', Command.BACKUP_HYSTERIA.value]) run_cmd(['bash', Command.BACKUP_HYSTERIA.value])
# endregion # endregion
@ -153,25 +153,25 @@ def backup_hysteria():
def list_users() -> dict | None: def list_users() -> dict | None:
""" '''
Lists all users. Lists all users.
""" '''
if res := run_cmd(['bash', Command.LIST_USERS.value]): if res := run_cmd(['bash', Command.LIST_USERS.value]):
return json.loads(res) return json.loads(res)
def get_user(username: str) -> dict | None: def get_user(username: str) -> dict | None:
""" '''
Retrieves information about a specific user. Retrieves information about a specific user.
""" '''
if res := run_cmd(['bash', Command.GET_USER.value, '-u', str(username)]): if res := run_cmd(['bash', Command.GET_USER.value, '-u', str(username)]):
return json.loads(res) return json.loads(res)
def add_user(username: str, traffic_limit: int, expiration_days: int, password: str | None, creation_date: str | None): def add_user(username: str, traffic_limit: int, expiration_days: int, password: str | None, creation_date: str | None):
""" '''
Adds a new user with the given parameters. Adds a new user with the given parameters.
""" '''
if not password: if not password:
password = generate_password() password = generate_password()
if not creation_date: if not creation_date:
@ -180,9 +180,9 @@ def add_user(username: str, traffic_limit: int, expiration_days: int, password:
def edit_user(username: str, new_username: str | None, new_traffic_limit: int | None, new_expiration_days: int | None, renew_password: bool, renew_creation_date: bool, blocked: bool): def edit_user(username: str, new_username: str | None, new_traffic_limit: int | None, new_expiration_days: int | None, renew_password: bool, renew_creation_date: bool, blocked: bool):
""" '''
Edits an existing user's details. Edits an existing user's details.
""" '''
if not username: if not username:
raise InvalidInputError('Error: username is required') raise InvalidInputError('Error: username is required')
if not any([new_username, new_traffic_limit, new_expiration_days, renew_password, renew_creation_date, blocked is not None]): if not any([new_username, new_traffic_limit, new_expiration_days, renew_password, renew_creation_date, blocked is not None]):
@ -194,11 +194,11 @@ def edit_user(username: str, new_username: str | None, new_traffic_limit: int |
if renew_password: if renew_password:
password = generate_password() password = generate_password()
else: else:
password = "" password = ''
if renew_creation_date: if renew_creation_date:
creation_date = datetime.now().strftime('%Y-%m-%d') creation_date = datetime.now().strftime('%Y-%m-%d')
else: else:
creation_date = "" creation_date = ''
command_args = [ command_args = [
'bash', 'bash',
Command.EDIT_USER.value, Command.EDIT_USER.value,
@ -214,23 +214,24 @@ def edit_user(username: str, new_username: str | None, new_traffic_limit: int |
def reset_user(username: str): def reset_user(username: str):
""" '''
Resets a user's configuration. Resets a user's configuration.
""" '''
run_cmd(['bash', Command.RESET_USER.value, username]) run_cmd(['bash', Command.RESET_USER.value, username])
def remove_user(username: str): def remove_user(username: str):
""" '''
Removes a user by username. Removes a user by username.
""" '''
run_cmd(['bash', Command.REMOVE_USER.value, username]) run_cmd(['bash', Command.REMOVE_USER.value, username])
# TODO: it's better to return json
def show_user_uri(username: str, qrcode: bool, ipv: int, all: bool, singbox: bool, normalsub: bool) -> str | None: def show_user_uri(username: str, qrcode: bool, ipv: int, all: bool, singbox: bool, normalsub: bool) -> str | None:
""" '''
Displays the URI for a user, with options for QR code and other formats. Displays the URI for a user, with options for QR code and other formats.
""" '''
command_args = ['bash', Command.SHOW_USER_URI.value, '-u', username] command_args = ['bash', Command.SHOW_USER_URI.value, '-u', username]
if qrcode: if qrcode:
command_args.append('-qr') command_args.append('-qr')
@ -250,19 +251,20 @@ def show_user_uri(username: str, qrcode: bool, ipv: int, all: bool, singbox: boo
def traffic_status(): def traffic_status():
"""Fetches traffic status.""" '''Fetches traffic status.'''
traffic.traffic_status() traffic.traffic_status()
# TODO: it's better to return json
def server_info() -> str | None: def server_info() -> str | None:
"""Retrieves server information.""" '''Retrieves server information.'''
return run_cmd(['bash', Command.SERVER_INFO.value]) return run_cmd(['bash', Command.SERVER_INFO.value])
def manage_obfs(remove: bool, generate: bool): def manage_obfs(remove: bool, generate: bool):
""" '''
Manages 'obfs' in Hysteria2 configuration. Manages 'obfs' in Hysteria2 configuration.
""" '''
if remove and generate: if remove and generate:
raise InvalidInputError('Error: You cannot use both --remove and --generate at the same time') raise InvalidInputError('Error: You cannot use both --remove and --generate at the same time')
elif remove: elif remove:
@ -270,51 +272,51 @@ def manage_obfs(remove: bool, generate: bool):
elif generate: elif generate:
run_cmd(['bash', Command.MANAGE_OBFS.value, '--generate']) run_cmd(['bash', Command.MANAGE_OBFS.value, '--generate'])
else: else:
raise InvalidInputError("Error: Please specify either --remove or --generate.") raise InvalidInputError('Error: Please specify either --remove or --generate.')
def ip_address(edit: bool, ipv4: str, ipv6: str): def ip_address(edit: bool, ipv4: str, ipv6: str):
""" '''
Manages IP address configuration with edit options. Manages IP address configuration with edit options.
""" '''
if edit: if edit:
if ipv4: if ipv4:
run_cmd(['bash', Command.IP_ADD.value, 'edit', '-4', ipv4]) run_cmd(['bash', Command.IP_ADD.value, 'edit', '-4', ipv4])
if ipv6: if ipv6:
run_cmd(['bash', Command.IP_ADD.value, 'edit', '-6', ipv6]) run_cmd(['bash', Command.IP_ADD.value, 'edit', '-6', ipv6])
if not ipv4 and not ipv6: if not ipv4 and not ipv6:
raise InvalidInputError("Error: --edit requires at least one of --ipv4 or --ipv6.") raise InvalidInputError('Error: --edit requires at least one of --ipv4 or --ipv6.')
else: else:
run_cmd(['bash', Command.IP_ADD.value, 'add']) run_cmd(['bash', Command.IP_ADD.value, 'add'])
def update_geo(country: str): def update_geo(country: str):
""" '''
Updates geographic data files based on the specified country. Updates geographic data files based on the specified country.
""" '''
script_path = Command.UPDATE_GEO.value script_path = Command.UPDATE_GEO.value
try: try:
subprocess.run(['python3', script_path, country.lower()], check=True) subprocess.run(['python3', script_path, country.lower()], check=True)
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
raise CommandExecutionError(f"Failed to update geo files: {e}") raise CommandExecutionError(f'Failed to update geo files: {e}')
except FileNotFoundError: except FileNotFoundError:
raise ScriptNotFoundError(f"Script not found: {script_path}") raise ScriptNotFoundError(f'Script not found: {script_path}')
except Exception as e: except Exception as e:
raise HysteriaError(f"An unexpected error occurred: {e}") raise HysteriaError(f'An unexpected error occurred: {e}')
def masquerade(remove: bool, enable: str): def masquerade(remove: bool, enable: str):
""" '''
Configures masquerade settings. Configures masquerade settings.
""" '''
if remove and enable: if remove and enable:
raise InvalidInputError("Error: You cannot use both --remove and --enable at the same time.") raise InvalidInputError('Error: You cannot use both --remove and --enable at the same time.')
if remove: if remove:
run_cmd(['bash', Command.MASQUERADE_SCRIPT.value, '2']) run_cmd(['bash', Command.MASQUERADE_SCRIPT.value, '2'])
elif enable: elif enable:
run_cmd(['bash', Command.MASQUERADE_SCRIPT.value, '1', enable]) run_cmd(['bash', Command.MASQUERADE_SCRIPT.value, '1', enable])
else: else:
raise InvalidInputError("Error: Please specify either --remove or --enable.") raise InvalidInputError('Error: Please specify either --remove or --enable.')
# endregion # endregion
@ -322,33 +324,33 @@ def masquerade(remove: bool, enable: str):
def install_tcp_brutal(): def install_tcp_brutal():
"""Installs TCP Brutal.""" '''Installs TCP Brutal.'''
run_cmd(['bash', Command.INSTALL_TCP_BRUTAL.value]) run_cmd(['bash', Command.INSTALL_TCP_BRUTAL.value])
def install_warp(): def install_warp():
"""Installs WARP.""" '''Installs WARP.'''
run_cmd(['bash', Command.INSTALL_WARP.value]) run_cmd(['bash', Command.INSTALL_WARP.value])
def uninstall_warp(): def uninstall_warp():
"""Uninstalls WARP.""" '''Uninstalls WARP.'''
run_cmd(['bash', Command.UNINSTALL_WARP.value]) run_cmd(['bash', Command.UNINSTALL_WARP.value])
def configure_warp(all: bool, popular_sites: bool, domestic_sites: bool, block_adult_sites: bool, warp_option: str, warp_key: str): def configure_warp(all: bool, popular_sites: bool, domestic_sites: bool, block_adult_sites: bool, warp_option: str, warp_key: str):
""" '''
Configures WARP with various options. Configures WARP with various options.
""" '''
if warp_option == 'warp plus' and not warp_key: if warp_option == 'warp plus' and not warp_key:
raise InvalidInputError("Error: WARP Plus key is required when 'warp plus' is selected.") raise InvalidInputError('Error: WARP Plus key is required when \'warp plus\' is selected.')
options = { options = {
"all": 'true' if all else 'false', 'all': 'true' if all else 'false',
"popular_sites": 'true' if popular_sites else 'false', 'popular_sites': 'true' if popular_sites else 'false',
"domestic_sites": 'true' if domestic_sites else 'false', 'domestic_sites': 'true' if domestic_sites else 'false',
"block_adult_sites": 'true' if block_adult_sites else 'false', 'block_adult_sites': 'true' if block_adult_sites else 'false',
"warp_option": warp_option or '', 'warp_option': warp_option or '',
"warp_key": warp_key or '' 'warp_key': warp_key or ''
} }
cmd_args = [ cmd_args = [
'bash', Command.CONFIGURE_WARP.value, 'bash', Command.CONFIGURE_WARP.value,
@ -364,17 +366,17 @@ def configure_warp(all: bool, popular_sites: bool, domestic_sites: bool, block_a
def warp_status() -> str | None: def warp_status() -> str | None:
"""Checks the status of WARP.""" '''Checks the status of WARP.'''
return run_cmd(['bash', Command.STATUS_WARP.value]) return run_cmd(['bash', Command.STATUS_WARP.value])
def telegram(action: str, token: str, adminid: str): def telegram(action: str, token: str, adminid: str):
""" '''
Manages the Telegram bot with start/stop actions. Manages the Telegram bot with start/stop actions.
""" '''
if action == 'start': if action == 'start':
if not token or not adminid: if not token or not adminid:
raise InvalidInputError("Error: Both --token and --adminid are required for the start action.") raise InvalidInputError('Error: Both --token and --adminid are required for the start action.')
admin_ids = f'{adminid}' admin_ids = f'{adminid}'
run_cmd(['bash', Command.INSTALL_TELEGRAMBOT.value, 'start', token, admin_ids]) run_cmd(['bash', Command.INSTALL_TELEGRAMBOT.value, 'start', token, admin_ids])
elif action == 'stop': elif action == 'stop':
@ -382,24 +384,24 @@ def telegram(action: str, token: str, adminid: str):
def singbox(action: str, domain: str, port: int): def singbox(action: str, domain: str, port: int):
""" '''
Manages Singbox with start/stop actions. Manages Singbox with start/stop actions.
""" '''
if action == 'start': if action == 'start':
if not domain or not port: if not domain or not port:
raise InvalidInputError("Error: Both --domain and --port are required for the start action.") raise InvalidInputError('Error: Both --domain and --port are required for the start action.')
run_cmd(['bash', Command.INSTALL_SINGBOX.value, 'start', domain, str(port)]) run_cmd(['bash', Command.INSTALL_SINGBOX.value, 'start', domain, str(port)])
elif action == 'stop': elif action == 'stop':
run_cmd(['bash', Command.INSTALL_SINGBOX.value, 'stop']) run_cmd(['bash', Command.INSTALL_SINGBOX.value, 'stop'])
def normalsub(action: str, domain: str, port: int): def normalsub(action: str, domain: str, port: int):
""" '''
Manages Normalsub with start/stop actions. Manages Normalsub with start/stop actions.
""" '''
if action == 'start': if action == 'start':
if not domain or not port: if not domain or not port:
raise InvalidInputError("Error: Both --domain and --port are required for the start action.") raise InvalidInputError('Error: Both --domain and --port are required for the start action.')
run_cmd(['bash', Command.INSTALL_NORMALSUB.value, 'start', domain, str(port)]) run_cmd(['bash', Command.INSTALL_NORMALSUB.value, 'start', domain, str(port)])
elif action == 'stop': elif action == 'stop':
run_cmd(['bash', Command.INSTALL_NORMALSUB.value, 'stop']) run_cmd(['bash', Command.INSTALL_NORMALSUB.value, 'stop'])