2 Commits

7 changed files with 421 additions and 383 deletions

View File

@ -3,14 +3,17 @@
import os import os
import shutil import shutil
import signal import signal
import base64
import secrets
import subprocess import subprocess
from subprocess import CompletedProcess, run
import re
import sys import sys
import tkinter as tk import tkinter as tk
from typing import Optional, Dict, Any, NoReturn, TextIO, Tuple, List from typing import Optional, Dict, Any, NoReturn, List
import zipfile import zipfile
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from subprocess import CompletedProcess
from tkinter import ttk, Toplevel from tkinter import ttk, Toplevel
from wp_app_config import AppConfig, Msg from wp_app_config import AppConfig, Msg
import requests import requests
@ -19,7 +22,7 @@ import requests
_ = AppConfig.setup_translations() _ = AppConfig.setup_translations()
class Create: class CryptoUtil:
""" """
This class is for the creation of the folders and files This class is for the creation of the folders and files
required by Wire-Py, as well as for decryption required by Wire-Py, as well as for decryption
@ -31,6 +34,9 @@ class Create:
""" """
Starts SSL dencrypt Starts SSL dencrypt
""" """
crypted_tunnel = [str(file) for file in AppConfig.CONFIG_DIR.glob("*.dat")]
if crypted_tunnel == []:
return
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = subprocess.run(
["pkexec", "/usr/local/bin/ssl_decrypt.py"], ["pkexec", "/usr/local/bin/ssl_decrypt.py"],
capture_output=True, capture_output=True,
@ -38,10 +44,6 @@ class Create:
check=False, check=False,
) )
# Output from Openssl
# if process.stdout:
# print(process.stdout)
# Output from Openssl Error # Output from Openssl Error
if process.stderr: if process.stderr:
print(process.stderr) print(process.stderr)
@ -72,6 +74,52 @@ class Create:
else: else:
print(f"Error process encrypt: Code {process.returncode}") print(f"Error process encrypt: Code {process.returncode}")
@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[bool] = 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
print(
f"Unexpected output from the external script:\nSTDOUT: {process.stdout}\nSTDERR: {process.stderr}"
)
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:
return False
return True
class LxTools(tk.Tk): class LxTools(tk.Tk):
""" """
@ -81,57 +129,6 @@ class LxTools(tk.Tk):
def __init__(self, *args: Any, **kwargs: Any) -> None: def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
@staticmethod
def ckeck_key_is_exist(
directorys: list[str], search_string: str = None
) -> bool | None:
"""
Check if the key is exist in the file
Args:
directorys (list[str]): list of directories to search in
search_string (str): string to search for
Returns:
bool: True if the key is found, False otherwise
"""
if search_string:
result = False
for directory in directorys:
in_paths = Path(directory)
if not in_paths.exists() or not in_paths.is_dir():
continue
# Get a list of all files in the directorys
files = [file for file in in_paths.iterdir() if file.is_file()]
if not files:
continue
# Search for the string in the files
for file in files:
try:
with open(file, "r", errors="ignore") as f:
for line in f:
if search_string in line:
# Set the result to True if the string is found
result = True
break
# If the string is found, stop searching for the string in other files
if result:
break
except Exception:
# Ignore errors and continue to the next file
continue
else:
result = None
print(result)
return result
@staticmethod @staticmethod
def center_window_cross_platform(window, width, height): def center_window_cross_platform(window, width, height):
""" """
@ -405,23 +402,73 @@ class Tunnel:
""" """
@staticmethod @staticmethod
def parse_files_to_dictionary() -> Dict[str, List[str]]: def parse_files_to_dictionary(
directory: Path = None, filepath: str = None, content: str = None
) -> dict | str | None:
data = {} data = {}
if not AppConfig.TEMP_DIR.exists() or not AppConfig.TEMP_DIR.is_dir(): if filepath is not None:
pass filepath = Path(filepath)
# Get a list of all files in the directorys
files = [file for file in AppConfig.TEMP_DIR.iterdir() if file.is_file()]
if not files:
pass
# Search for the string in the files
for file in files:
try: try:
with open(file, "r") as f: content = filepath.read_text()
content = f.read()
# Hier parsen wir die relevanten Zeilen aus dem Inhalt # parse the content
address_line = next(
line for line in content.splitlines() if line.startswith("Address")
)
dns_line = next(
line for line in content.splitlines() if line.startswith("DNS")
)
endpoint_line = next(
line for line in content.splitlines() if line.startswith("Endpoint")
)
private_key_line = next(
line
for line in content.splitlines()
if line.startswith("PrivateKey")
)
content = secrets.token_bytes(len(content))
# extract the values
address = address_line.split("=")[1].strip()
dns = dns_line.split("=")[1].strip()
endpoint = endpoint_line.split("=")[1].strip()
private_key = private_key_line.split("=")[1].strip()
# Shorten the tunnel name to the maximum allowed length if it exceeds 12 characters.
original_stem = filepath.stem
truncated_stem = (
original_stem[-12:] if len(original_stem) > 12 else original_stem
)
# save in the dictionary
data[truncated_stem] = {
"Address": address,
"DNS": dns,
"Endpoint": endpoint,
"PrivateKey": private_key,
}
content = secrets.token_bytes(len(content))
except StopIteration:
pass
elif directory is not None:
if not directory.exists() or not directory.is_dir():
print("Temp directory does not exist or is not a directory.")
return None
# Get a list of all files in the directory
files = [file for file in AppConfig.TEMP_DIR.iterdir() if file.is_file()]
# Search for the string in the files
for file in files:
try:
content = file.read_text()
# parse the content
address_line = next( address_line = next(
line line
for line in content.splitlines() for line in content.splitlines()
@ -436,67 +483,27 @@ class Tunnel:
if line.startswith("Endpoint") if line.startswith("Endpoint")
) )
# Extrahiere die Werte # extract values
address = address_line.split("=")[1].strip() address = address_line.split("=")[1].strip()
dns = dns_line.split("=")[1].strip() dns = dns_line.split("=")[1].strip()
endpoint = endpoint_line.split("=")[1].strip() endpoint = endpoint_line.split("=")[1].strip()
# Speichere im Dictionary # save values to dictionary
data[file.stem] = { data[file.stem] = {
"Address": address, "Address": address,
"DNS": dns, "DNS": dns,
"Endpoint": endpoint, "Endpoint": endpoint,
} }
except Exception:
# Ignore errors and continue to the next file
continue
return data except Exception:
# Ignore errors and continue to the next file
@classmethod continue
def con_to_dict(cls, file: TextIO) -> Tuple[str, str, str, Optional[str]]: if content is not None:
""" content = secrets.token_bytes(len(content))
Returns tuple of (address, dns, endpoint, pre_key) if filepath is not None:
""" return data, truncated_stem
else:
dictlist: List[str] = [] return data
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 @staticmethod
def active() -> str: def active() -> str:

