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

remove USER_FILE usage in ssl_decrypt.py and ssl_encrypt.py; switch to argparse for command-line arguments

Browse Source
This commit is contained in:
Désiré Werner Menrath 2025-05-21 21:29:21 +02:00
parent 4cdcfadbac
commit 5ac37ad9ad
7 changed files with 106 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
common_tools.py
View File

@ -6,7 +6,7 @@ import signal
import base64 import base64
import secrets import secrets
import subprocess import subprocess
from subprocess import CompletedProcess from subprocess import CompletedProcess, run
import re import re
import sys import sys
import tkinter as tk import tkinter as tk
@ -30,36 +30,36 @@ class CryptoUtil:
""" """
@staticmethod @staticmethod
def decrypt() -> None: def decrypt(user) -> None:
""" """
Starts SSL dencrypt Starts SSL dencrypt
""" """
crypted_tunnel = [str(file) for file in AppConfig.CONFIG_DIR.glob("*.dat")] if len([file.stem for file in AppConfig.CONFIG_DIR.glob("*.dat")]) == 0:
if crypted_tunnel == []: pass
return
process: CompletedProcess[str] = subprocess.run(
["pkexec", "/usr/local/bin/ssl_decrypt.py"],
capture_output=True,
text=True,
check=False,
)
# Output from Openssl Error
if process.stderr:
print(process.stderr)
if process.returncode == 0:
print("Files successfully decrypted...")
else: else:
print(f"Error process decrypt: Code {process.returncode}") process: CompletedProcess[str] = run(
["pkexec", "/usr/local/bin/ssl_decrypt.py", "--user", user],
capture_output=True,
text=True,
check=False,
)
# Output from Openssl Error
if process.stderr:
print(process.stderr)
if process.returncode == 0:
print("Files successfully decrypted...")
else:
print(f"Error process decrypt: Code {process.returncode}")
@staticmethod @staticmethod
def encrypt() -> None: def encrypt(user) -> None:
""" """
Starts SSL encryption Starts SSL encryption
""" """
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
["pkexec", "/usr/local/bin/ssl_encrypt.py"], ["pkexec", "/usr/local/bin/ssl_encrypt.py", "--user", user],
capture_output=True, capture_output=True,
text=True, text=True,
check=False, check=False,
@ -80,7 +80,7 @@ class CryptoUtil:
Checks if the private key already exists in the system using an external script. 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). Returns True only if the full key is found exactly (no partial match).
""" """
process: CompletedProcess[bool] = subprocess.run( process: CompletedProcess[bool] = run(
["pkexec", "/usr/local/bin/match_found.py", key], ["pkexec", "/usr/local/bin/match_found.py", key],
capture_output=True, capture_output=True,
text=True, text=True,
@ -224,9 +224,9 @@ class LxTools:
even if the script is running with root privileges. even if the script is running with root privileges.
""" """
try: try:
result = subprocess.run( result = run(
["logname"], ["logname"],
stdout=subprocess.PIPE, stdout=PIPE,
text=True, text=True,
check=True, check=True,
) )

3
match_found.py
View File

@ -42,11 +42,10 @@ def search_string_in_directory(
except Exception: except Exception:
continue # Skip files that cause errors continue # Skip files that cause errors
# Invert the logic: return False if string is found, True otherwise
return result return result
def main() -> bool: def main() -> None:
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description="Script only for use to compare the private key in the Network configurations to avoid errors with the network manager." description="Script only for use to compare the private key in the Network configurations to avoid errors with the network manager."
) )

33
ssl_decrypt.py
View File

@ -1,19 +1,30 @@
#!/usr/bin/python3 #!/usr/bin/python3
""" This Script decrypt Wireguard files for Wirepy users """ """ This Script decrypt Wireguard files for Wirepy users """
import argparse
from pathlib import Path from pathlib import Path
import pwd
import shutil import shutil
from subprocess import CompletedProcess from subprocess import CompletedProcess, run
import subprocess
from wp_app_config import AppConfig from wp_app_config import AppConfig
log_name = AppConfig.USER_FILE.read_text().strip() parser = argparse.ArgumentParser()
parser.add_argument("--user", required=True, help="Username of the target file system")
args = parser.parse_args()
keyfile: Path = Path(f"/home/{log_name}/.config/wire_py/pbwgk.pem") try:
path_of_crypted_tunnel: Path = Path(f"/home/{log_name}/.config/wire_py") # Retrieve UID and GID
user_info = pwd.getpwnam(args.user)
uid = user_info.pw_uid # User ID (e.g., 1000)
gid = user_info.pw_gid # Group ID (e.g., 1000)
except KeyError:
print(f"User '{args.user}' not found.")
exit(1)
keyfile: Path = Path(f"/home/{args.user}/.config/wire_py/pbwgk.pem")
path_of_crypted_tunnel: Path = Path(f"/home/{args.user}/.config/wire_py")
if not keyfile.is_file(): if not keyfile.is_file():
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
[ [
"openssl", "openssl",
"rsa", "rsa",
@ -34,9 +45,9 @@ if not keyfile.is_file():
print("Public key generated successfully.") print("Public key generated successfully.")
else: else:
print(f"Error with the following code... {process.returncode}") print(f"Error with the following code... {process.returncode}")
shutil.chown(keyfile, 1000, 1000) shutil.chown(keyfile, uid, gid)
if AppConfig.PUBLICKEY.exists: if AppConfig.PUBLICKEY.exists():
crypted__tunnel = [str(file) for file in path_of_crypted_tunnel.glob("*.dat")] crypted__tunnel = [str(file) for file in path_of_crypted_tunnel.glob("*.dat")]
@ -44,7 +55,7 @@ if AppConfig.PUBLICKEY.exists:
base_name = Path(tunnel_path).stem base_name = Path(tunnel_path).stem
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
[ [
"openssl", "openssl",
"pkeyutl", "pkeyutl",
@ -60,7 +71,7 @@ if AppConfig.PUBLICKEY.exists:
text=True, text=True,
check=False, check=False,
) )
shutil.chown(f"{AppConfig.TEMP_DIR}/{base_name}.conf", 1000, 1000) shutil.chown(f"{AppConfig.TEMP_DIR}/{base_name}.conf", uid, gid)
print(f"Processing of the file: {tunnel_path}") print(f"Processing of the file: {tunnel_path}")
if process.stdout: if process.stdout:

29
ssl_encrypt.py
View File

@ -1,20 +1,33 @@
#!/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 argparse
from pathlib import Path from pathlib import Path
import pwd
import shutil import shutil
import subprocess from subprocess import CompletedProcess, run
from subprocess import CompletedProcess
from wp_app_config import AppConfig from wp_app_config import AppConfig
log_name = AppConfig.USER_FILE.read_text().strip() parser = argparse.ArgumentParser()
parser.add_argument("--user", required=True, help="Username of the target file system")
args = parser.parse_args()
keyfile: Path = Path(f"/home/{log_name}/.config/wire_py/pbwgk.pem") try:
# Retrieve UID and GID
user_info = pwd.getpwnam(args.user)
uid = user_info.pw_uid # User ID (e.g., 1000)
gid = user_info.pw_gid # Group ID (e.g., 1000)
except KeyError:
print(f"User '{args.user}' not found.")
exit(1)
target: Path = Path(f"/home/{log_name}/.config/wire_py/") keyfile: Path = Path(f"/home/{args.user}/.config/wire_py/pbwgk.pem")
target: Path = Path(f"/home/{args.user}/.config/wire_py/")
if not keyfile.is_file(): if not keyfile.is_file():
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
[ [
"openssl", "openssl",
"rsa", "rsa",
@ -43,7 +56,7 @@ if not keyfile.is_file():
else: else:
print(f"Error generate Publickey: Code: {process.returncode}") print(f"Error generate Publickey: Code: {process.returncode}")
shutil.chown(keyfile, 1000, 1000) shutil.chown(keyfile, uid, gid)
# any() get True when directory is not empty # any() get True when directory is not empty
if AppConfig.TEMP_DIR.exists() and any(AppConfig.TEMP_DIR.iterdir()): if AppConfig.TEMP_DIR.exists() and any(AppConfig.TEMP_DIR.iterdir()):
@ -51,7 +64,7 @@ if AppConfig.TEMP_DIR.exists() and any(AppConfig.TEMP_DIR.iterdir()):
for config_file in clear_files: for config_file in clear_files:
base_name = Path(config_file).stem base_name = Path(config_file).stem
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
[ [
"openssl", "openssl",
"pkeyutl", "pkeyutl",

5
start_wg.py
View File

@ -3,15 +3,14 @@
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
""" """
import subprocess from subprocess import CompletedProcess, run
from subprocess import CompletedProcess
from wp_app_config import AppConfig from wp_app_config import AppConfig
from common_tools import ConfigManager from common_tools import ConfigManager
ConfigManager.init(AppConfig.SETTINGS_FILE) ConfigManager.init(AppConfig.SETTINGS_FILE)
if ConfigManager.get("autostart") != "off": if ConfigManager.get("autostart") != "off":
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
["nmcli", "connection", "up", ConfigManager.get("autostart")], ["nmcli", "connection", "up", ConfigManager.get("autostart")],
capture_output=True, capture_output=True,
text=True, text=True,

49
wirepy.py
View File

@ -10,7 +10,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 CompletedProcess from subprocess import CompletedProcess, run
from tkinter import TclError, filedialog, ttk from tkinter import TclError, filedialog, ttk
from common_tools import ( from common_tools import (
@ -24,10 +24,9 @@ from common_tools import (
) )
from wp_app_config import AppConfig, Msg from wp_app_config import AppConfig, Msg
AppConfig.USER_FILE.write_text(getpass.getuser())
AppConfig.ensure_directories() AppConfig.ensure_directories()
AppConfig.create_default_settings() AppConfig.create_default_settings()
CryptoUtil.decrypt() CryptoUtil.decrypt(getpass.getuser())
class Wirepy(tk.Tk): class Wirepy(tk.Tk):
@ -538,7 +537,13 @@ class FrameWidgets(ttk.Frame):
self.tooltip_label.set(_("Enable Tooltips")) self.tooltip_label.set(_("Enable Tooltips"))
def tooltips_toggle(self): def tooltips_toggle(self):
"""Toggles tooltips on/off and updates the menu label""" """
Toggles the visibility of tooltips (on/off) and updates
the corresponding menu label. Inverts the current tooltip state
(`self.tooltip_state`), saves the new value in the configuration,
and applies the change immediately. Updates the menu entry's label to
reflect the new tooltip status (e.g., "Tooltips: On" or "Tooltips: Off").
"""
# Toggle the boolean state # Toggle the boolean state
new_bool_state = not self.tooltip_state.get() new_bool_state = not self.tooltip_state.get()
# Save the converted value in the configuration # Save the converted value in the configuration
@ -674,7 +679,7 @@ class FrameWidgets(ttk.Frame):
self.tl.update(data_import) self.tl.update(data_import)
if self.a != "": if self.a != "":
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
["nmcli", "connection", "down", self.a], ["nmcli", "connection", "down", self.a],
capture_output=True, capture_output=True,
text=True, text=True,
@ -688,7 +693,7 @@ class FrameWidgets(ttk.Frame):
print(f"Error process decrypt: Code {process.returncode}") print(f"Error process decrypt: Code {process.returncode}")
self.reset_fields() self.reset_fields()
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
[ [
"nmcli", "nmcli",
"connection", "connection",
@ -711,7 +716,7 @@ class FrameWidgets(ttk.Frame):
else: else:
print(f"Error process decrypt: Code {process.returncode}") print(f"Error process decrypt: Code {process.returncode}")
CryptoUtil.encrypt() CryptoUtil.encrypt(getpass.getuser())
LxTools.clean_files(AppConfig.TEMP_DIR, file=None) LxTools.clean_files(AppConfig.TEMP_DIR, file=None)
AppConfig.ensure_directories() AppConfig.ensure_directories()
self.str_var.set("") self.str_var.set("")
@ -740,7 +745,7 @@ class FrameWidgets(ttk.Frame):
self.stop() self.stop()
self.handle_tunnel_data(self.a, self.tl) self.handle_tunnel_data(self.a, self.tl)
self.show_data() self.show_data()
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
["nmcli", "con", "mod", self.a, "connection.autoconnect", "no"], ["nmcli", "con", "mod", self.a, "connection.autoconnect", "no"],
capture_output=True, capture_output=True,
text=True, text=True,
@ -779,7 +784,7 @@ class FrameWidgets(ttk.Frame):
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])
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
["nmcli", "connection", "delete", select_tl], ["nmcli", "connection", "delete", select_tl],
capture_output=True, capture_output=True,
text=True, text=True,
@ -973,7 +978,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
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
[ [
"nmcli", "nmcli",
"connection", "connection",
@ -1036,7 +1041,13 @@ class FrameWidgets(ttk.Frame):
def show_data(self) -> None: def show_data(self) -> None:
""" """
shows data in the label Displays network-related data (address, DNS, endpoint)
in the UI using ttk.Label widgets.
Creates three labels for address, DNS, and endpoint with
specific styling (color, font), positioning them in a
grid layout (`lb_frame` and `lb_frame2`).
Each label is linked to a corresponding text variable
(`self.add`, `self.DNS`, `self.enp`) for dynamic data updates.
""" """
# Address Label # Address Label
self.address = ttk.Label( self.address = ttk.Label(
@ -1059,7 +1070,13 @@ class FrameWidgets(ttk.Frame):
def wg_switch(self, event=None) -> None: def wg_switch(self, event=None) -> None:
""" """
Deals with switching the VPN connection Manages switching between active and inactiveVPN connections.
If no tunnel is selected (`self.a == ""`), it starts a new connection
with the selected tunnel from the listbox (`l_box`).
Otherwise, it stops the current connection and updates
tunnel data using `handle_tunnel_data`.
Handles errors like `IndexError` by displaying appropriate
messages if no items are selected or the listbox is empty.
""" """
try: try:
if self.a == "": if self.a == "":
@ -1102,7 +1119,7 @@ class FrameWidgets(ttk.Frame):
""" """
if action == "stop": if action == "stop":
if self.a: if self.a:
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
["nmcli", "connection", "down", self.a], ["nmcli", "connection", "down", self.a],
capture_output=True, capture_output=True,
text=True, text=True,
@ -1122,7 +1139,7 @@ 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
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
["nmcli", "connection", "up", target_tunnel], ["nmcli", "connection", "up", target_tunnel],
capture_output=True, capture_output=True,
text=True, text=True,
@ -1164,7 +1181,7 @@ class FrameWidgets(ttk.Frame):
if __name__ == "__main__": if __name__ == "__main__":
_ = AppConfig.setup_translations() _ = AppConfig.setup_translations()
LxTools.sigi(AppConfig.TEMP_DIR, AppConfig.USER_FILE) LxTools.sigi(AppConfig.TEMP_DIR)
window = Wirepy() window = Wirepy()
""" """
the hidden files are hidden in Filedialog the hidden files are hidden in Filedialog
@ -1177,5 +1194,5 @@ if __name__ == "__main__":
window.tk.call("set", "::tk::dialog::file::showHiddenVar", "0") window.tk.call("set", "::tk::dialog::file::showHiddenVar", "0")
window.mainloop() window.mainloop()
LxTools.clean_files(AppConfig.TEMP_DIR, AppConfig.USER_FILE) LxTools.clean_files(AppConfig.TEMP_DIR)
sys.exit(0) sys.exit(0)

7
wp_app_config.py
View File

@ -4,7 +4,7 @@
import gettext import gettext
import locale import locale
from pathlib import Path from pathlib import Path
import subprocess from subprocess import CompletedProcess, run
from typing import Dict, Any from typing import Dict, Any
@ -36,7 +36,6 @@ class AppConfig:
BASE_DIR: Path = Path.home() BASE_DIR: Path = Path.home()
CONFIG_DIR: Path = BASE_DIR / ".config/wire_py" CONFIG_DIR: Path = BASE_DIR / ".config/wire_py"
TEMP_DIR: Path = Path("/tmp/tlecdcwg") TEMP_DIR: Path = Path("/tmp/tlecdcwg")
USER_FILE: Path = Path("/tmp/.log_user")
PUBLICKEY: Path = CONFIG_DIR / "pbwgk.pem" PUBLICKEY: Path = CONFIG_DIR / "pbwgk.pem"
# Configuration files # Configuration files
@ -135,14 +134,12 @@ class AppConfig:
if not cls.SYSTEMD_USER_FOLDER.exists(): if not cls.SYSTEMD_USER_FOLDER.exists():
cls.SYSTEMD_USER_FOLDER.mkdir(parents=True, exist_ok=True) cls.SYSTEMD_USER_FOLDER.mkdir(parents=True, exist_ok=True)
from subprocess import CompletedProcess
if not cls.AUTOSTART_SERVICE.is_file(): if not cls.AUTOSTART_SERVICE.is_file():
content = "\n".join([line for line in SYSTEMD_FILE]) content = "\n".join([line for line in SYSTEMD_FILE])
cls.AUTOSTART_SERVICE.write_text(content) cls.AUTOSTART_SERVICE.write_text(content)
process: CompletedProcess[str] = subprocess.run( process: CompletedProcess[str] = run(
["systemctl", "--user", "enable", "wg_start.service"], ["systemctl", "--user", "enable", "wg_start.service"],
capture_output=True, capture_output=True,
text=True, text=True,