punix/Wire-Py
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases 13 Wiki Activity

04-06-2025_large_update #35

Merged
punix merged 61 commits from 04-06-2025_large_update into main 2025-06-04 19:34:40 +02:00
Conversation 0 Commits 61 Files Changed 43 +98 -129
5 changed files with 98 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit fb0158d1cd - Show all commits

80
common_tools.py
View File

@@ -10,7 +10,7 @@ from typing import Optional, Dict, Any, NoReturn, TextIO, Tuple, List
import zipfile import zipfile
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from subprocess import check_call, CompletedProcess 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
@@ -26,63 +26,6 @@ class Create:
the tunnel from the user's home directory 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"
AppConfig.KEYS_FILE
if sett.exists():
pass
else:
sett.touch()
sett.write_text(
"[UPDATES]\non\n[THEME]\nlight\n[TOOLTIP]\nTrue\n[AUTOSTART ON]\noff\n"
)
if AppConfig.KEYS_FILE.exists():
pass
else:
AppConfig.KEYS_FILE.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"""
if AppConfig.TEMP_DIR.exists():
pass
else:
AppConfig.TEMP_DIR.mkdir()
@staticmethod @staticmethod
def decrypt() -> str: def decrypt() -> str:
""" """
@@ -96,17 +39,17 @@ class Create:
) )
# Output from Openssl # Output from Openssl
if process.stdout: # if process.stdout:
print(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)
if process.returncode == 0: if process.returncode == 0:
print("Datei entschlüsseln wurde erfolgreich entschlossen.") print("Files successfully decrypted...")
else: else:
print(f"Fehler bei der Verarbeitung von Dateien: Code {process.returncode}") print(f"Error process decrypt: Code {process.returncode}")
@staticmethod @staticmethod
def encrypt() -> str: def encrypt() -> str:
@@ -115,15 +58,18 @@ class Create:
""" """
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = subprocess.run(
["pkexec", "/usr/local/bin/ssl_encrypt.py"], ["pkexec", "/usr/local/bin/ssl_encrypt.py"],
stdout=subprocess.PIPE, capture_output=True,
text=True, text=True,
check=True, check=False,
) )
print(process.stdout) # Output from Openssl Error
if process.stderr:
print(process.stderr)
if process.returncode == 0: if process.returncode == 0:
print("All Files successfully encrypted...") print("Files successfully encrypted...")
else: else:
print(f"Error with the following code... {process.returncode}") print(f"Error process encrypt: Code {process.returncode}")
class LxTools(tk.Tk): class LxTools(tk.Tk):

102
ssl_encrypt.py
View File

@@ -1,12 +1,11 @@
#!/usr/bin/python3 #!/usr/bin/python3
""" This Script encrypt Wireguardfiles for Wirepy users for more Security """ """ This Script encrypt Wireguardfiles for Wirepy users for more Security """
import os
from pathlib import Path from pathlib import Path
import shutil import shutil
from subprocess import check_call import subprocess
from subprocess import CompletedProcess
from wp_app_config import AppConfig from wp_app_config import AppConfig
from common_tools import LxTools
log_name = AppConfig.USER_FILE.read_text() log_name = AppConfig.USER_FILE.read_text()
@@ -14,7 +13,7 @@ keyfile: Path = Path(f"/home/{log_name}/.config/wire_py/pbwgk.pem")
if not keyfile.is_file(): if not keyfile.is_file():
check_call( process: CompletedProcess[str] = subprocess.run(
[ [
"openssl", "openssl",
"rsa", "rsa",
@@ -25,56 +24,59 @@ if not keyfile.is_file():
"-outform", "-outform",
"PEM", "PEM",
"-pubout", "-pubout",
] ],
capture_output=True,
text=True,
check=False,
) )
if process.stdout:
print(process.stdout)
# Output from Openssl Error
if process.stderr:
print("(Error):", process.stderr)
if process.returncode == 0:
print("Public key generated successfully.")
else:
print(f"Error generate Publickey: Code: {process.returncode}")
shutil.chown(keyfile, 1000, 1000) shutil.chown(keyfile, 1000, 1000)
if AppConfig.TEMP_DIR.exists(): if AppConfig.TEMP_DIR.exists() and not any(AppConfig.TEMP_DIR.iterdir()):
tl = LxTools.get_file_name(AppConfig.TEMP_DIR) clear_files = [str(file) for file in path_of_crypted_tunnel.glob()]
CPTH: str = f"{keyfile}"
CRYPTFILES: str = CPTH[:-9]
if keyfile.exists() and len(tl) != 0: for config_file in clear_files:
for tunnels in tl: base_name = Path(config_file).stem
sourcetl: str = f"{AppConfig.TEMP_DIR}/{tunnels}" process: CompletedProcess[str] = subprocess.run(
tlname: str = f"{CRYPTFILES}{tunnels[:-5]}.dat" [
check_call( "openssl",
[ "pkeyutl",
"openssl", "-encrypt",
"pkeyutl", "-inkey",
"-encrypt", keyfile,
"-inkey", "-pubin",
keyfile, "-in",
"-pubin", config_file,
"-in", "-out",
sourcetl, f"{AppConfig.CONFIG_DIR}/{base_name}.dat",
"-out", ],
tlname, capture_output=True,
] text=True,
) check=False,
)
else: print(f"Processing of the file: {config_file}")
if AppConfig.TEMP_DIR.exists(): if process.stdout:
tl: list[str] = os.listdir(f"{AppConfig.TEMP_DIR}") print(process.stdout)
CPTH: str = f"{keyfile}"
CRYPTFILES: str = CPTH[:-9]
if keyfile.exists() and len(tl) != 0: # Output from Openssl Error
for tunnels in tl: if process.stderr:
sourcetl: str = f"{AppConfig.TEMP_DIR}/{tunnels}" print("(Error):", process.stderr)
tlname: str = f"{CRYPTFILES}{tunnels[:-5]}.dat"
check_call( if process.returncode == 0:
[ print(f"File {base_name}.dat successfully encrypted.")
"openssl", else:
"pkeyutl", print(f"Error by {config_file}: Code: {process.returncode}")
"-encrypt",
"-inkey",
keyfile,
"-pubin",
"-in",
sourcetl,
"-out",
tlname,
]
)

14
start_wg.py
View File

@@ -4,13 +4,23 @@
""" """
from pathlib import Path from pathlib import Path
from subprocess import check_call import subprocess
from subprocess import CompletedProcess
path_to_file = Path(Path.home() / ".config/wire_py/settings") path_to_file = Path(Path.home() / ".config/wire_py/settings")
a_con = Path(path_to_file).read_text(encoding="utf-8").splitlines(keepends=True) a_con = Path(path_to_file).read_text(encoding="utf-8").splitlines(keepends=True)
a_con = a_con[7].strip() a_con = a_con[7].strip()
if a_con != "off": if a_con != "off":
check_call(["nmcli", "connection", "up", a_con]) process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "up", a_con],
capture_output=True,
text=True,
check=False,
)
# Output from start_wg error
if process.stderr:
print(process.stderr) # this is for the error, later on logfile
else: else:
pass pass

1
wg_start.service
View File

@@ -6,5 +6,6 @@ After=network-online.target
Type=oneshot Type=oneshot
ExecStartPre=/bin/sleep 5 ExecStartPre=/bin/sleep 5
ExecStart=/usr/local/bin/start_wg.py ExecStart=/usr/local/bin/start_wg.py
[Install] [Install]
WantedBy=default.target WantedBy=default.target

30
wirepy.py
View File

@@ -11,7 +11,7 @@ import sys
import tkinter as tk import tkinter as tk
import webbrowser import webbrowser
from pathlib import Path from pathlib import Path
from subprocess import check_call from subprocess import CompletedProcess
from tkinter import TclError, filedialog, ttk from tkinter import TclError, filedialog, ttk
from common_tools import ( from common_tools import (
@@ -645,7 +645,7 @@ class FrameWidgets(ttk.Frame):
def import_sl(self) -> None: def import_sl(self) -> None:
"""validity check of wireguard config files""" """validity check of wireguard config files"""
Create.dir_and_files() AppConfig.ensure_directories()
try: try:
filepath = filedialog.askopenfilename( filepath = filedialog.askopenfilename(
initialdir=f"{Path.home()}", initialdir=f"{Path.home()}",
@@ -696,10 +696,12 @@ class FrameWidgets(ttk.Frame):
new_conf = f"{AppConfig.TEMP_DIR}/{path_split}" new_conf = f"{AppConfig.TEMP_DIR}/{path_split}"
if self.a != "": if self.a != "":
check_call(["nmcli", "connection", "down", self.a]) process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "down", self.a]
)
self.reset_fields() self.reset_fields()
subprocess.check_output( process: CompletedProcess[str] = subprocess.run(
[ [
"nmcli", "nmcli",
"connection", "connection",
@@ -716,10 +718,12 @@ class FrameWidgets(ttk.Frame):
shutil.copy(filepath, f"{AppConfig.TEMP_DIR}/") shutil.copy(filepath, f"{AppConfig.TEMP_DIR}/")
if self.a != "": if self.a != "":
check_call(["nmcli", "connection", "down", self.a]) process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "down", self.a]
)
self.reset_fields() self.reset_fields()
subprocess.check_output( process: CompletedProcess[str] = subprocess.run(
[ [
"nmcli", "nmcli",
"connection", "connection",
@@ -760,7 +764,7 @@ class FrameWidgets(ttk.Frame):
self.color_label() self.color_label()
self.stop() self.stop()
data = self.handle_tunnel_data(self.a) data = self.handle_tunnel_data(self.a)
check_call( process: CompletedProcess[str] = subprocess.run(
[ [
"nmcli", "nmcli",
"con", "con",
@@ -801,7 +805,9 @@ class FrameWidgets(ttk.Frame):
) as file2: ) as file2:
key = Tunnel.con_to_dict(file2) key = Tunnel.con_to_dict(file2)
pre_key = key[3] pre_key = key[3]
check_call(["nmcli", "connection", "delete", select_tl]) process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "delete", select_tl]
)
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: with open(AppConfig.SETTINGS_FILE, "r", encoding="utf-8") as set_f6:
lines6 = set_f6.readlines() lines6 = set_f6.readlines()
@@ -1145,7 +1151,9 @@ class FrameWidgets(ttk.Frame):
""" """
if action == "stop": if action == "stop":
if self.a: if self.a:
check_call(["nmcli", "connection", "down", self.a]) process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "down", self.a]
)
self.update_connection_display() self.update_connection_display()
self.reset_fields() self.reset_fields()
self.start() self.start()
@@ -1153,7 +1161,9 @@ class FrameWidgets(ttk.Frame):
elif action == "start": elif action == "start":
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
check_call(["nmcli", "connection", "up", target_tunnel]) process: CompletedProcess[str] = subprocess.run(
["nmcli", "connection", "up", target_tunnel]
)
self.update_connection_display() self.update_connection_display()
data = self.handle_tunnel_data(self.a) data = self.handle_tunnel_data(self.a)
self.init_and_report(data) self.init_and_report(data)