View File

@ -18,6 +18,7 @@ install_file_with(){
else else
sudo apt install python3-tk && \ sudo apt install python3-tk && \
sudo cp -fv wirepy.py start_wg.py wp_app_config.py common_tools.py ssl_encrypt.py ssl_decrypt.py /usr/local/bin/ && \ sudo cp -fv wirepy.py start_wg.py wp_app_config.py common_tools.py ssl_encrypt.py ssl_decrypt.py /usr/local/bin/ && \
sudo cp -fv match_found.py /usr/local/bin/ && \
sudo cp -uR lx-icons /usr/share/icons/ && sudo cp -uR TK-Themes /usr/share/ && \ sudo cp -uR lx-icons /usr/share/icons/ && sudo cp -uR TK-Themes /usr/share/ && \
sudo cp -u languages/de/*.mo /usr/share/locale/de/LC_MESSAGES/ && \ sudo cp -u languages/de/*.mo /usr/share/locale/de/LC_MESSAGES/ && \
sudo cp -fv Wire-Py.desktop /usr/share/applications/ && \ sudo cp -fv Wire-Py.desktop /usr/share/applications/ && \
@ -44,6 +45,7 @@ install_arch_d(){
else else
sudo pacman -S --noconfirm tk python3 python-requests && \ sudo pacman -S --noconfirm tk python3 python-requests && \
sudo cp -fv wirepy.py start_wg.py wp_app_config.py common_tools.py ssl_encrypt.py ssl_decrypt.py /usr/local/bin/ && \ sudo cp -fv wirepy.py start_wg.py wp_app_config.py common_tools.py ssl_encrypt.py ssl_decrypt.py /usr/local/bin/ && \
sudo cp -fv match_found.py /usr/local/bin/ && \
sudo cp -uR lx-icons /usr/share/icons/ && sudo cp -uR TK-Themes /usr/share/ && \ sudo cp -uR lx-icons /usr/share/icons/ && sudo cp -uR TK-Themes /usr/share/ && \
sudo cp -u languages/de/*.mo /usr/share/locale/de/LC_MESSAGES/ && \ sudo cp -u languages/de/*.mo /usr/share/locale/de/LC_MESSAGES/ && \
sudo cp -fv Wire-Py.desktop /usr/share/applications/ && \ sudo cp -fv Wire-Py.desktop /usr/share/applications/ && \
@ -121,6 +123,7 @@ install(){
else else
sudo dnf install python3-tkinter -y sudo dnf install python3-tkinter -y
sudo cp -fv wirepy.py start_wg.py wp_app_config.py common_tools.py ssl_encrypt.py ssl_decrypt.py /usr/local/bin/ && \ sudo cp -fv wirepy.py start_wg.py wp_app_config.py common_tools.py ssl_encrypt.py ssl_decrypt.py /usr/local/bin/ && \
sudo cp -fv match_found.py /usr/local/bin/ && \
sudo cp -uR lx-icons /usr/share/icons/ && sudo cp -uR TK-Themes /usr/share/ && \ sudo cp -uR lx-icons /usr/share/icons/ && sudo cp -uR TK-Themes /usr/share/ && \
sudo cp -u languages/de/*.mo /usr/share/locale/de/LC_MESSAGES/ && \ sudo cp -u languages/de/*.mo /usr/share/locale/de/LC_MESSAGES/ && \
sudo cp -fv Wire-Py.desktop /usr/share/applications/ && \ sudo cp -fv Wire-Py.desktop /usr/share/applications/ && \
@ -146,6 +149,7 @@ install(){
exit 0 exit 0
else else
sudo cp -fv wirepy.py start_wg.py wp_app_config.py common_tools.py ssl_encrypt.py ssl_decrypt.py /usr/local/bin/ && \ sudo cp -fv wirepy.py start_wg.py wp_app_config.py common_tools.py ssl_encrypt.py ssl_decrypt.py /usr/local/bin/ && \
sudo cp -fv match_found.py /usr/local/bin/ && \
sudo cp -uR lx-icons /usr/share/icons/ && sudo cp -uR TK-Themes /usr/share/ && \ sudo cp -uR lx-icons /usr/share/icons/ && sudo cp -uR TK-Themes /usr/share/ && \
sudo cp -u languages/de/*.mo /usr/share/locale/de/LC_MESSAGES/ && \ sudo cp -u languages/de/*.mo /usr/share/locale/de/LC_MESSAGES/ && \
sudo cp -fv Wire-Py.desktop /usr/share/applications/ && \ sudo cp -fv Wire-Py.desktop /usr/share/applications/ && \
@ -181,7 +185,8 @@ install(){
remove(){ remove(){
sudo rm -f /usr/local/bin/wirepy /usr/local/bin/wirepy.py /usr/local/bin/start_wg.py \ sudo rm -f /usr/local/bin/wirepy /usr/local/bin/wirepy.py /usr/local/bin/start_wg.py \
/usr/local/bin/wp_app_config.py common_tools.py /usr/local/bin/ssl_encrypt.py /usr/local/bin/ssl_decrypt.py /usr/local/bin/wp_app_config.py common_tools.py /usr/local/bin/ssl_encrypt.py \
/usr/local/bin/ssl_decrypt.py /usr/local/bin/match_found.py
if [ $? -ne 0 ] if [ $? -ne 0 ]
then then
exit 0 exit 0

61
match_found.py Executable file
View File

@ -0,0 +1,61 @@
#!/usr/bin/python3
import argparse
from pathlib import Path
directorys: list[str] = [
"/etc/netplan/",
"/etc/NetworkManager/system-connections/",
"/var/lib/NetworkManager/user-connections/",
]
def search_string_in_directory(
directories: list[str] = directorys, # Use the predefined list as default
search_string: str = "", # Default is empty string
) -> bool:
if len(search_string) == 0:
return False
result = False
for directory in directories:
in_paths = Path(directory)
if not in_paths.exists() or not in_paths.is_dir():
continue
files = [file for file in in_paths.iterdir() if file.is_file()]
if not files:
continue
# Search for the string in each file
for file in files:
try:
with open(file, "r", errors="ignore") as f:
for line in f:
if search_string in line:
result = True # String found
break
if result:
break # No need to check further
except Exception:
continue # Skip files that cause errors
# Invert the logic: return False if string is found, True otherwise
return result
def main() -> bool:
parser = argparse.ArgumentParser(
description="Script only for use to compare the private key in the Network configurations to avoid errors with the network manager."
)
parser.add_argument("search_string", help="Search string")
args = parser.parse_args()
result = search_string_in_directory(search_string=args.search_string)
print(result)
if __name__ == "__main__":
main()

View File

@ -38,6 +38,14 @@ License along with this library. If not, see
<allow_active>yes</allow_active> <allow_active>yes</allow_active>
</defaults> </defaults>
<annotate key="org.freedesktop.policykit.exec.path">/usr/local/bin/ssl_decrypt.py</annotate> <annotate key="org.freedesktop.policykit.exec.path">/usr/local/bin/ssl_decrypt.py</annotate>
</action>
<action id="org.match_found">
<defaults>
<allow_any>auth_admin_keep</allow_any>
<allow_inactive>auth_admin_keep</allow_inactive>
<allow_active>yes</allow_active>
</defaults>
<annotate key="org.freedesktop.policykit.exec.path">/usr/local/bin/match_found.py</annotate>
</action> </action>
</policyconfig> </policyconfig>

View File

@ -3,17 +3,16 @@
This script belongs to wirepy and is for the auto start of the tunnel This script belongs to wirepy and is for the auto start of the tunnel
""" """
from pathlib import Path
import subprocess import subprocess
from subprocess import CompletedProcess from subprocess import CompletedProcess
from wp_app_config import AppConfig
from common_tools import ConfigManager
path_to_file = Path(Path.home() / ".config/wire_py/settings") ConfigManager.init(AppConfig.SETTINGS_FILE)
a_con = Path(path_to_file).read_text(encoding="utf-8").splitlines(keepends=True) if ConfigManager.get("autostart") != "off":
a_con = a_con[7].strip()
if a_con != "off":
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "up", a_con], ["nmcli", "connection", "up", ConfigManager.get("autostart")],
capture_output=True, capture_output=True,
text=True, text=True,
check=False, check=False,

446
wirepy.py
View File

@ -4,7 +4,6 @@ this script is a simple GUI for managing Wireguard Tunnels
""" """
import getpass import getpass
import os
import shutil import shutil
import subprocess import subprocess
import sys import sys
@ -17,7 +16,7 @@ from tkinter import TclError, filedialog, ttk
from common_tools import ( from common_tools import (
ConfigManager, ConfigManager,
ThemeManager, ThemeManager,
Create, CryptoUtil,
GiteaUpdate, GiteaUpdate,
Tunnel, Tunnel,
Tooltip, Tooltip,
@ -28,7 +27,7 @@ from wp_app_config import AppConfig, Msg
AppConfig.USER_FILE.write_text(getpass.getuser()) AppConfig.USER_FILE.write_text(getpass.getuser())
AppConfig.ensure_directories() AppConfig.ensure_directories()
AppConfig.create_default_settings() AppConfig.create_default_settings()
Create.decrypt() CryptoUtil.decrypt()
class Wirepy(tk.Tk): class Wirepy(tk.Tk):
@ -242,15 +241,19 @@ class FrameWidgets(ttk.Frame):
self.l_box.configure(yscrollcommand=self.scrollbar.set) self.l_box.configure(yscrollcommand=self.scrollbar.set)
# Tunnel List # Tunnel List
self.tl = LxTools.get_file_name(AppConfig.TEMP_DIR) self.tl = Tunnel.parse_files_to_dictionary(directory=AppConfig.TEMP_DIR)
for tunnels in self.tl: LxTools.clean_files(AppConfig.TEMP_DIR, file=None)
AppConfig.ensure_directories()
# self.tl = LxTools.get_file_name(AppConfig.TEMP_DIR)
for tunnels, values in self.tl.items():
self.l_box.insert("end", tunnels) self.l_box.insert("end", tunnels)
self.l_box.update() self.l_box.update()
# Button Vpn # Button Vpn
if self.a != "": if self.a != "":
self.stop() self.stop()
data = self.handle_tunnel_data(self.a) self.handle_tunnel_data(self.a, self.tl)
self.show_data()
else: else:
self.start() self.start()
@ -586,22 +589,6 @@ class FrameWidgets(ttk.Frame):
else: else:
Tooltip(self.btn_stst, Msg.TTIP["start_tl"], self.tooltip_state) Tooltip(self.btn_stst, Msg.TTIP["start_tl"], self.tooltip_state)
def handle_tunnel_data(self, tunnel_name: str) -> tuple[str, str, str, str | None]:
"""_summary_
Args:
tunnel_name (str): name of a tunnel
Returns:
tuple[str, str]: tuple with tunnel data
"""
wg_read = f"/tmp/tlecdcwg/{tunnel_name}.conf"
with open(wg_read, "r", encoding="utf-8") as file:
data = Tunnel.con_to_dict(file)
self.init_and_report(data)
self.show_data()
return data
def color_label(self) -> None: def color_label(self) -> None:
""" """
View activ Tunnel in the color green or yellow View activ Tunnel in the color green or yellow
@ -649,143 +636,133 @@ class FrameWidgets(ttk.Frame):
try: try:
filepath = filedialog.askopenfilename( filepath = filedialog.askopenfilename(
initialdir=f"{Path.home()}", initialdir=f"{Path.home()}",
title=_("Select Wireguard config File"), title="Select Wireguard config File",
filetypes=[(_("WG config files"), "*.conf")], filetypes=[("WG config files", "*.conf")],
) )
data_import, key_name = Tunnel.parse_files_to_dictionary(filepath=filepath)
# Überprüfe, ob der Benutzer den Dialog abgebrochen hat if CryptoUtil.find_key(f"{data_import[key_name]["PrivateKey"]}="):
if not filepath:
print("File import: abort by user...")
return
with open(filepath, "r", encoding="utf-8") as file:
read = file.read()
path_split = filepath.split("/")
path_split1 = path_split[-1]
if (
"PrivateKey = " in read
and "PublicKey = " in read
and "Endpoint =" in read
):
with open(filepath, "r", encoding="utf-8") as file:
key = Tunnel.con_to_dict(file)
pre_key = key[3]
if len(pre_key) != 0:
p_key = AppConfig.KEYS_FILE.read_text(encoding="utf-8")
if pre_key in p_key or f"{pre_key}\n" in p_key:
LxTools.msg_window(
AppConfig.IMAGE_PATHS["icon_error"],
AppConfig.IMAGE_PATHS["icon_msg"],
Msg.STR["imp_err"],
Msg.STR["tl_exist"],
)
else:
with open(
AppConfig.KEYS_FILE, "a", encoding="utf-8"
) as keyfile:
keyfile.write(f"{pre_key}\r")
if len(path_split1) > 17:
p1 = shutil.copy(filepath, AppConfig.TEMP_DIR)
path_split = path_split1[len(path_split1) - 17 :]
os.rename(p1, f"{AppConfig.TEMP_DIR}/{path_split}")
new_conf = f"{AppConfig.TEMP_DIR}/{path_split}"
if self.a != "":
process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "down", self.a]
)
self.reset_fields()
process: CompletedProcess[str] = subprocess.run(
[
"nmcli",
"connection",
"import",
"type",
"wireguard",
"file",
new_conf,
],
text=True,
)
Create.encrypt()
else:
shutil.copy(filepath, f"{AppConfig.TEMP_DIR}/")
if self.a != "":
process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "down", self.a]
)
self.reset_fields()
process: CompletedProcess[str] = subprocess.run(
[
"nmcli",
"connection",
"import",
"type",
"wireguard",
"file",
filepath,
],
text=True,
)
Create.encrypt()
self.str_var.set("")
self.a = Tunnel.active()
self.l_box.insert(0, self.a)
self.wg_autostart.configure(state="normal")
self.l_box.selection_clear(0, tk.END)
self.l_box.update()
self.l_box.selection_set(0)
Tooltip(
self.wg_autostart,
Msg.TTIP["autostart"],
self.tooltip_state,
x_offset=-10,
y_offset=-40,
)
Tooltip(self.btn_tr, Msg.TTIP["trash_tl"], self.tooltip_state)
Tooltip(self.btn_exp, Msg.TTIP["export_tl"], self.tooltip_state)
Tooltip(
self.btn_rename, Msg.TTIP["rename_tl"], self.tooltip_state
)
self.lb_rename.insert(0, "Max. 12 characters!")
self.str_var = tk.StringVar()
self.str_var.set(self.a)
self.color_label()
self.stop()
data = self.handle_tunnel_data(self.a)
process: CompletedProcess[str] = subprocess.run(
[
"nmcli",
"con",
"mod",
self.a,
"connection.autoconnect",
"no",
]
)
elif ("PrivateKey = " in read) and ("Endpoint = " in read):
pass
else:
LxTools.msg_window( LxTools.msg_window(
AppConfig.IMAGE_PATHS["icon_error"], AppConfig.IMAGE_PATHS["icon_error"],
AppConfig.IMAGE_PATHS["icon_msg"], AppConfig.IMAGE_PATHS["icon_msg"],
Msg.STR["imp_err"], Msg.STR["imp_err"],
Msg.STR["no_valid_file"], Msg.STR["tl_exist"],
) )
elif not CryptoUtil.is_valid_base64(
f"{data_import[key_name]["PrivateKey"]}="
): # 2. Second check: Is it valid Base64?
LxTools.msg_window(
AppConfig.IMAGE_PATHS["icon_error"],
AppConfig.IMAGE_PATHS["icon_msg"],
Msg.STR["imp_err"],
Msg.STR["invalid_base64"],
)
else:
print("Key is valid and does not exist import allowed!")
filepath = Path(filepath)
# Shorten the tunnel name to the maximum allowed length if it exceeds 12 characters.
original_name = filepath.name
truncated_name = (
original_name[-17:] if len(original_name) > 17 else original_name
)
import_file = shutil.copy2(
filepath, AppConfig.TEMP_DIR / truncated_name
)
import_file = Path(import_file)
del data_import[key_name]["PrivateKey"]
self.tl.update(data_import)
if self.a != "":
process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "down", self.a],
capture_output=True,
text=True,
check=False,
)
if process.stderr:
print(process.stderr)
else:
print(f"Error process decrypt: Code {process.returncode}")
self.reset_fields()
process: CompletedProcess[str] = subprocess.run(
[
"nmcli",
"connection",
"import",
"type",
"wireguard",
"file",
import_file,
],
capture_output=True,
text=True,
check=False,
)
if process.stderr:
print(process.stderr)
if process.returncode == 0:
print(f"Tunnel >> {import_file.stem} << import successfull")
else:
print(f"Error process decrypt: Code {process.returncode}")
CryptoUtil.encrypt()
LxTools.clean_files(AppConfig.TEMP_DIR, file=None)
AppConfig.ensure_directories()
self.str_var.set("")
self.a = Tunnel.active()
self.l_box.insert(0, self.a)
self.wg_autostart.configure(state="normal")
self.l_box.selection_clear(0, tk.END)
self.l_box.update()
self.l_box.selection_set(0)
Tooltip(
self.wg_autostart,
Msg.TTIP["autostart"],
self.tooltip_state,
x_offset=-10,
y_offset=-40,
)
Tooltip(self.btn_tr, Msg.TTIP["trash_tl"], self.tooltip_state)
Tooltip(self.btn_exp, Msg.TTIP["export_tl"], self.tooltip_state)
Tooltip(self.btn_rename, Msg.TTIP["rename_tl"], self.tooltip_state)
self.lb_rename.insert(0, "Max. 12 characters!")
self.str_var = tk.StringVar()
self.str_var.set(self.a)
self.color_label()
self.stop()
self.handle_tunnel_data(self.a, self.tl)
self.show_data()
process: CompletedProcess[str] = subprocess.run(
["nmcli", "con", "mod", self.a, "connection.autoconnect", "no"],
capture_output=True,
text=True,
check=False,
)
if process.stderr:
print(process.stderr)
if process.returncode == 0:
print(f">> {import_file.stem} << autostart is disabled by default")
except UnboundLocalError:
LxTools.msg_window(
AppConfig.IMAGE_PATHS["icon_error"],
AppConfig.IMAGE_PATHS["icon_msg"],
Msg.STR["imp_err"],
Msg.STR["no_valid_file"],
)
except IsADirectoryError:
print("File import: abort by user...")
except EOFError as e: except EOFError as e:
print(e) print(e)
except TypeError: except TypeError:
@ -802,37 +779,30 @@ class FrameWidgets(ttk.Frame):
try: try:
self.select_tunnel = self.l_box.curselection() self.select_tunnel = self.l_box.curselection()
select_tl = self.l_box.get(self.select_tunnel[0]) select_tl = self.l_box.get(self.select_tunnel[0])
with open(
f"/tmp/tlecdcwg/{select_tl}.conf", "r+", encoding="utf-8"
) as file2:
key = Tunnel.con_to_dict(file2)
pre_key = key[3]
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "delete", select_tl] ["nmcli", "connection", "delete", select_tl],
capture_output=True,
text=True,
check=False,
) )
if process.stderr:
print(process.stderr)
if process.returncode == 0:
print(f"Tunnel >> {select_tl} << successfully deleted...")
else:
print(f"Error process: Code {process.returncode}")
self.l_box.delete(self.select_tunnel[0]) self.l_box.delete(self.select_tunnel[0])
with open(AppConfig.SETTINGS_FILE, "r", encoding="utf-8") as set_f6: Path.unlink(f"{AppConfig.CONFIG_DIR}/{select_tl}.dat")
lines6 = set_f6.readlines()
if select_tl == lines6[7].strip() and "off\n" not in lines6[7].strip(): if select_tl == ConfigManager.get("autostart"):
lines6[7] = "off\n" ConfigManager.set("autostart", "off")
with open(AppConfig.SETTINGS_FILE, "w", encoding="utf-8") as set_f7: self.selected_option.set(0)
set_f7.writelines(lines6) self.autoconnect_var.set(_("no Autoconnect"))
self.selected_option.set(0)
self.autoconnect_var.set(_("no Autoconnect"))
is_encrypt = Path.home() / f".config/wire_py/{select_tl}.dat"
if is_encrypt.is_file():
Path.unlink(f"{Path.home()}/.config/wire_py/{select_tl}.dat")
Path.unlink(f"/tmp/tlecdcwg/{select_tl}.conf")
with open(AppConfig.KEYS_FILE, "r", encoding="utf-8") as readfile:
with open(
f"{Path.home()}/.config/wire_py/keys2", "w", encoding="utf-8"
) as writefile:
for line in readfile:
if pre_key not in line.strip("\n"):
writefile.write(line)
file_one = Path(f"{Path.home()}/.config/wire_py/keys2")
file_two = file_one.with_name("keys")
file_one.replace(file_two)
self.wg_autostart.configure(state="disabled") self.wg_autostart.configure(state="disabled")
# for disabling checkbox when Listbox empty # for disabling checkbox when Listbox empty
@ -895,22 +865,18 @@ class FrameWidgets(ttk.Frame):
Set (on), the selected tunnel is displayed in the label. Set (on), the selected tunnel is displayed in the label.
At (off) the label is first emptied then filled with No Autoconnect At (off) the label is first emptied then filled with No Autoconnect
""" """
lines = (
Path(AppConfig.SETTINGS_FILE)
.read_text(encoding="utf-8")
.splitlines(keepends=True)
)
if lines[7] != "off\n": if ConfigManager.get("autostart") != "off":
print(f"{lines[7]} starts automatically when the system starts.") print(
f"{ConfigManager.get("autostart")} starts automatically when the system starts."
)
self.selected_option.set(1) self.selected_option.set(1)
self.autoconnect_var.set("") self.autoconnect_var.set("")
self.auto_con = lines[7] self.auto_con = ConfigManager.get("autostart")
else: else:
self.selected_option.set(0) self.selected_option.set(0)
self.auto_con = _("no Autoconnect") self.auto_con = _("no Autoconnect")
print("Autostart disabled.")
self.autoconnect_var.set("") self.autoconnect_var.set("")
self.autoconnect_var = tk.StringVar() self.autoconnect_var = tk.StringVar()
self.autoconnect_var.set(self.auto_con) self.autoconnect_var.set(self.auto_con)
@ -940,31 +906,15 @@ class FrameWidgets(ttk.Frame):
select_tl = self.l_box.get(select_tunnel[0]) select_tl = self.l_box.get(select_tunnel[0])
if self.selected_option.get() == 0: if self.selected_option.get() == 0:
lines = ( ConfigManager.set("autostart", "off")
Path(AppConfig.SETTINGS_FILE)
.read_text(encoding="utf-8")
.splitlines(keepends=True)
)
lines[7] = "off\n"
Path(AppConfig.SETTINGS_FILE).write_text(
"".join(lines), encoding="utf-8"
)
tl = LxTools.get_file_name(AppConfig.TEMP_DIR) tl = [f"{file}" for file in AppConfig.CONFIG_DIR.glob("*.dat")]
if len(tl) == 0: if len(tl) == 0:
self.wg_autostart.configure(state="disabled") self.wg_autostart.configure(state="disabled")
if self.selected_option.get() >= 1: if self.selected_option.get() >= 1:
lines = ( ConfigManager.set("autostart", select_tl)
Path(AppConfig.SETTINGS_FILE)
.read_text(encoding="utf-8")
.splitlines(keepends=True)
)
lines[7] = select_tl
Path(AppConfig.SETTINGS_FILE).write_text(
"".join(lines), encoding="utf-8"
)
except IndexError: except IndexError:
self.selected_option.set(1) self.selected_option.set(1)
@ -1021,7 +971,7 @@ class FrameWidgets(ttk.Frame):
select_tl = self.l_box.get(self.select_tunnel[0]) select_tl = self.l_box.get(self.select_tunnel[0])
# nmcli connection modify old connection.id iphone # nmcli connection modify old connection.id iphone
subprocess.check_output( process: CompletedProcess[str] = subprocess.run(
[ [
"nmcli", "nmcli",
"connection", "connection",
@ -1030,30 +980,28 @@ class FrameWidgets(ttk.Frame):
"connection.id", "connection.id",
self.lb_rename.get(), self.lb_rename.get(),
], ],
capture_output=True,
text=True, text=True,
check=False,
) )
source = Path(f"/tmp/tlecdcwg/{select_tl}.conf") if process.stderr:
destination = source.with_name(f"{self.lb_rename.get()}.conf") print(process.stderr)
source.replace(destination)
Path.unlink(f"{Path.home()}/.config/wire_py/{select_tl}.dat")
if process.returncode != 0:
print(f"Error process: Code {process.returncode}")
source = Path(f"{AppConfig.CONFIG_DIR}/{select_tl}.dat")
destination = AppConfig.CONFIG_DIR / f"{self.lb_rename.get()}.dat"
source.replace(destination)
self.tl[self.lb_rename.get()] = self.tl.pop(select_tl)
if select_tl == ConfigManager.get("autostart"):
ConfigManager.set("autostart", self.lb_rename.get())
self.autoconnect_var.set(value=self.lb_rename.get())
self.l_box.delete(self.select_tunnel[0]) self.l_box.delete(self.select_tunnel[0])
self.l_box.insert("end", self.lb_rename.get()) self.l_box.insert("end", self.lb_rename.get())
self.l_box.update() self.l_box.update()
new_a_connect = self.lb_rename.get()
self.lb_rename.delete(0, tk.END) self.lb_rename.delete(0, tk.END)
with open(AppConfig.SETTINGS_FILE, "r", encoding="utf-8") as set_f5:
lines5 = set_f5.readlines()
if select_tl == lines5[7].strip() and "off\n" not in lines5[7].strip():
lines5[7] = new_a_connect
with open(
AppConfig.SETTINGS_FILE, "w", encoding="utf-8"
) as theme_set5:
theme_set5.writelines(lines5)
self.autoconnect_var.set(value=new_a_connect)
self.update_connection_display() self.update_connection_display()
Create.encrypt()
except IndexError: except IndexError:
@ -1070,18 +1018,17 @@ class FrameWidgets(ttk.Frame):
except EOFError as e: except EOFError as e:
print(e) print(e)
def init_and_report(self, data=None) -> None: def handle_tunnel_data(self, active=None, data=None) -> None:
"""
Displays the value address, DNS and peer in the labels tunnel = active
or empty it again values = data[tunnel]
"""
# Address Label # Address Label
self.add = tk.StringVar() self.add = tk.StringVar()
self.add.set(f"{_("Address: ")}{data[0]}") self.add.set(f"Address: {values['Address']}")
self.DNS = tk.StringVar() self.DNS = tk.StringVar()
self.DNS.set(f" DNS: {data[1]}") self.DNS.set(f" DNS: {values['DNS']}")
self.enp = tk.StringVar() self.enp = tk.StringVar()
self.enp.set(f"{_("Endpoint: ")}{data[2]}") self.enp.set(f"Endpoint: {values['Endpoint']}")
def show_data(self) -> None: def show_data(self) -> None:
""" """
@ -1118,10 +1065,8 @@ class FrameWidgets(ttk.Frame):
else: else:
data = self.handle_tunnel_data(self.a) self.handle_tunnel_data(self.a, self.tl)
if data: self.handle_connection_state("stop")
self.handle_connection_state("stop")
except IndexError: except IndexError:
@ -1154,8 +1099,18 @@ class FrameWidgets(ttk.Frame):
if action == "stop": if action == "stop":
if self.a: if self.a:
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "down", self.a] ["nmcli", "connection", "down", self.a],
capture_output=True,
text=True,
check=False,
) )
if process.stderr:
print(process.stderr)
if process.returncode != 0:
print(f"Error process: Code {process.returncode}")
self.update_connection_display() self.update_connection_display()
self.reset_fields() self.reset_fields()
self.start() self.start()
@ -1164,11 +1119,22 @@ class FrameWidgets(ttk.Frame):
if tunnel_name or self.a: if tunnel_name or self.a:
target_tunnel = tunnel_name or self.a target_tunnel = tunnel_name or self.a
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "up", target_tunnel] ["nmcli", "connection", "up", target_tunnel],
capture_output=True,
text=True,
check=False,
) )
if process.stderr:
print(process.stderr)
if process.returncode == 0:
print(f"Tunnel >> {target_tunnel} << started")
else:
print(f"Error process: Code {process.returncode}")
self.update_connection_display() self.update_connection_display()
data = self.handle_tunnel_data(self.a) self.handle_tunnel_data(self.a, self.tl)
self.init_and_report(data)
self.show_data() self.show_data()
self.color_label() self.color_label()
self.stop() self.stop()

View File

@ -24,7 +24,6 @@ class AppConfig:
# Configuration files # Configuration files
SETTINGS_FILE: Path = CONFIG_DIR / "settings" SETTINGS_FILE: Path = CONFIG_DIR / "settings"
KEYS_FILE: Path = CONFIG_DIR / "keys"
SYSTEMD_USER_FOLDER: Path = Path.home() / ".config/systemd/user" SYSTEMD_USER_FOLDER: Path = Path.home() / ".config/systemd/user"
AUTOSTART_SERVICE: Path = Path.home() / ".config/systemd/user/wg_start.service" AUTOSTART_SERVICE: Path = Path.home() / ".config/systemd/user/wg_start.service"
DEFAULT_SETTINGS: Dict[str, str] = { DEFAULT_SETTINGS: Dict[str, str] = {
@ -57,13 +56,6 @@ class AppConfig:
"pkey_path": "/usr/local/etc/ssl/pwgk.pem", "pkey_path": "/usr/local/etc/ssl/pwgk.pem",
} }
# Lists of searches
DIRECTORYS: list[str] = [
"/etc/netplan/",
"/etc/NetworkManager/system-connections/",
"/var/lib/NetworkManager/user-connections/",
]
# Images and icons paths # Images and icons paths
IMAGE_PATHS: Dict[str, str] = { IMAGE_PATHS: Dict[str, str] = {
"icon_vpn": "/usr/share/icons/lx-icons/48/wg_vpn.png", "icon_vpn": "/usr/share/icons/lx-icons/48/wg_vpn.png",
@ -96,10 +88,7 @@ class AppConfig:
"""Ensures that all required directories exist""" """Ensures that all required directories exist"""
if not cls.CONFIG_DIR.exists(): if not cls.CONFIG_DIR.exists():
cls.CONFIG_DIR.mkdir(parents=True, exist_ok=True) cls.CONFIG_DIR.mkdir(parents=True, exist_ok=True)
cls.KEYS_FILE.touch()
cls.TEMP_DIR.mkdir(parents=True, exist_ok=True) cls.TEMP_DIR.mkdir(parents=True, exist_ok=True)
if not cls.KEYS_FILE.exists():
cls.KEYS_FILE.touch()
@classmethod @classmethod
def create_default_settings(cls) -> None: def create_default_settings(cls) -> None:
@ -187,7 +176,7 @@ class Msg:
"sel_list": _("Please select a tunnel from the list"), "sel_list": _("Please select a tunnel from the list"),
"sign_len": _("The new name may contain only 12 characters"), "sign_len": _("The new name may contain only 12 characters"),
"zero_signs": _("At least one character must be entered"), "zero_signs": _("At least one character must be entered"),
"false signs": _( "false_signs": _(
"No valid sign. These must not be used.\nBlank, Slash, Backslash and { }\n" "No valid sign. These must not be used.\nBlank, Slash, Backslash and { }\n"
), ),
"is_in_use": _("The tunnel is already in use"), "is_in_use": _("The tunnel is already in use"),
@ -195,6 +184,9 @@ class Msg:
"Oh... no valid Wireguard File!\nPlease select a valid Wireguard File" "Oh... no valid Wireguard File!\nPlease select a valid Wireguard File"
), ),
"tl_exist": _("Tunnel already available!\nPlease use another file for import"), "tl_exist": _("Tunnel already available!\nPlease use another file for import"),
"invalid_base64": _(
"Invalid base64 format!\nPlease use a Config file with valid key."
),
} }
TTIP: Dict[str, str] = { TTIP: Dict[str, str] = {
# Strings for Tooltips # Strings for Tooltips