From fb0158d1cd7391d1b29f77d9826ca98170f26567 Mon Sep 17 00:00:00 2001 From: punix Date: Sun, 11 May 2025 22:00:28 +0200 Subject: [PATCH] replace all check_call with subprocess.run --- common_tools.py | 80 ++++++------------------------------- ssl_encrypt.py | 102 ++++++++++++++++++++++++----------------------- start_wg.py | 14 ++++++- wg_start.service | 1 + wirepy.py | 30 +++++++++----- 5 files changed, 98 insertions(+), 129 deletions(-) diff --git a/common_tools.py b/common_tools.py index 331384f..224df96 100755 --- a/common_tools.py +++ b/common_tools.py @@ -10,7 +10,7 @@ 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 subprocess import CompletedProcess from tkinter import ttk, Toplevel from wp_app_config import AppConfig, Msg import requests @@ -26,63 +26,6 @@ class Create: 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 def decrypt() -> str: """ @@ -96,17 +39,17 @@ class Create: ) # Output from Openssl - if process.stdout: - print(process.stdout) + # if process.stdout: + # print(process.stdout) # Output from Openssl Error if process.stderr: print(process.stderr) if process.returncode == 0: - print("Datei entschlüsseln wurde erfolgreich entschlossen.") + print("Files successfully decrypted...") else: - print(f"Fehler bei der Verarbeitung von Dateien: Code {process.returncode}") + print(f"Error process decrypt: Code {process.returncode}") @staticmethod def encrypt() -> str: @@ -115,15 +58,18 @@ class Create: """ process: CompletedProcess[str] = subprocess.run( ["pkexec", "/usr/local/bin/ssl_encrypt.py"], - stdout=subprocess.PIPE, + capture_output=True, text=True, - check=True, + check=False, ) - print(process.stdout) + # Output from Openssl Error + if process.stderr: + print(process.stderr) + if process.returncode == 0: - print("All Files successfully encrypted...") + print("Files successfully encrypted...") else: - print(f"Error with the following code... {process.returncode}") + print(f"Error process encrypt: Code {process.returncode}") class LxTools(tk.Tk): diff --git a/ssl_encrypt.py b/ssl_encrypt.py index 306928d..a92b35c 100755 --- a/ssl_encrypt.py +++ b/ssl_encrypt.py @@ -1,12 +1,11 @@ #!/usr/bin/python3 """ This Script encrypt Wireguardfiles for Wirepy users for more Security """ -import os from pathlib import Path import shutil -from subprocess import check_call +import subprocess +from subprocess import CompletedProcess from wp_app_config import AppConfig -from common_tools import LxTools 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(): - check_call( + process: CompletedProcess[str] = subprocess.run( [ "openssl", "rsa", @@ -25,56 +24,59 @@ if not keyfile.is_file(): "-outform", "PEM", "-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) - if AppConfig.TEMP_DIR.exists(): - tl = LxTools.get_file_name(AppConfig.TEMP_DIR) - CPTH: str = f"{keyfile}" - CRYPTFILES: str = CPTH[:-9] + if AppConfig.TEMP_DIR.exists() and not any(AppConfig.TEMP_DIR.iterdir()): + clear_files = [str(file) for file in path_of_crypted_tunnel.glob()] - if keyfile.exists() and len(tl) != 0: - for tunnels in tl: - sourcetl: str = f"{AppConfig.TEMP_DIR}/{tunnels}" - tlname: str = f"{CRYPTFILES}{tunnels[:-5]}.dat" - check_call( - [ - "openssl", - "pkeyutl", - "-encrypt", - "-inkey", - keyfile, - "-pubin", - "-in", - sourcetl, - "-out", - tlname, - ] - ) + for config_file in clear_files: + base_name = Path(config_file).stem + process: CompletedProcess[str] = subprocess.run( + [ + "openssl", + "pkeyutl", + "-encrypt", + "-inkey", + keyfile, + "-pubin", + "-in", + config_file, + "-out", + f"{AppConfig.CONFIG_DIR}/{base_name}.dat", + ], + capture_output=True, + text=True, + check=False, + ) -else: + print(f"Processing of the file: {config_file}") - if AppConfig.TEMP_DIR.exists(): - tl: list[str] = os.listdir(f"{AppConfig.TEMP_DIR}") - CPTH: str = f"{keyfile}" - CRYPTFILES: str = CPTH[:-9] + if process.stdout: + print(process.stdout) - if keyfile.exists() and len(tl) != 0: - for tunnels in tl: - sourcetl: str = f"{AppConfig.TEMP_DIR}/{tunnels}" - tlname: str = f"{CRYPTFILES}{tunnels[:-5]}.dat" - check_call( - [ - "openssl", - "pkeyutl", - "-encrypt", - "-inkey", - keyfile, - "-pubin", - "-in", - sourcetl, - "-out", - tlname, - ] - ) + # Output from Openssl Error + if process.stderr: + print("(Error):", process.stderr) + + if process.returncode == 0: + print(f"File {base_name}.dat successfully encrypted.") + else: + print(f"Error by {config_file}: Code: {process.returncode}") diff --git a/start_wg.py b/start_wg.py index 0262a49..20ca206 100755 --- a/start_wg.py +++ b/start_wg.py @@ -4,13 +4,23 @@ """ 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") a_con = Path(path_to_file).read_text(encoding="utf-8").splitlines(keepends=True) a_con = a_con[7].strip() 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: pass diff --git a/wg_start.service b/wg_start.service index 0352b4e..5d41844 100644 --- a/wg_start.service +++ b/wg_start.service @@ -6,5 +6,6 @@ After=network-online.target Type=oneshot ExecStartPre=/bin/sleep 5 ExecStart=/usr/local/bin/start_wg.py + [Install] WantedBy=default.target diff --git a/wirepy.py b/wirepy.py index 457b2df..1e288de 100755 --- a/wirepy.py +++ b/wirepy.py @@ -11,7 +11,7 @@ import sys import tkinter as tk import webbrowser from pathlib import Path -from subprocess import check_call +from subprocess import CompletedProcess from tkinter import TclError, filedialog, ttk from common_tools import ( @@ -645,7 +645,7 @@ class FrameWidgets(ttk.Frame): def import_sl(self) -> None: """validity check of wireguard config files""" - Create.dir_and_files() + AppConfig.ensure_directories() try: filepath = filedialog.askopenfilename( initialdir=f"{Path.home()}", @@ -696,10 +696,12 @@ class FrameWidgets(ttk.Frame): new_conf = f"{AppConfig.TEMP_DIR}/{path_split}" if self.a != "": - check_call(["nmcli", "connection", "down", self.a]) + process: CompletedProcess[str] = subprocess.run( + ["nmcli", "connection", "down", self.a] + ) self.reset_fields() - subprocess.check_output( + process: CompletedProcess[str] = subprocess.run( [ "nmcli", "connection", @@ -716,10 +718,12 @@ class FrameWidgets(ttk.Frame): shutil.copy(filepath, f"{AppConfig.TEMP_DIR}/") if self.a != "": - check_call(["nmcli", "connection", "down", self.a]) + process: CompletedProcess[str] = subprocess.run( + ["nmcli", "connection", "down", self.a] + ) self.reset_fields() - subprocess.check_output( + process: CompletedProcess[str] = subprocess.run( [ "nmcli", "connection", @@ -760,7 +764,7 @@ class FrameWidgets(ttk.Frame): self.color_label() self.stop() data = self.handle_tunnel_data(self.a) - check_call( + process: CompletedProcess[str] = subprocess.run( [ "nmcli", "con", @@ -801,7 +805,9 @@ class FrameWidgets(ttk.Frame): ) as file2: key = Tunnel.con_to_dict(file2) 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]) with open(AppConfig.SETTINGS_FILE, "r", encoding="utf-8") as set_f6: lines6 = set_f6.readlines() @@ -1145,7 +1151,9 @@ class FrameWidgets(ttk.Frame): """ if action == "stop": 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.reset_fields() self.start() @@ -1153,7 +1161,9 @@ class FrameWidgets(ttk.Frame): elif action == "start": if 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() data = self.handle_tunnel_data(self.a) self.init_and_report(data)