496 lines
17 KiB
Python
Executable File
496 lines
17 KiB
Python
Executable File
""" Classes Method and Functions for lx Apps """
|
|
|
|
import gettext
|
|
import locale
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import tkinter as tk
|
|
from typing import Optional, Dict, Any, NoReturn, TextIO, Tuple, List
|
|
import zipfile
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from subprocess import check_call, CompletedProcess
|
|
from tkinter import ttk, Toplevel
|
|
|
|
import requests
|
|
|
|
APP = "wirepy"
|
|
LOCALE_DIR = "/usr/share/locale/"
|
|
locale.bindtextdomain(APP, LOCALE_DIR)
|
|
gettext.bindtextdomain(APP, LOCALE_DIR)
|
|
gettext.textdomain(APP)
|
|
_ = gettext.gettext
|
|
|
|
|
|
class Create:
|
|
"""
|
|
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 dir_and_files() -> None:
|
|
"""
|
|
check and create folders and files if not present
|
|
"""
|
|
|
|
pth: Path = Path.home() / ".config/wire_py"
|
|
pth.mkdir(parents=True, exist_ok=True)
|
|
sett: Path = Path.home() / ".config/wire_py/settings"
|
|
ks: Path = Path.home() / ".config/wire_py/keys"
|
|
|
|
if sett.exists():
|
|
pass
|
|
|
|
else:
|
|
sett.touch()
|
|
sett.write_text("[UPDATES]\non\n[THEME]\nlight\n[TOOLTIP]\nTrue\n[AUTOSTART ON]\noff\n")
|
|
|
|
if ks.exists():
|
|
pass
|
|
|
|
else:
|
|
ks.touch()
|
|
|
|
@staticmethod
|
|
def files_for_autostart() -> None:
|
|
"""
|
|
check and create a file for auto start if not present and enable the service
|
|
"""
|
|
|
|
pth2: Path = Path.home() / ".config/systemd/user"
|
|
pth2.mkdir(parents=True, exist_ok=True)
|
|
wg_ser: Path = Path.home() / ".config/systemd/user/wg_start.service"
|
|
|
|
if wg_ser.exists():
|
|
pass
|
|
|
|
else:
|
|
wg_ser.touch()
|
|
wg_ser.write_text("[Unit]\nDescription=Automatic Tunnel Start\nAfter=network-online.target\n\n[Service]\n"
|
|
"Type=oneshot\nExecStartPre=/bin/sleep 5\nExecStart=/usr/local/bin/start_wg.py\n[Install]"
|
|
"\nWantedBy=default.target")
|
|
check_call(["systemctl", "--user", "enable", "wg_start.service"])
|
|
|
|
@staticmethod
|
|
def make_dir() -> None:
|
|
"""Folder Name "tlecdewg" = Tunnel Encrypt Decrypt Wireguard"""
|
|
|
|
folder_path: Path = Path("/tmp/tlecdcwg/")
|
|
if folder_path.exists():
|
|
pass
|
|
else:
|
|
folder_path.mkdir()
|
|
|
|
@staticmethod
|
|
def decrypt() -> None:
|
|
"""
|
|
Starts SSL dencrypt
|
|
"""
|
|
process: CompletedProcess[str] = subprocess.run(["pkexec", "/usr/local/bin/ssl_decrypt.py"],
|
|
stdout=subprocess.PIPE, text=True, check=True)
|
|
path: Path = Path.home() / ".config/wire_py/"
|
|
file_in_path: list[Path] = list(path.rglob("*.dat"))
|
|
if file_in_path:
|
|
if process.returncode == 0:
|
|
print("File successfully decrypted...")
|
|
else:
|
|
print(f"Error with the following code... {process.returncode}")
|
|
else:
|
|
print(_("Ready for import"))
|
|
|
|
@staticmethod
|
|
def encrypt() -> None:
|
|
"""
|
|
Starts SSL encryption
|
|
"""
|
|
process: CompletedProcess[str] = subprocess.run(["pkexec", "/usr/local/bin/ssl_encrypt.py"],
|
|
stdout=subprocess.PIPE, text=True, check=True)
|
|
print(process.stdout)
|
|
if process.returncode == 0:
|
|
print("All Files successfully encrypted...")
|
|
else:
|
|
print(f"Error with the following code... {process.returncode}")
|
|
|
|
|
|
class LxTools(tk.Tk):
|
|
"""
|
|
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 uos() -> None:
|
|
"""
|
|
uos = LOGIN USERNAME
|
|
|
|
This method displays the username of the logged-in user,
|
|
even if you are rooted in a shell
|
|
"""
|
|
log_name: str = f"{Path.home()}"[6:]
|
|
file: Path = Path.home() / "/tmp/.log_user"
|
|
with open(file, "w", encoding="utf-8") as f:
|
|
f.write(log_name)
|
|
|
|
@staticmethod
|
|
def clean_files(folder_path: Path = None, file: Path = None) -> None:
|
|
"""
|
|
method that can be added after need to delete a folder and a file when quitting.
|
|
Args:
|
|
:param file: default None
|
|
:param folder_path: default None
|
|
"""
|
|
if folder_path is not None:
|
|
shutil.rmtree(folder_path)
|
|
if file is not None:
|
|
Path.unlink(file)
|
|
|
|
@staticmethod
|
|
def if_tip(path: Path) -> bool:
|
|
"""
|
|
method that writes in file whether tooltip is displayed or not
|
|
"""
|
|
with open(path, "r", encoding="utf-8") as set_f2:
|
|
lines2 = set_f2.readlines()
|
|
if "False\n" in lines2:
|
|
tip = False
|
|
else:
|
|
tip = True
|
|
return tip
|
|
|
|
@staticmethod
|
|
def msg_window(img_w: str, img_i: str, w_title: str, w_txt: str, txt2: Optional[str] = None,
|
|
com: Optional[str] = None) -> None:
|
|
"""
|
|
Creates message windows
|
|
|
|
:argument img_w = Image for TK window which is displayed to the left of the text
|
|
:argument img_i = Image for Task Icon
|
|
:argument w_title = Windows Title
|
|
:argument w_txt = Text for Tk Window
|
|
:argument txt2 = Text for Button two
|
|
:argument com = function for Button command
|
|
"""
|
|
msg: tk.Toplevel = tk.Toplevel()
|
|
msg.resizable(width=False, height=False)
|
|
msg.title(w_title)
|
|
msg.configure(pady=15, padx=15)
|
|
msg.img = tk.PhotoImage(file=img_w)
|
|
msg.i_window = tk.Label(msg, image=msg.img)
|
|
|
|
label: tk.Label = tk.Label(msg, text=w_txt)
|
|
|
|
label.grid(column=1, row=0)
|
|
|
|
if txt2 is not None and com is not None:
|
|
label.config(font=("Ubuntu", 11), padx=15, justify="left")
|
|
msg.i_window.grid(column=0, row=0, sticky="nw")
|
|
button2: ttk.Button = ttk.Button(msg, text=f"{txt2}", command=com, padding=4)
|
|
button2.grid(column=0, row=1, sticky="e", columnspan=2)
|
|
button: ttk.Button = ttk.Button(msg, text="OK", command=msg.destroy, padding=4)
|
|
button.grid(column=0, row=1, sticky="w", columnspan=2)
|
|
|
|
else:
|
|
label.config(font=("Ubuntu", 11), padx=15)
|
|
msg.i_window.grid(column=0, row=0)
|
|
button: ttk.Button = ttk.Button(msg, text="OK", command=msg.destroy, padding=4)
|
|
button.grid(column=0, columnspan=2, row=1)
|
|
|
|
img_i: tk.PhotoImage = tk.PhotoImage(file=img_i)
|
|
msg.iconphoto(True, img_i)
|
|
msg.columnconfigure(0, weight=1)
|
|
msg.rowconfigure(0, weight=1)
|
|
msg.winfo_toplevel()
|
|
|
|
@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
|
|
print(f"\nSignal {signal_name} {signum} received. => Aborting with exit code {exit_code}.")
|
|
LxTools.clean_files(file_path, file)
|
|
print("Breakdown by user...")
|
|
sys.exit(exit_code)
|
|
else:
|
|
print(f"Signal {signum} received and ignored.")
|
|
LxTools.clean_files(file_path, file)
|
|
print("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)
|
|
|
|
|
|
class GiteaUpdate:
|
|
"""
|
|
Calling download requests the download URL of the running script,
|
|
the taskbar image for the “Download OK” window, the taskbar image for the
|
|
“Download error” window and the variable res
|
|
"""
|
|
|
|
@staticmethod
|
|
def api_down(update_api_url: str, version: str, file: Optional[Path] = None) -> str:
|
|
"""
|
|
Checks for updates via API
|
|
|
|
Args:
|
|
update_api_url: Update API URL
|
|
version: Current version
|
|
file: Optional - Configuration file
|
|
|
|
Returns:
|
|
New version or status message
|
|
"""
|
|
try:
|
|
response: requests.Response = requests.get(update_api_url, timeout=10)
|
|
response_dict: Any = response.json()
|
|
response_dict: Dict[str, Any] = response_dict[0]
|
|
with open(file, "r", encoding="utf-8") as set_f:
|
|
set_f = set_f.read()
|
|
if "on\n" in set_f:
|
|
if version[3:] != response_dict["tag_name"]:
|
|
req: str = response_dict["tag_name"]
|
|
else:
|
|
req: str = "No Updates"
|
|
else:
|
|
req: str = "False"
|
|
return req
|
|
except requests.exceptions.RequestException:
|
|
req: str = "No Internet Connection!"
|
|
return req
|
|
|
|
@staticmethod
|
|
def download(urld: str, res: str, img_w: str = None, img_i: str = None, img_w2: str = None, img_i2: str = None) -> None:
|
|
"""
|
|
Downloads new version of wirepy
|
|
|
|
Args:
|
|
urld: Download URL
|
|
res: Result filename
|
|
img_w: Image for TK window which is displayed to the left of the text
|
|
img_i: Image for Task Icon
|
|
img_w2: Image for TK window which is displayed to the left of the text
|
|
img_i2: Image for Task Icon
|
|
"""
|
|
try:
|
|
to_down: str = f"wget -qP {Path.home()} {" "} {urld}"
|
|
result: int = subprocess.call(to_down, shell=True)
|
|
if result == 0:
|
|
shutil.chown(f"{Path.home()}/{res}.zip", 1000, 1000)
|
|
|
|
wt: str = _("Download Successful")
|
|
msg_t: str = _("Your zip file is in home directory")
|
|
LxTools.msg_window(img_w, img_i, wt, msg_t)
|
|
|
|
else:
|
|
|
|
wt: str = _("Download error")
|
|
msg_t: str = _("Download failed! Please try again")
|
|
LxTools.msg_window(img_w2, img_i2, wt, msg_t)
|
|
|
|
except subprocess.CalledProcessError:
|
|
|
|
wt: str = _("Download error")
|
|
msg_t: str = _("Download failed! No internet connection!")
|
|
LxTools.msg_window(img_w2, img_i2, wt, msg_t)
|
|
|
|
|
|
class Tunnel:
|
|
"""
|
|
Class of Methods for Wire-Py
|
|
"""
|
|
|
|
@classmethod
|
|
def con_to_dict(cls, file: TextIO) -> Tuple[str, str, str, Optional[str]]:
|
|
"""
|
|
Returns tuple of (address, dns, endpoint, pre_key)
|
|
"""
|
|
|
|
dictlist: List[str] = []
|
|
for lines in file.readlines():
|
|
line_plit: List[str] = lines.split()
|
|
dictlist = dictlist + line_plit
|
|
dictlist.remove("[Interface]")
|
|
dictlist.remove("[Peer]")
|
|
for items in dictlist:
|
|
if items == "=":
|
|
dictlist.remove(items)
|
|
if items == "::/0":
|
|
dictlist.remove(items)
|
|
|
|
# Here is the beginning (Loop) of convert List to Dictionary
|
|
for _ in dictlist:
|
|
a: List[str] = [dictlist[0], dictlist[1]]
|
|
b: List[str] = [dictlist[2], dictlist[3]]
|
|
c: List[str] = [dictlist[4], dictlist[5]]
|
|
d: List[str] = [dictlist[6], dictlist[7]]
|
|
e: List[str] = [dictlist[8], dictlist[9]]
|
|
f: List[str] = [dictlist[10], dictlist[11]]
|
|
g: List[str] = [dictlist[12], dictlist[13]]
|
|
h: List[str] = [dictlist[14], dictlist[15]]
|
|
new_list: List[List[str]] = [a, b, c, d, e, f, g, h]
|
|
final_dict: Dict[str, str] = {}
|
|
for elements in new_list:
|
|
final_dict[elements[0]] = elements[1]
|
|
|
|
# end... result a Dictionary
|
|
|
|
address: str = final_dict["Address"]
|
|
dns: str = final_dict["DNS"]
|
|
if "," in dns:
|
|
dns = dns[:-1]
|
|
endpoint: str = final_dict["Endpoint"]
|
|
pre_key: Optional[str] = final_dict.get("PresharedKey")
|
|
if pre_key is None:
|
|
pre_key: Optional[str] = final_dict.get("PreSharedKey")
|
|
return address, dns, endpoint, pre_key
|
|
|
|
@staticmethod
|
|
def active() -> str:
|
|
"""
|
|
Shows the Active Tunnel
|
|
"""
|
|
active = (os.popen('nmcli con show --active | grep -iPo "(.*)(wireguard)"').read().split())
|
|
if not active:
|
|
active = ""
|
|
else:
|
|
active = active[0]
|
|
|
|
return active
|
|
|
|
@staticmethod
|
|
def list() -> List[str]:
|
|
"""
|
|
Returns a list of Wireguard tunnel names
|
|
"""
|
|
folder_path: Path = Path("/tmp/tlecdcwg/")
|
|
wg_s: List[str] = os.listdir(folder_path)
|
|
|
|
return wg_s
|
|
|
|
@staticmethod
|
|
def export(img_w: str = None, img_i: str = None, img_w2: str = None, img_i2: str = None, sl: str = None, pfit:str = None) -> None:
|
|
"""
|
|
This will export the tunnels.
|
|
A zipfile with the current date and time is created
|
|
in the user's home directory with the correct right
|
|
Args:
|
|
img_w: Image for TK window which is displayed to the left of the text
|
|
img_i: Image for Task Icon
|
|
img_w2: Image for TK window which is displayed to the left of the text
|
|
img_i2: Image for Task Icon
|
|
"""
|
|
now_time: datetime = datetime.now()
|
|
now_datetime: str = now_time.strftime("wg-exp-%m-%d-%Y-%H:%M")
|
|
tl: List[str] = Tunnel.list()
|
|
|
|
try:
|
|
if len(tl) != 0:
|
|
wg_tar: str = f"{Path.home()}/{now_datetime}"
|
|
shutil.copytree("/tmp/tlecdcwg/", "/tmp/wire_py", dirs_exist_ok=True)
|
|
source: Path = Path("/tmp/wire_py")
|
|
shutil.make_archive(wg_tar, "zip", source)
|
|
shutil.rmtree(source)
|
|
with zipfile.ZipFile(f"{wg_tar}.zip", "r") as zf:
|
|
if len(zf.namelist()) != 0:
|
|
|
|
msg_t: str = _("Your zip file is in home directory")
|
|
LxTools.msg_window(img_w, img_i, _("Export Successful"), msg_t)
|
|
|
|
else:
|
|
|
|
msg_t: str = _("Export failed! Please try again")
|
|
LxTools.msg_window(img_w2, img_i2, _("Export error"), msg_t)
|
|
|
|
else:
|
|
|
|
LxTools.msg_window(img_w, img_i2, sl, pfit)
|
|
|
|
except TypeError:
|
|
pass
|
|
|
|
|
|
class Tooltip:
|
|
"""
|
|
class for Tooltip
|
|
|
|
import Tooltip
|
|
|
|
example: Tooltip(label, "Show tooltip on label")
|
|
example: Tooltip(button, "Show tooltip on button")
|
|
info: label and button are parent.
|
|
"""
|
|
|
|
def __init__(self, widget: Any, text: str, tips: Optional[bool] = None) -> None:
|
|
"""
|
|
Tooltip Class
|
|
"""
|
|
|
|
self.widget: Any = widget
|
|
self.text: str = text
|
|
self.tooltip_window: Optional[Toplevel] = None
|
|
if tips:
|
|
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() + 65
|
|
y += self.widget.winfo_rooty() + 40
|
|
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()
|
|
|
|
def hide_tooltip(self, event: Optional[Any] = None) -> None:
|
|
"""
|
|
Hides the tooltip
|
|
"""
|
|
if self.tooltip_window:
|
|
self.tooltip_window.destroy()
|
|
self.tooltip_window = None
|
|
|