Compare commits
2 Commits
0c4d000d96
...
55f2119bc3
Author | SHA1 | Date | |
---|---|---|---|
55f2119bc3 | |||
d6c20b81f9 |
253
common_tools.py
253
common_tools.py
@ -3,14 +3,17 @@
|
||||
import os
|
||||
import shutil
|
||||
import signal
|
||||
import base64
|
||||
import secrets
|
||||
import subprocess
|
||||
from subprocess import CompletedProcess, run
|
||||
import re
|
||||
import sys
|
||||
import tkinter as tk
|
||||
from typing import Optional, Dict, Any, NoReturn, TextIO, Tuple, List
|
||||
from typing import Optional, Dict, Any, NoReturn, List
|
||||
import zipfile
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from subprocess import CompletedProcess
|
||||
from tkinter import ttk, Toplevel
|
||||
from wp_app_config import AppConfig, Msg
|
||||
import requests
|
||||
@ -19,7 +22,7 @@ import requests
|
||||
_ = AppConfig.setup_translations()
|
||||
|
||||
|
||||
class Create:
|
||||
class CryptoUtil:
|
||||
"""
|
||||
This class is for the creation of the folders and files
|
||||
required by Wire-Py, as well as for decryption
|
||||
@ -31,6 +34,9 @@ class Create:
|
||||
"""
|
||||
Starts SSL dencrypt
|
||||
"""
|
||||
crypted_tunnel = [str(file) for file in AppConfig.CONFIG_DIR.glob("*.dat")]
|
||||
if crypted_tunnel == []:
|
||||
return
|
||||
process: CompletedProcess[str] = subprocess.run(
|
||||
["pkexec", "/usr/local/bin/ssl_decrypt.py"],
|
||||
capture_output=True,
|
||||
@ -38,10 +44,6 @@ class Create:
|
||||
check=False,
|
||||
)
|
||||
|
||||
# Output from Openssl
|
||||
# if process.stdout:
|
||||
# print(process.stdout)
|
||||
|
||||
# Output from Openssl Error
|
||||
if process.stderr:
|
||||
print(process.stderr)
|
||||
@ -72,6 +74,52 @@ class Create:
|
||||
else:
|
||||
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):
|
||||
"""
|
||||
@ -81,57 +129,6 @@ class LxTools(tk.Tk):
|
||||
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
||||
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
|
||||
def center_window_cross_platform(window, width, height):
|
||||
"""
|
||||
@ -405,23 +402,73 @@ class Tunnel:
|
||||
"""
|
||||
|
||||
@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 = {}
|
||||
|
||||
if not AppConfig.TEMP_DIR.exists() or not AppConfig.TEMP_DIR.is_dir():
|
||||
pass
|
||||
|
||||
# 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:
|
||||
if filepath is not None:
|
||||
filepath = Path(filepath)
|
||||
try:
|
||||
with open(file, "r") as f:
|
||||
content = f.read()
|
||||
# Hier parsen wir die relevanten Zeilen aus dem Inhalt
|
||||
content = filepath.read_text()
|
||||
|
||||
# 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(
|
||||
line
|
||||
for line in content.splitlines()
|
||||
@ -436,67 +483,27 @@ class Tunnel:
|
||||
if line.startswith("Endpoint")
|
||||
)
|
||||
|
||||
# Extrahiere die Werte
|
||||
# extract values
|
||||
address = address_line.split("=")[1].strip()
|
||||
dns = dns_line.split("=")[1].strip()
|
||||
endpoint = endpoint_line.split("=")[1].strip()
|
||||
|
||||
# Speichere im Dictionary
|
||||
# save values to dictionary
|
||||
data[file.stem] = {
|
||||
"Address": address,
|
||||
"DNS": dns,
|
||||
"Endpoint": endpoint,
|
||||
}
|
||||
except Exception:
|
||||
# Ignore errors and continue to the next file
|
||||
continue
|
||||
|
||||
return data
|
||||
|
||||
@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
|
||||
except Exception:
|
||||
# Ignore errors and continue to the next file
|
||||
continue
|
||||
if content is not None:
|
||||
content = secrets.token_bytes(len(content))
|
||||
if filepath is not None:
|
||||
return data, truncated_stem
|
||||
else:
|
||||
return data
|
||||
|
||||
@staticmethod
|
||||
def active() -> str:
|
||||
|
7
install
7
install
@ -18,6 +18,7 @@ install_file_with(){
|
||||
else
|
||||
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 match_found.py /usr/local/bin/ && \
|
||||
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 -fv Wire-Py.desktop /usr/share/applications/ && \
|
||||
@ -44,6 +45,7 @@ install_arch_d(){
|
||||
else
|
||||
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 match_found.py /usr/local/bin/ && \
|
||||
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 -fv Wire-Py.desktop /usr/share/applications/ && \
|
||||
@ -121,6 +123,7 @@ install(){
|
||||
else
|
||||
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 match_found.py /usr/local/bin/ && \
|
||||
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 -fv Wire-Py.desktop /usr/share/applications/ && \
|
||||
@ -146,6 +149,7 @@ install(){
|
||||
exit 0
|
||||
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 match_found.py /usr/local/bin/ && \
|
||||
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 -fv Wire-Py.desktop /usr/share/applications/ && \
|
||||
@ -181,7 +185,8 @@ install(){
|
||||
|
||||
remove(){
|
||||
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 ]
|
||||
then
|
||||
exit 0
|
||||
|
61
match_found.py
Executable file
61
match_found.py
Executable 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()
|
@ -38,6 +38,14 @@ License along with this library. If not, see
|
||||
<allow_active>yes</allow_active>
|
||||
</defaults>
|
||||
<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>
|
||||
</policyconfig>
|
11
start_wg.py
11
start_wg.py
@ -3,17 +3,16 @@
|
||||
This script belongs to wirepy and is for the auto start of the tunnel
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
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)
|
||||
a_con = a_con[7].strip()
|
||||
if a_con != "off":
|
||||
if ConfigManager.get("autostart") != "off":
|
||||
process: CompletedProcess[str] = subprocess.run(
|
||||
["nmcli", "connection", "up", a_con],
|
||||
["nmcli", "connection", "up", ConfigManager.get("autostart")],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
|
446
wirepy.py
446
wirepy.py
@ -4,7 +4,6 @@ this script is a simple GUI for managing Wireguard Tunnels
|
||||
"""
|
||||
|
||||
import getpass
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
@ -17,7 +16,7 @@ from tkinter import TclError, filedialog, ttk
|
||||
from common_tools import (
|
||||
ConfigManager,
|
||||
ThemeManager,
|
||||
Create,
|
||||
CryptoUtil,
|
||||
GiteaUpdate,
|
||||
Tunnel,
|
||||
Tooltip,
|
||||
@ -28,7 +27,7 @@ from wp_app_config import AppConfig, Msg
|
||||
AppConfig.USER_FILE.write_text(getpass.getuser())
|
||||
AppConfig.ensure_directories()
|
||||
AppConfig.create_default_settings()
|
||||
Create.decrypt()
|
||||
CryptoUtil.decrypt()
|
||||
|
||||
|
||||
class Wirepy(tk.Tk):
|
||||
@ -242,15 +241,19 @@ class FrameWidgets(ttk.Frame):
|
||||
self.l_box.configure(yscrollcommand=self.scrollbar.set)
|
||||
|
||||
# Tunnel List
|
||||
self.tl = LxTools.get_file_name(AppConfig.TEMP_DIR)
|
||||
for tunnels in self.tl:
|
||||
self.tl = Tunnel.parse_files_to_dictionary(directory=AppConfig.TEMP_DIR)
|
||||
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.update()
|
||||
|
||||
# Button Vpn
|
||||
if self.a != "":
|
||||
self.stop()
|
||||
data = self.handle_tunnel_data(self.a)
|
||||
self.handle_tunnel_data(self.a, self.tl)
|
||||
self.show_data()
|
||||
else:
|
||||
self.start()
|
||||
|
||||
@ -586,22 +589,6 @@ class FrameWidgets(ttk.Frame):
|
||||
else:
|
||||
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:
|
||||
"""
|
||||
View activ Tunnel in the color green or yellow
|
||||
@ -649,143 +636,133 @@ class FrameWidgets(ttk.Frame):
|
||||
try:
|
||||
filepath = filedialog.askopenfilename(
|
||||
initialdir=f"{Path.home()}",
|
||||
title=_("Select Wireguard config File"),
|
||||
filetypes=[(_("WG config files"), "*.conf")],
|
||||
title="Select Wireguard config File",
|
||||
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 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:
|
||||
if CryptoUtil.find_key(f"{data_import[key_name]["PrivateKey"]}="):
|
||||
LxTools.msg_window(
|
||||
AppConfig.IMAGE_PATHS["icon_error"],
|
||||
AppConfig.IMAGE_PATHS["icon_msg"],
|
||||
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:
|
||||
print(e)
|
||||
except TypeError:
|
||||
@ -802,37 +779,30 @@ class FrameWidgets(ttk.Frame):
|
||||
try:
|
||||
self.select_tunnel = self.l_box.curselection()
|
||||
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(
|
||||
["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])
|
||||
with open(AppConfig.SETTINGS_FILE, "r", encoding="utf-8") as set_f6:
|
||||
lines6 = set_f6.readlines()
|
||||
if select_tl == lines6[7].strip() and "off\n" not in lines6[7].strip():
|
||||
lines6[7] = "off\n"
|
||||
with open(AppConfig.SETTINGS_FILE, "w", encoding="utf-8") as set_f7:
|
||||
set_f7.writelines(lines6)
|
||||
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)
|
||||
Path.unlink(f"{AppConfig.CONFIG_DIR}/{select_tl}.dat")
|
||||
|
||||
if select_tl == ConfigManager.get("autostart"):
|
||||
ConfigManager.set("autostart", "off")
|
||||
self.selected_option.set(0)
|
||||
self.autoconnect_var.set(_("no Autoconnect"))
|
||||
|
||||
self.wg_autostart.configure(state="disabled")
|
||||
|
||||
# for disabling checkbox when Listbox empty
|
||||
@ -895,22 +865,18 @@ class FrameWidgets(ttk.Frame):
|
||||
Set (on), the selected tunnel is displayed in the label.
|
||||
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":
|
||||
print(f"{lines[7]} starts automatically when the system starts.")
|
||||
if ConfigManager.get("autostart") != "off":
|
||||
print(
|
||||
f"{ConfigManager.get("autostart")} starts automatically when the system starts."
|
||||
)
|
||||
self.selected_option.set(1)
|
||||
self.autoconnect_var.set("")
|
||||
self.auto_con = lines[7]
|
||||
self.auto_con = ConfigManager.get("autostart")
|
||||
|
||||
else:
|
||||
self.selected_option.set(0)
|
||||
self.auto_con = _("no Autoconnect")
|
||||
print("Autostart disabled.")
|
||||
self.autoconnect_var.set("")
|
||||
self.autoconnect_var = tk.StringVar()
|
||||
self.autoconnect_var.set(self.auto_con)
|
||||
@ -940,31 +906,15 @@ class FrameWidgets(ttk.Frame):
|
||||
select_tl = self.l_box.get(select_tunnel[0])
|
||||
|
||||
if self.selected_option.get() == 0:
|
||||
lines = (
|
||||
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"
|
||||
)
|
||||
ConfigManager.set("autostart", "off")
|
||||
|
||||
tl = LxTools.get_file_name(AppConfig.TEMP_DIR)
|
||||
tl = [f"{file}" for file in AppConfig.CONFIG_DIR.glob("*.dat")]
|
||||
|
||||
if len(tl) == 0:
|
||||
self.wg_autostart.configure(state="disabled")
|
||||
|
||||
if self.selected_option.get() >= 1:
|
||||
lines = (
|
||||
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"
|
||||
)
|
||||
ConfigManager.set("autostart", select_tl)
|
||||
|
||||
except IndexError:
|
||||
self.selected_option.set(1)
|
||||
@ -1021,7 +971,7 @@ class FrameWidgets(ttk.Frame):
|
||||
select_tl = self.l_box.get(self.select_tunnel[0])
|
||||
|
||||
# nmcli connection modify old connection.id iphone
|
||||
subprocess.check_output(
|
||||
process: CompletedProcess[str] = subprocess.run(
|
||||
[
|
||||
"nmcli",
|
||||
"connection",
|
||||
@ -1030,30 +980,28 @@ class FrameWidgets(ttk.Frame):
|
||||
"connection.id",
|
||||
self.lb_rename.get(),
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
source = Path(f"/tmp/tlecdcwg/{select_tl}.conf")
|
||||
destination = source.with_name(f"{self.lb_rename.get()}.conf")
|
||||
source.replace(destination)
|
||||
Path.unlink(f"{Path.home()}/.config/wire_py/{select_tl}.dat")
|
||||
if process.stderr:
|
||||
print(process.stderr)
|
||||
|
||||
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.insert("end", self.lb_rename.get())
|
||||
self.l_box.update()
|
||||
new_a_connect = self.lb_rename.get()
|
||||
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()
|
||||
Create.encrypt()
|
||||
|
||||
except IndexError:
|
||||
|
||||
@ -1070,18 +1018,17 @@ class FrameWidgets(ttk.Frame):
|
||||
except EOFError as e:
|
||||
print(e)
|
||||
|
||||
def init_and_report(self, data=None) -> None:
|
||||
"""
|
||||
Displays the value address, DNS and peer in the labels
|
||||
or empty it again
|
||||
"""
|
||||
def handle_tunnel_data(self, active=None, data=None) -> None:
|
||||
|
||||
tunnel = active
|
||||
values = data[tunnel]
|
||||
# Address Label
|
||||
self.add = tk.StringVar()
|
||||
self.add.set(f"{_("Address: ")}{data[0]}")
|
||||
self.add.set(f"Address: {values['Address']}")
|
||||
self.DNS = tk.StringVar()
|
||||
self.DNS.set(f" DNS: {data[1]}")
|
||||
self.DNS.set(f" DNS: {values['DNS']}")
|
||||
self.enp = tk.StringVar()
|
||||
self.enp.set(f"{_("Endpoint: ")}{data[2]}")
|
||||
self.enp.set(f"Endpoint: {values['Endpoint']}")
|
||||
|
||||
def show_data(self) -> None:
|
||||
"""
|
||||
@ -1118,10 +1065,8 @@ class FrameWidgets(ttk.Frame):
|
||||
|
||||
else:
|
||||
|
||||
data = self.handle_tunnel_data(self.a)
|
||||
if data:
|
||||
|
||||
self.handle_connection_state("stop")
|
||||
self.handle_tunnel_data(self.a, self.tl)
|
||||
self.handle_connection_state("stop")
|
||||
|
||||
except IndexError:
|
||||
|
||||
@ -1154,8 +1099,18 @@ class FrameWidgets(ttk.Frame):
|
||||
if action == "stop":
|
||||
if self.a:
|
||||
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.reset_fields()
|
||||
self.start()
|
||||
@ -1164,11 +1119,22 @@ class FrameWidgets(ttk.Frame):
|
||||
if tunnel_name or self.a:
|
||||
target_tunnel = tunnel_name or self.a
|
||||
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()
|
||||
data = self.handle_tunnel_data(self.a)
|
||||
self.init_and_report(data)
|
||||
self.handle_tunnel_data(self.a, self.tl)
|
||||
self.show_data()
|
||||
self.color_label()
|
||||
self.stop()
|
||||
|
@ -24,7 +24,6 @@ class AppConfig:
|
||||
|
||||
# Configuration files
|
||||
SETTINGS_FILE: Path = CONFIG_DIR / "settings"
|
||||
KEYS_FILE: Path = CONFIG_DIR / "keys"
|
||||
SYSTEMD_USER_FOLDER: Path = Path.home() / ".config/systemd/user"
|
||||
AUTOSTART_SERVICE: Path = Path.home() / ".config/systemd/user/wg_start.service"
|
||||
DEFAULT_SETTINGS: Dict[str, str] = {
|
||||
@ -57,13 +56,6 @@ class AppConfig:
|
||||
"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
|
||||
IMAGE_PATHS: Dict[str, str] = {
|
||||
"icon_vpn": "/usr/share/icons/lx-icons/48/wg_vpn.png",
|
||||
@ -96,10 +88,7 @@ class AppConfig:
|
||||
"""Ensures that all required directories exist"""
|
||||
if not cls.CONFIG_DIR.exists():
|
||||
cls.CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
cls.KEYS_FILE.touch()
|
||||
cls.TEMP_DIR.mkdir(parents=True, exist_ok=True)
|
||||
if not cls.KEYS_FILE.exists():
|
||||
cls.KEYS_FILE.touch()
|
||||
|
||||
@classmethod
|
||||
def create_default_settings(cls) -> None:
|
||||
@ -187,7 +176,7 @@ class Msg:
|
||||
"sel_list": _("Please select a tunnel from the list"),
|
||||
"sign_len": _("The new name may contain only 12 characters"),
|
||||
"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"
|
||||
),
|
||||
"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"
|
||||
),
|
||||
"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] = {
|
||||
# Strings for Tooltips
|
||||
|
Reference in New Issue
Block a user