506 lines
17 KiB
Python
Executable File
506 lines
17 KiB
Python
Executable File
""" Classes Method and Functions for lx Apps """
|
|
|
|
import logging
|
|
import signal
|
|
import base64
|
|
from subprocess import CompletedProcess, run
|
|
import re
|
|
import sys
|
|
import shutil
|
|
import tkinter as tk
|
|
from typing import Optional, Dict, Any, NoReturn
|
|
from pathlib import Path
|
|
from tkinter import Toplevel
|
|
|
|
|
|
class CryptoUtil:
|
|
"""
|
|
This class is for the creation of the folders and files
|
|
required by Wire-Py, as well as for decryption
|
|
the tunnel from the user's home directory
|
|
"""
|
|
|
|
@staticmethod
|
|
def decrypt(user) -> None:
|
|
"""
|
|
Starts SSL dencrypt
|
|
"""
|
|
process: CompletedProcess[str] = run(
|
|
["pkexec", "/usr/local/bin/ssl_decrypt.py", "--user", user],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
# Output from Openssl Error
|
|
if process.stderr:
|
|
logging.error(process.stderr, exc_info=True)
|
|
|
|
if process.returncode == 0:
|
|
logging.info("Files successfully decrypted...", exc_info=True)
|
|
else:
|
|
|
|
logging.error(
|
|
f"Error process decrypt: Code {process.returncode}", exc_info=True
|
|
)
|
|
|
|
@staticmethod
|
|
def encrypt(user) -> None:
|
|
"""
|
|
Starts SSL encryption
|
|
"""
|
|
process: CompletedProcess[str] = run(
|
|
["pkexec", "/usr/local/bin/ssl_encrypt.py", "--user", user],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
# Output from Openssl Error
|
|
if process.stderr:
|
|
logging.error(process.stderr, exc_info=True)
|
|
|
|
if process.returncode == 0:
|
|
logging.info("Files successfully encrypted...", exc_info=True)
|
|
else:
|
|
logging.error(
|
|
f"Error process encrypt: Code {process.returncode}", exc_info=True
|
|
)
|
|
|
|
@staticmethod
|
|
def find_key(key: str = "") -> bool:
|
|
"""
|
|
Checks if the private key already exists in the system using an external script.
|
|
Returns True only if the full key is found exactly (no partial match).
|
|
"""
|
|
process: CompletedProcess[str] = run(
|
|
["pkexec", "/usr/local/bin/match_found.py", key],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if "True" in process.stdout:
|
|
return True
|
|
elif "False" in process.stdout:
|
|
return False
|
|
logging.error(
|
|
f"Unexpected output from the external script:\nSTDOUT: {process.stdout}\nSTDERR: {process.stderr}",
|
|
exc_info=True,
|
|
)
|
|
return False
|
|
|
|
@staticmethod
|
|
def is_valid_base64(key: str) -> bool:
|
|
"""
|
|
Validates if the input is a valid Base64 string (WireGuard private key format).
|
|
Returns True only for non-empty strings that match the expected length.
|
|
"""
|
|
# Check for empty string
|
|
if not key or key.strip() == "":
|
|
return False
|
|
|
|
# Regex pattern to validate Base64: [A-Za-z0-9+/]+={0,2}
|
|
base64_pattern = r"^[A-Za-z0-9+/]+={0,2}$"
|
|
if not re.match(base64_pattern, key):
|
|
return False
|
|
|
|
try:
|
|
# Decode and check length (WireGuard private keys are 32 bytes long)
|
|
decoded = base64.b64decode(key)
|
|
if len(decoded) != 32: # 32 bytes = 256 bits
|
|
return False
|
|
except Exception as e:
|
|
logging.error(f"Error on decode Base64: {e}", exc_info=True)
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
class LxTools:
|
|
"""
|
|
Class LinuxTools methods that can also be used for other apps
|
|
"""
|
|
|
|
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
|
super().__init__(*args, **kwargs)
|
|
|
|
@staticmethod
|
|
def center_window_cross_platform(window, width, height):
|
|
"""
|
|
Centers a window on the primary monitor in a way that works on both X11 and Wayland
|
|
|
|
Args:
|
|
window: The tkinter window to center
|
|
width: Window width
|
|
height: Window height
|
|
"""
|
|
# Calculate the position before showing the window
|
|
|
|
# First attempt: Try to use GDK if available (works on both X11 and Wayland)
|
|
try:
|
|
import gi
|
|
|
|
gi.require_version("Gdk", "3.0")
|
|
from gi.repository import Gdk
|
|
|
|
display = Gdk.Display.get_default()
|
|
monitor = display.get_primary_monitor() or display.get_monitor(0)
|
|
geometry = monitor.get_geometry()
|
|
scale_factor = monitor.get_scale_factor()
|
|
|
|
# Calculate center position on the primary monitor
|
|
x = geometry.x + (geometry.width - width // scale_factor) // 2
|
|
y = geometry.y + (geometry.height - height // scale_factor) // 2
|
|
|
|
# Set window geometry
|
|
window.geometry(f"{width}x{height}+{x}+{y}")
|
|
return
|
|
except (ImportError, AttributeError):
|
|
pass
|
|
|
|
# Second attempt: Try xrandr for X11
|
|
try:
|
|
import subprocess
|
|
|
|
output = subprocess.check_output(
|
|
["xrandr", "--query"], universal_newlines=True
|
|
)
|
|
|
|
# Parse the output to find the primary monitor
|
|
primary_info = None
|
|
for line in output.splitlines():
|
|
if "primary" in line:
|
|
parts = line.split()
|
|
for part in parts:
|
|
if "x" in part and "+" in part:
|
|
primary_info = part
|
|
break
|
|
break
|
|
|
|
if primary_info:
|
|
# Parse the geometry: WIDTH x HEIGHT+X+Y
|
|
geometry = primary_info.split("+")
|
|
dimensions = geometry[0].split("x")
|
|
primary_width = int(dimensions[0])
|
|
primary_height = int(dimensions[1])
|
|
primary_x = int(geometry[1])
|
|
primary_y = int(geometry[2])
|
|
|
|
# Calculate center position on the primary monitor
|
|
x = primary_x + (primary_width - width) // 2
|
|
y = primary_y + (primary_height - height) // 2
|
|
|
|
# Set window geometry
|
|
window.geometry(f"{width}x{height}+{x}+{y}")
|
|
return
|
|
except (ImportError, IndexError, ValueError):
|
|
pass
|
|
|
|
# Final fallback: Use standard Tkinter method
|
|
screen_width = window.winfo_screenwidth()
|
|
screen_height = window.winfo_screenheight()
|
|
|
|
# Try to make an educated guess for multi-monitor setups
|
|
# If screen width is much larger than height, assume multiple monitors side by side
|
|
if (
|
|
screen_width > screen_height * 1.8
|
|
): # Heuristic for detecting multiple monitors
|
|
# Assume the primary monitor is on the left half
|
|
screen_width = screen_width // 2
|
|
|
|
x = (screen_width - width) // 2
|
|
y = (screen_height - height) // 2
|
|
window.geometry(f"{width}x{height}+{x}+{y}")
|
|
|
|
@staticmethod
|
|
def clean_files(tmp_dir: Path = None, file: Path = None) -> None:
|
|
"""
|
|
Deletes temporary files and directories for cleanup when exiting the application.
|
|
|
|
This method safely removes an optional directory defined by `AppConfig.TEMP_DIR`
|
|
and a single file to free up resources at the end of the program's execution.
|
|
All operations are performed securely, and errors such as `FileNotFoundError`
|
|
are ignored if the target files or directories do not exist.
|
|
:param tmp_dir: (Path, optional): Path to the temporary directory that should be deleted.
|
|
If `None`, the value of `AppConfig.TEMP_DIR` is used.
|
|
:param file: (Path, optional): Path to the file that should be deleted.
|
|
If `None`, no additional file will be deleted.
|
|
|
|
Returns:
|
|
None: The method does not return any value.
|
|
"""
|
|
|
|
if tmp_dir is not None:
|
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
|
try:
|
|
if file is not None:
|
|
Path.unlink(file)
|
|
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
@staticmethod
|
|
def sigi(file_path: Optional[Path] = None, file: Optional[Path] = None) -> None:
|
|
"""
|
|
Function for cleanup after a program interruption
|
|
|
|
:param file: Optional - File to be deleted
|
|
:param file_path: Optional - Directory to be deleted
|
|
"""
|
|
|
|
def signal_handler(signum: int, frame: Any) -> NoReturn:
|
|
"""
|
|
Determines clear text names for signal numbers and handles signals
|
|
|
|
Args:
|
|
signum: The signal number
|
|
frame: The current stack frame
|
|
|
|
Returns:
|
|
NoReturn since the function either exits the program or continues execution
|
|
"""
|
|
|
|
signals_to_names_dict: Dict[int, str] = dict(
|
|
(getattr(signal, n), n)
|
|
for n in dir(signal)
|
|
if n.startswith("SIG") and "_" not in n
|
|
)
|
|
|
|
signal_name: str = signals_to_names_dict.get(
|
|
signum, f"Unnamed signal: {signum}"
|
|
)
|
|
|
|
# End program for certain signals, report to others only reception
|
|
if signum in (signal.SIGINT, signal.SIGTERM):
|
|
exit_code: int = 1
|
|
logging.error(
|
|
f"\nSignal {signal_name} {signum} received. => Aborting with exit code {exit_code}.",
|
|
exc_info=True,
|
|
)
|
|
LxTools.clean_files(file_path, file)
|
|
logging.info("Breakdown by user...")
|
|
sys.exit(exit_code)
|
|
else:
|
|
logging.info(f"Signal {signum} received and ignored.")
|
|
LxTools.clean_files(file_path, file)
|
|
logging.error("Process unexpectedly ended...")
|
|
|
|
# Register signal handlers for various signals
|
|
signal.signal(signal.SIGINT, signal_handler)
|
|
signal.signal(signal.SIGTERM, signal_handler)
|
|
signal.signal(signal.SIGHUP, signal_handler)
|
|
|
|
|
|
# ConfigManager with caching
|
|
class ConfigManager:
|
|
"""
|
|
Universal class for managing configuration files with caching support.
|
|
|
|
This class provides a general solution to load, save, and manage configuration
|
|
files across different projects. It uses a caching system to optimize access efficiency.
|
|
The `init()` method initializes the configuration file path, while `load()` and `save()`
|
|
synchronize data between the file and internal memory structures.
|
|
|
|
Key Features:
|
|
- Caching to minimize I/O operations.
|
|
- Default values for missing or corrupted configuration files.
|
|
- Reusability across different projects and use cases.
|
|
|
|
The class is designed for central application configuration management, working closely
|
|
with `ThemeManager` to dynamically manage themes or other settings.
|
|
"""
|
|
|
|
_config = None
|
|
_config_file = None
|
|
|
|
@classmethod
|
|
def init(cls, config_file):
|
|
"""Initial the Configmanager with the given config file"""
|
|
cls._config_file = config_file
|
|
cls._config = None # Reset the cache
|
|
|
|
@classmethod
|
|
def load(cls):
|
|
"""Load the config file and return the config as dict"""
|
|
if not cls._config:
|
|
try:
|
|
lines = Path(cls._config_file).read_text(encoding="utf-8").splitlines()
|
|
cls._config = {
|
|
"updates": lines[1].strip(),
|
|
"theme": lines[3].strip(),
|
|
"tooltips": lines[5].strip()
|
|
== "True", # is converted here to boolean!!!
|
|
"autostart": lines[7].strip() if len(lines) > 7 else "off",
|
|
"logfile": lines[9].strip(),
|
|
}
|
|
except (IndexError, FileNotFoundError):
|
|
# DeDefault values in case of error
|
|
cls._config = {
|
|
"updates": "on",
|
|
"theme": "light",
|
|
"tooltips": "True", # Default Value as string!
|
|
"autostart": "off",
|
|
"logfile": LOG_FILE_PATH,
|
|
}
|
|
return cls._config
|
|
|
|
@classmethod
|
|
def save(cls):
|
|
"""Save the config to the config file"""
|
|
if cls._config:
|
|
lines = [
|
|
"# Configuration\n",
|
|
f"{cls._config['updates']}\n",
|
|
"# Theme\n",
|
|
f"{cls._config['theme']}\n",
|
|
"# Tooltips\n",
|
|
f"{str(cls._config['tooltips'])}\n",
|
|
"# Autostart\n",
|
|
f"{cls._config['autostart']}\n",
|
|
"# Logfile\n",
|
|
f"{cls._config['logfile']}\n",
|
|
]
|
|
Path(cls._config_file).write_text("".join(lines), encoding="utf-8")
|
|
|
|
@classmethod
|
|
def set(cls, key, value):
|
|
"""Sets a configuration value and saves the change"""
|
|
cls.load()
|
|
cls._config[key] = value
|
|
cls.save()
|
|
|
|
@classmethod
|
|
def get(cls, key, default=None):
|
|
"""Returns a configuration value"""
|
|
config = cls.load()
|
|
return config.get(key, default)
|
|
|
|
|
|
class ThemeManager:
|
|
"""
|
|
Class for central theme management and UI customization.
|
|
|
|
This static class allows dynamic adjustment of the application's appearance.
|
|
The method `change_theme()` updates the current theme and saves
|
|
the selection in the configuration file via `ConfigManager`.
|
|
It ensures a consistent visual design across the entire project.
|
|
|
|
Key Features:
|
|
- Central control over themes.
|
|
- Automatic saving of theme settings to the configuration file.
|
|
- Tight integration with `ConfigManager` for persistent storage of preferences.
|
|
|
|
The class is designed to apply themes consistently throughout the application,
|
|
ensuring that changes are traceable and uniform across all parts of the project.
|
|
"""
|
|
|
|
@staticmethod
|
|
def change_theme(root, theme_in_use, theme_name=None):
|
|
"""Change application theme centrally"""
|
|
root.tk.call("set_theme", theme_in_use)
|
|
if theme_in_use == theme_name:
|
|
ConfigManager.set("theme", theme_in_use)
|
|
|
|
|
|
class Tooltip:
|
|
"""Class for Tooltip
|
|
from common_tools.py import Tooltip
|
|
example: Tooltip(label, "Show tooltip on label")
|
|
example: Tooltip(button, "Show tooltip on button")
|
|
example: Tooltip(widget, "Text", state_var=tk.BooleanVar())
|
|
example: Tooltip(widget, "Text", state_var=tk.BooleanVar(), x_offset=20, y_offset=10)
|
|
|
|
info: label and button are parent widgets.
|
|
NOTE: When using with state_var, pass the tk.BooleanVar object directly,
|
|
NOT its value. For example: use state_var=my_bool_var, NOT state_var=my_bool_var.get()
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
widget: Any,
|
|
text: str,
|
|
state_var: Optional[tk.BooleanVar] = None,
|
|
x_offset: int = 65,
|
|
y_offset: int = 40,
|
|
) -> None:
|
|
"""Tooltip Class"""
|
|
self.widget: Any = widget
|
|
self.text: str = text
|
|
self.tooltip_window: Optional[Toplevel] = None
|
|
self.state_var = state_var
|
|
self.x_offset = x_offset
|
|
self.y_offset = y_offset
|
|
|
|
# Initial binding based on the current state
|
|
self.update_bindings()
|
|
|
|
# Add trace to the state_var if provided
|
|
if self.state_var is not None:
|
|
self.state_var.trace_add("write", self.update_bindings)
|
|
|
|
def update_bindings(self, *args) -> None:
|
|
"""Updates the bindings based on the current state"""
|
|
# Remove existing bindings first
|
|
self.widget.unbind("<Enter>")
|
|
self.widget.unbind("<Leave>")
|
|
|
|
# Add new bindings if tooltips are enabled
|
|
if self.state_var is None or self.state_var.get():
|
|
self.widget.bind("<Enter>", self.show_tooltip)
|
|
self.widget.bind("<Leave>", self.hide_tooltip)
|
|
|
|
def show_tooltip(self, event: Optional[Any] = None) -> None:
|
|
"""Shows the tooltip"""
|
|
if self.tooltip_window or not self.text:
|
|
return
|
|
|
|
x: int
|
|
y: int
|
|
cx: int
|
|
cy: int
|
|
|
|
x, y, cx, cy = self.widget.bbox("insert")
|
|
x += self.widget.winfo_rootx() + self.x_offset
|
|
y += self.widget.winfo_rooty() + self.y_offset
|
|
|
|
self.tooltip_window = tw = tk.Toplevel(self.widget)
|
|
tw.wm_overrideredirect(True)
|
|
tw.wm_geometry(f"+{x}+{y}")
|
|
|
|
label: tk.Label = tk.Label(
|
|
tw,
|
|
text=self.text,
|
|
background="lightgreen",
|
|
foreground="black",
|
|
relief="solid",
|
|
borderwidth=1,
|
|
padx=5,
|
|
pady=5,
|
|
)
|
|
label.grid()
|
|
|
|
self.tooltip_window.after(2200, lambda: tw.destroy())
|
|
|
|
def hide_tooltip(self, event: Optional[Any] = None) -> None:
|
|
"""Hides the tooltip"""
|
|
if self.tooltip_window:
|
|
self.tooltip_window.destroy()
|
|
self.tooltip_window = None
|
|
|
|
|
|
class LogConfig:
|
|
@staticmethod
|
|
def logger(file_path) -> None:
|
|
|
|
file_handler = logging.FileHandler(
|
|
filename=f"{file_path}",
|
|
mode="a",
|
|
encoding="utf-8",
|
|
)
|
|
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
|
|
file_handler.setFormatter(formatter)
|
|
file_handler.setLevel(logging.DEBUG)
|
|
|
|
logger = logging.getLogger()
|
|
logger.addHandler(file_handler)
|