93 lines
3.1 KiB
Python
Executable File
93 lines
3.1 KiB
Python
Executable File
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
from core.constants import *
|
|
|
|
|
|
class Config:
|
|
def __init__(self, file_name=DATA_FILE_NAME):
|
|
self.file_name = file_name
|
|
self.data = {
|
|
"cipher_enabled": True,
|
|
"server_url": "http://localhost:8080",
|
|
"websocket_url": "",
|
|
"username": "",
|
|
"hashed_password": None,
|
|
"cookie": None,
|
|
"maxsize": None,
|
|
"hash_rounds": 664937,
|
|
"salt": "",
|
|
"csrf_token": "",
|
|
"notification": True,
|
|
"save_password": False,
|
|
"password": "",
|
|
"max_clipboard_size_local_limit_bytes": None,
|
|
"enable_image_sharing": True,
|
|
"enable_file_sharing": True,
|
|
"default_file_download_location": "",
|
|
"server_mode": "P2S",
|
|
"stun_url": "",
|
|
"ssl_ca_bundle": "",
|
|
}
|
|
|
|
def save(self):
|
|
"""
|
|
Save data to file
|
|
"""
|
|
try:
|
|
temp = self.data.copy()
|
|
if self.data.get("cipher_enabled") and self.data.get("hashed_password"):
|
|
temp["hashed_password"] = base64.b64encode(
|
|
temp["hashed_password"]
|
|
).decode("utf-8")
|
|
with open(self.file_name, "w") as f:
|
|
json.dump(temp, f, indent=4)
|
|
except Exception as e:
|
|
logging.error(f"Failed to save data: {e}")
|
|
|
|
def load(self):
|
|
"""
|
|
Load data from file
|
|
"""
|
|
if os.path.isfile(self.file_name):
|
|
try:
|
|
with open(self.file_name, "r") as f:
|
|
file_data = json.load(f)
|
|
self.data.update(file_data)
|
|
# Decode hashed_password if present
|
|
if self.data.get("hashed_password"):
|
|
self.data["hashed_password"] = base64.b64decode(
|
|
self.data["hashed_password"]
|
|
)
|
|
return True
|
|
except Exception as e:
|
|
logging.error(f"Failed to load data: {e}")
|
|
logging.error(
|
|
"Try deleting DATA file in the program directory, and re-run the program again"
|
|
)
|
|
return False
|
|
|
|
@staticmethod
|
|
def convert_to_websocket_url(input_url: str, endpoint: str = None) -> str:
|
|
if not input_url or not isinstance(input_url, str):
|
|
raise ValueError("Invalid URL provided")
|
|
|
|
# Trim whitespace, remove trailing slashes, and convert to lowercase
|
|
input_url = re.sub(r"/+$", "", input_url.strip()).lower()
|
|
|
|
# Determine protocol and convert
|
|
if input_url.startswith("https://"):
|
|
ws_url = input_url.replace("https://", "wss://", 1)
|
|
elif input_url.startswith("http://"):
|
|
ws_url = input_url.replace("http://", "ws://", 1)
|
|
else:
|
|
raise ValueError(f"Unsupported protocol in URL: {input_url}")
|
|
|
|
if endpoint is not None:
|
|
# Append the WebSocket endpoint and remove any trailing slash
|
|
ws_url += endpoint
|
|
ws_url = re.sub(r"/+$", "", ws_url)
|
|
|
|
return ws_url
|