Files
VoxVera/voxvera/cli.py
T

876 lines
29 KiB
Python

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
import datetime
from pathlib import Path
from importlib import resources
from importlib.resources.abc import Traversable
from InquirerPy import prompt, inquirer
from rich.console import Console
# package root (contains bundled templates and src/)
ROOT = Path(__file__).resolve().parent
def _template_res(*parts) -> Traversable:
"""Return a Traversable for files under the packaged ``templates`` folder."""
return resources.files(__package__).joinpath("templates", *parts)
def _src_res(*parts) -> Traversable:
"""Return a Traversable for files under the packaged ``src`` folder."""
return resources.files(__package__).joinpath("src", *parts)
def require_cmd(cmd: str):
if shutil.which(cmd) is None:
print(
f"Required command '{cmd}' not found. Please install it.", file=sys.stderr
)
return False
return True
def check_deps():
console = Console()
# External CLI tools still required at runtime
cli_tools = ["onionshare-cli"]
# Python packages used by the build pipeline
py_packages = {
"qrcode": "qrcode",
"PIL (Pillow)": "PIL",
"jsmin": "jsmin",
"htmlmin": "htmlmin",
}
console.rule("Dependency Check")
for t in cli_tools:
if shutil.which(t):
console.print(f"{t}: [green]found[/green]")
else:
console.print(f"{t}: [red]missing[/red]")
for label, module in py_packages.items():
try:
__import__(module)
console.print(f"{label}: [green]found[/green]")
except ImportError:
console.print(f"{label}: [red]missing[/red]")
all_cli = all(shutil.which(t) for t in cli_tools)
all_py = True
for module in py_packages.values():
try:
__import__(module)
except ImportError:
all_py = False
break
if all_cli and all_py:
console.print("[green]All dependencies are installed.[/green]")
else:
console.print(
"[red]Some dependencies are missing. Run: pip install 'voxvera[all]'[/red]"
)
def load_config(path: str) -> dict:
with open(path, "r") as fh:
return json.load(fh)
def save_config(data: dict, path: str):
with open(path, "w") as fh:
json.dump(data, fh, indent=2)
def _open_editor_terminal(initial: str) -> str:
"""Fallback to opening the user's $EDITOR in the terminal."""
import tempfile
editor = os.environ.get("EDITOR", "nano")
fd, path = tempfile.mkstemp(suffix=".txt")
try:
with os.fdopen(fd, "w", encoding="utf-8") as fh:
fh.write(initial or "")
subprocess.call([editor, path])
with open(path, "r", encoding="utf-8") as fh:
return fh.read()
finally:
os.unlink(path)
# 🔧 merged conflicting changes from codex/populate-gui-text-fields-for-editing vs main
def open_editor(initial: str) -> str:
"""Open a simple GUI text editor with pre-populated content if possible.
Existing text is pre-filled in the editor. If tkinter or a display
server is unavailable, falls back to the user's $EDITOR in the terminal.
"""
try:
import tkinter as tk
from tkinter import scrolledtext
root = tk.Tk()
root.title("Edit text")
except Exception:
return _open_editor_terminal(initial)
result = {"text": initial or ""}
text = scrolledtext.ScrolledText(root, width=80, height=20)
text.pack(expand=True, fill="both")
if initial:
text.insert("1.0", initial)
text.focus_set()
def save_and_close():
result["text"] = text.get("1.0", "end-1c")
root.destroy()
save_btn = tk.Button(root, text="Save", command=save_and_close)
save_btn.pack()
root.protocol("WM_DELETE_WINDOW", save_and_close)
root.mainloop()
return result["text"]
def _len_transform(limit: int):
def _t(val: str) -> str:
length = len(val)
if length > limit:
return f"[red]{val} ({length}/{limit})[/red]"
return f"{val} ({length}/{limit})"
return _t
def _len_validator(limit: int):
def _v(val: str):
length = len(val)
if length > limit:
return f"❌ Input too long: {length}/{limit} characters. Please shorten it."
return True
return _v
def _folder_name_validator(val: str):
if len(val) > 63:
return "Folder name must be at most 63 characters"
if not re.fullmatch(r"[a-z0-9-]+", val):
return "Use only lowercase letters, numbers and '-'"
return True
def interactive_update(config_path: str):
data = load_config(config_path)
console = Console()
console.rule("Metadata")
meta_qs = [
{
"type": "input",
"message": "Name",
"name": "name",
"default": data.get("name", ""),
"transformer": _len_transform(60),
"validate": _len_validator(60),
},
{
"type": "input",
"message": "Folder Name",
"name": "folder_name",
"default": data.get("folder_name", ""),
"transformer": _len_transform(63),
"validate": _folder_name_validator,
},
{
"type": "input",
"message": "Title",
"name": "title",
"default": data.get("title", ""),
"transformer": _len_transform(60),
"validate": _len_validator(60),
},
{
"type": "input",
"message": "Subtitle",
"name": "subtitle",
"default": data.get("subtitle", ""),
"transformer": _len_transform(80),
"validate": _len_validator(80),
},
{
"type": "input",
"message": "Headline",
"name": "headline",
"default": data.get("headline", ""),
"transformer": _len_transform(80),
"validate": _len_validator(80),
},
]
data.update(prompt(meta_qs))
console.rule("Body text")
while True:
body = open_editor(data.get("content", ""))
length = len(body)
if length > 1000:
console.print(f"Body length: {length}/1000 exceeds limit", style="red")
if not inquirer.confirm(message="Edit again?", default=True).execute():
break
else:
console.print(f"Body length: {length}/1000", style="green")
break
data["content"] = body
console.rule("Links")
# URL is the main content link (user-configurable, e.g., external resource)
# Tear-off link will be auto-generated from Tor keys when serving
link_qs = [
{
"type": "input",
"message": "URL (content link - external resource)",
"name": "url",
"default": data.get("url", ""),
"transformer": _len_transform(200),
"validate": _len_validator(200),
},
{
"type": "input",
"message": "URL message",
"name": "url_message",
"default": data.get("url_message", ""),
"transformer": _len_transform(120),
"validate": _len_validator(120),
},
{
"type": "input",
"message": "Binary message",
"name": "binary_message",
"default": data.get("binary_message", ""),
"transformer": _len_transform(120),
"validate": _len_validator(120),
},
]
data.update(prompt(link_qs))
save_config(data, config_path)
def copy_template(name: str) -> str:
"""Copy a template directory into dist/ with a datestamped folder."""
date = datetime.date.today().strftime("%Y%m%d")
with resources.as_file(_template_res(name)) as src:
if not src.is_dir():
print(f"Template {name} not found", file=sys.stderr)
sys.exit(1)
dest = ROOT / "dist" / f"{name}-{date}"
os.makedirs(ROOT / "dist", exist_ok=True)
shutil.copytree(src, dest, dirs_exist_ok=True)
print(f"Template copied to {dest}")
return str(dest)
def build_assets(
config_path: str, download_path: str | None = None
):
from voxvera.build import generate_qr
with resources.as_file(_src_res()) as src_dir:
config_path = Path(config_path)
# generate QR codes (pure Python)
generate_qr(config_path, src_dir)
data = load_config(config_path)
with open(src_dir / "index-master.html", "r") as fh:
html = fh.read()
# Statically replace {{placeholders}} with config values for Tor JS-disabled compatibility
for key, value in data.items():
html = html.replace(f"{{{{{key}}}}}", str(value))
folder_name = data["folder_name"]
dest = ROOT / "host" / folder_name
os.makedirs(dest, exist_ok=True)
with open(dest / "index.html", "w") as fh:
fh.write(html)
if download_path:
os.makedirs(dest / "download", exist_ok=True)
shutil.copy(download_path, dest / "download" / "download.zip")
# Only copy config if it's not already in the destination
if Path(config_path) != dest / "config.json":
shutil.copy(config_path, dest / "config.json")
# Copy QR codes only if they exist (may not be generated if URLs not set)
for qr_file in ["qrcode-content.png", "qrcode-tear-offs.png"]:
qr_src = Path(src_dir) / qr_file
if qr_src.exists():
shutil.copy(qr_src, dest / qr_file)
print(f"Flyer files created under {dest}")
def get_tor_ports():
"""Auto-detect Tor SOCKS and Control ports from common defaults."""
# Common default ports for Tor
socks_defaults = ["9050", "9150"]
ctl_defaults = ["9051", "9151"]
socks_port = os.getenv("TOR_SOCKS_PORT")
ctl_port = os.getenv("TOR_CONTROL_PORT")
# If env vars not set, try to auto-detect from running Tor
if not socks_port or not ctl_port:
import socket
for port in socks_defaults:
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(0.5)
result = sock.connect_ex(("127.0.0.1", int(port)))
sock.close()
if result == 0:
socks_port = port
break
except Exception:
pass
for port in ctl_defaults:
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(0.5)
result = sock.connect_ex(("127.0.0.1", int(port)))
sock.close()
if result == 0:
ctl_port = port
break
except Exception:
pass
# Fall back to defaults if nothing detected
if not socks_port:
socks_port = "9050"
if not ctl_port:
ctl_port = "9051"
return socks_port, ctl_port
def serve(config_path: str) -> str | None:
if not require_cmd("onionshare-cli"):
sys.exit(1)
socks, ctl = get_tor_ports()
folder_name = load_config(config_path)["folder_name"]
dir_path = ROOT / "host" / folder_name
if not dir_path.is_dir():
print(f"Directory {dir_path} not found", file=sys.stderr)
sys.exit(1)
logfile = dir_path / "onionshare.log"
# Set environment variables for OnionShare to use the detected Tor ports
env = os.environ.copy()
env["TOR_SOCKS_PORT"] = socks
env["TOR_CONTROL_PORT"] = ctl
cmd = [
"onionshare-cli",
"--website",
"--public",
"--persistent",
str(dir_path / ".onionshare-session"),
"--connect-timeout",
"300",
"--disable_csp",
"-v",
str(dir_path),
]
log_fh = open(logfile, "w")
proc = subprocess.Popen(cmd, stdout=log_fh, stderr=subprocess.STDOUT, env=env)
pid_file = dir_path / "server.pid"
with open(pid_file, "w") as f:
f.write(str(proc.pid))
try:
import time
import re as _re
onion_url = None
console = Console()
with console.status(f"[bold green][{folder_name}] Connecting to Tor relays and setting up hidden service...") as status:
last_progress = None
while onion_url is None:
time.sleep(1)
if proc.poll() is not None:
status.stop()
console.print(f"[bold red]OnionShare for {folder_name} exited unexpectedly.[/bold red]")
console.print(f"[yellow]Common causes: Network blocked, inaccurate system clock, or OnionShare already running.[/yellow]")
console.print(f"[dim]See {logfile} for full logs.[/dim]")
with open(logfile) as fh:
lines = fh.readlines()
# Print the last 20 lines of the log to help debug
for line in lines[-20:]:
sys.stderr.write(line)
pid_file.unlink(missing_ok=True)
return None
if os.path.exists(logfile):
with open(logfile) as fh:
content = fh.read()
# Look for Tor bootstrap progress in logs to give feedback
m_progress = _re.findall(r"Bootstrapped (\d+)%", content)
if m_progress:
current_progress = m_progress[-1]
if current_progress != last_progress:
status.update(f"[bold green][{folder_name}] Tor Bootstrapping: {current_progress}% ...")
last_progress = current_progress
m = _re.search(r"https?://[a-z0-9]+\.onion", content)
if m:
onion_url = m.group(0)
# update config with the onion URL in the host directory
config_file = dir_path / "config.json"
data = load_config(config_file)
data["tear_off_link"] = onion_url
save_config(data, config_file)
# regenerate assets with the new URL using the host-local config
build_assets(str(config_file))
print(f"[{folder_name}] Onion URL: {onion_url}")
print(f"[{folder_name}] OnionShare running (PID {proc.pid}).")
return onion_url
except KeyboardInterrupt:
return None
finally:
log_fh.close()
def _clear_host_dir(dir_path: Path):
"""Remove host directory contents but preserve the OnionShare session key."""
session = dir_path / ".onionshare-session"
saved = session.read_bytes() if session.exists() else None
shutil.rmtree(dir_path, ignore_errors=True)
if saved is not None:
os.makedirs(dir_path, exist_ok=True)
session.write_bytes(saved)
import zipfile
import tempfile
def get_export_dir() -> Path:
"""Get the default export directory (~/voxvera-exports)."""
export_dir = Path.home() / "voxvera-exports"
os.makedirs(export_dir, exist_ok=True)
return export_dir
def export_site(folder_name: str, output_path: str = None):
"""Export a site folder and its keys into a zip archive."""
source_dir = ROOT / "host" / folder_name
if not source_dir.exists():
print(f"Error: Site folder '{folder_name}' not found in {ROOT / 'host'}")
return
if not output_path:
export_dir = get_export_dir()
output_path = export_dir / f"{folder_name}_export.zip"
else:
output_path = Path(output_path)
with zipfile.ZipFile(output_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for root, dirs, files in os.walk(source_dir):
for file in files:
file_path = Path(root) / file
arcname = file_path.relative_to(source_dir)
zipf.write(file_path, arcname)
print(f"Exported '{folder_name}' to {output_path}")
def export_all_sites():
"""Export all available sites in the host directory."""
servers = get_servers()
if not servers:
print("No sites found to export.")
return
export_dir = get_export_dir()
print(f"Exporting {len(servers)} sites to {export_dir}...")
for server in servers:
export_site(server)
print("Export all complete.")
def import_site(zip_path: str):
"""Import a site folder and its keys from a zip archive."""
zip_path = Path(zip_path)
if not zip_path.exists():
print(f"Error: Zip file '{zip_path}' not found.")
return
with tempfile.TemporaryDirectory() as tmpdir:
with zipfile.ZipFile(zip_path, 'r') as zipf:
zipf.extractall(tmpdir)
tmp_path = Path(tmpdir)
config_file = tmp_path / "config.json"
if not config_file.exists():
print("Error: Invalid export file. Missing 'config.json'.")
return
data = load_config(config_file)
folder_name = data.get("folder_name")
if not folder_name:
print("Error: Invalid config.json. Missing 'folder_name'.")
return
dest_dir = ROOT / "host" / folder_name
if dest_dir.exists():
confirm = inquirer.confirm(message=f"Site '{folder_name}' already exists. Overwrite?").execute()
if not confirm:
print("Import cancelled.")
return
shutil.rmtree(dest_dir)
shutil.copytree(tmp_path, dest_dir)
print(f"Successfully imported '{folder_name}' to {dest_dir}")
def import_multiple_sites(source_dir: str = None):
"""Import all zip files from a directory (defaults to ~/voxvera-exports)."""
import glob
if not source_dir:
source_dir = get_export_dir()
else:
source_dir = Path(source_dir)
if not source_dir.exists():
print(f"Error: Directory '{source_dir}' not found.")
return
zip_files = sorted(glob.glob(str(source_dir / "*.zip")))
if not zip_files:
print(f"No zip files found in {source_dir}")
return
print(f"Found {len(zip_files)} site(s) to import from {source_dir}...")
for zip_path in zip_files:
print(f"\nImporting {Path(zip_path).name}...")
import_site(zip_path)
print("\nImport multiple complete.")
def batch_import_configs():
import glob
files = sorted(glob.glob(str(ROOT / "imports" / "*.json")))
if not files:
print("No JSON files found in imports")
return
for json_file in files:
print(f"Processing {json_file}")
dest_config = ROOT / "src" / "config.json"
shutil.copy(json_file, dest_config)
folder_name = load_config(json_file)["folder_name"]
_clear_host_dir(ROOT / "host" / folder_name)
build_assets(dest_config)
def get_servers() -> list[str]:
host_dir = ROOT / "host"
if not host_dir.exists():
return []
servers = []
for d in host_dir.iterdir():
if d.is_dir() and (d / "config.json").exists():
servers.append(d.name)
return sorted(servers)
def is_server_running(folder_name: str) -> bool:
pid_file = ROOT / "host" / folder_name / "server.pid"
if not pid_file.exists():
return False
try:
with open(pid_file, "r") as f:
pid = int(f.read().strip())
except ValueError:
pid_file.unlink(missing_ok=True)
return False
import platform
import subprocess
if platform.system() == "Windows":
try:
output = subprocess.check_output(f'tasklist /FI "PID eq {pid}" /NH', shell=True, stderr=subprocess.STDOUT).decode()
if "No tasks are running" in output:
pid_file.unlink(missing_ok=True)
return False
return True
except subprocess.CalledProcessError:
pid_file.unlink(missing_ok=True)
return False
else:
try:
os.kill(pid, 0)
except OSError:
pid_file.unlink(missing_ok=True)
return False
try:
output = subprocess.check_output(["ps", "-p", str(pid), "-o", "command="], stderr=subprocess.STDOUT).decode().lower()
if "onionshare" in output or "python" in output or "tor" in output:
return True
return True
except Exception:
return True
def stop_server(folder_name: str):
pid_file = ROOT / "host" / folder_name / "server.pid"
if not pid_file.exists():
print(f"Server {folder_name} is not running.")
return
try:
with open(pid_file, "r") as f:
pid = int(f.read().strip())
import platform, signal
if platform.system() == "Windows":
subprocess.run(["taskkill", "/F", "/PID", str(pid)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
else:
os.kill(pid, signal.SIGTERM)
print(f"Stopped {folder_name} (PID {pid})")
except Exception as e:
print(f"Error stopping {folder_name}: {e}")
finally:
pid_file.unlink(missing_ok=True)
def delete_server(folder_name: str):
if is_server_running(folder_name):
stop_server(folder_name)
dir_path = ROOT / "host" / folder_name
if dir_path.exists():
shutil.rmtree(dir_path, ignore_errors=True)
print(f"Deleted server {folder_name}")
def manage_servers():
from InquirerPy import inquirer
from InquirerPy.base.control import Choice
console = Console()
while True:
servers = get_servers()
if not servers:
console.print("[red]No servers found. Use 'voxvera build' or 'voxvera quickstart' to create one.[/red]")
return
choices = []
for s in servers:
running = is_server_running(s)
status_text = "[ON] " if running else "[OFF]"
label = f"{status_text} {s}"
if running:
try:
site_data = load_config(ROOT / "host" / s / "config.json")
url = site_data.get("tear_off_link")
if url:
label += f" ({url})"
except Exception:
pass
choices.append(Choice(s, label))
choices.insert(0, Choice("create_new", "--- Create New Site/Flyer ---"))
choices.append(Choice("start_all", "--- Start All Sites ---"))
choices.append(Choice("stop_all", "--- Stop All Sites ---"))
choices.append(Choice("export_all", "--- Export All Sites ---"))
choices.append(Choice("import_multiple", "--- Import Multiple Sites ---"))
choices.append(Choice(None, "Exit"))
selected = inquirer.select(
message="Select a server to manage (or Exit):",
choices=choices
).execute()
if selected is None:
break
if selected == "create_new":
config_path = ROOT / "src" / "config.json"
interactive_update(str(config_path))
build_assets(str(config_path))
try:
serve(str(config_path))
except SystemExit:
pass
continue
if selected == "start_all":
results = []
for s in servers:
if not is_server_running(s):
config_path = ROOT / "host" / s / "config.json"
try:
url = serve(str(config_path))
if url:
results.append((s, url))
except Exception as e:
console.print(f"[red]Error starting {s}: {e}[/red]")
if results:
console.rule("Startup Summary")
for s, url in results:
console.print(f"[green]✔ {s}: [bold]{url}[/bold][/green]")
inquirer.confirm(message="Press Enter to continue", default=True).execute()
continue
if selected == "stop_all":
for s in servers:
if is_server_running(s):
stop_server(s)
continue
if selected == "export_all":
export_all_sites()
continue
if selected == "import_multiple":
import_multiple_sites()
continue
running = is_server_running(selected)
action_choices = [
Choice("toggle", "Stop" if running else "Start"),
Choice("export", "Export to Zip"),
Choice("delete", "Delete"),
Choice(None, "Back")
]
action = inquirer.select(
message=f"Manage {selected}:",
choices=action_choices
).execute()
if action == "toggle":
if running:
stop_server(selected)
else:
config_path = ROOT / "host" / selected / "config.json"
try:
serve(str(config_path))
except SystemExit:
pass
elif action == "export":
export_site(selected)
elif action == "delete":
confirm = inquirer.confirm(message=f"Are you sure you want to delete {selected}?").execute()
if confirm:
delete_server(selected)
def main(argv=None):
parser = argparse.ArgumentParser(prog="voxvera")
parser.add_argument(
"--config",
default=str(ROOT / "src" / "config.json"),
help="Path to config.json",
)
sub = parser.add_subparsers(dest="command")
p_init = sub.add_parser(
"init", help="Update configuration interactively"
)
p_init.add_argument("--template", help="Copy a template into dist/")
p_init.add_argument("--non-interactive", action="store_true")
p_build = sub.add_parser("build", help="Build flyer assets from config")
p_build.add_argument("--download")
p_export = sub.add_parser("export", help="Export a site and its keys into a zip archive")
p_export.add_argument("folder_name", help="Name of the site folder in host/")
p_export.add_argument("--output", help="Custom output path for the zip")
sub.add_parser("export-all", help="Export all sites and their keys to ~/voxvera-exports")
p_import = sub.add_parser("import", help="Import a site and its keys from a zip archive")
p_import.add_argument("zip_path", help="Path to the exported zip file")
p_import_multiple = sub.add_parser("import-multiple", help="Import multiple sites and their keys from a directory")
p_import_multiple.add_argument("--dir", help="Directory containing export zips (defaults to ~/voxvera-exports)")
sub.add_parser("batch-import", help="Batch import JSON files from imports/")
sub.add_parser("serve", help="Serve flyer over OnionShare using config")
p_quickstart = sub.add_parser(
"quickstart", help="Init, build and serve in sequence"
)
p_quickstart.add_argument(
"--non-interactive",
action="store_true",
help="Skip interactive prompts and use existing config",
)
sub.add_parser("check", help="Check for required external dependencies")
sub.add_parser("manage", help="Manage VoxVera servers interactively")
args = parser.parse_args(argv)
config_path = Path(args.config).resolve()
if args.command == "init":
if args.template:
copy_template(args.template)
return
elif not args.non_interactive:
interactive_update(config_path)
build_assets(config_path)
serve(config_path)
elif args.command == "build":
build_assets(config_path, download_path=args.download)
elif args.command == "export":
export_site(args.folder_name, args.output)
elif args.command == "export-all":
export_all_sites()
elif args.command == "import":
import_site(args.zip_path)
elif args.command == "import-multiple":
import_multiple_sites(args.dir)
elif args.command == "serve":
serve(config_path)
elif args.command == "batch-import":
batch_import_configs()
elif args.command == "manage":
manage_servers()
elif args.command == "quickstart":
if not args.non_interactive:
if not sys.stdin.isatty():
print(
"Error: quickstart requires an interactive terminal. "
"Use --non-interactive or run 'voxvera init' first.",
file=sys.stderr,
)
sys.exit(1)
interactive_update(config_path)
build_assets(config_path)
serve(config_path)
elif args.command == "check":
check_deps()
else:
parser.print_help()
if __name__ == "__main__":
main()