Brian Krabach
514ed5dc77
python-multipart is required by FastAPI for form POST handling (login). Was missing from pyproject.toml — crash on macOS first run. launchd plist now uses the muxplex entry point script directly instead of python -m muxplex, so macOS shows 'muxplex' in Activity Monitor, launchctl list, and login items instead of 'python3'.
374 lines
12 KiB
Python
374 lines
12 KiB
Python
"""muxplex CLI — web-based tmux session dashboard."""
|
|
|
|
import argparse
|
|
import os
|
|
import platform
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import secrets as _secrets
|
|
|
|
from muxplex.auth import (
|
|
get_password_path,
|
|
get_secret_path,
|
|
load_password,
|
|
pam_available,
|
|
)
|
|
|
|
# Module-level path constants (overridable in tests via monkeypatch)
|
|
_system_service_path = Path("/etc/systemd/system/muxplex.service")
|
|
|
|
|
|
def reset_secret() -> None:
|
|
"""Regenerate the signing secret and warn that all sessions are now invalid."""
|
|
path = get_secret_path()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
secret = _secrets.token_urlsafe(32)
|
|
path.write_text(secret + "\n")
|
|
path.chmod(0o600)
|
|
print(f"Secret written to {path}")
|
|
print("Warning: all active sessions are now invalid.")
|
|
|
|
|
|
def show_password() -> None:
|
|
"""Print the current muxplex password or indicate PAM mode."""
|
|
auth_mode = os.environ.get("MUXPLEX_AUTH", "").lower()
|
|
if auth_mode != "password" and pam_available():
|
|
print("Auth mode: PAM — no password file used")
|
|
return
|
|
pw = load_password()
|
|
if pw:
|
|
print(f"Password: {pw}")
|
|
else:
|
|
print("No password file found. Start muxplex to auto-generate one.")
|
|
|
|
|
|
def serve(
|
|
host: str = "127.0.0.1",
|
|
port: int = 8088,
|
|
auth: str = "pam",
|
|
session_ttl: int = 604800,
|
|
) -> None:
|
|
"""Start the muxplex server."""
|
|
import uvicorn # noqa: PLC0415
|
|
|
|
os.environ.setdefault("MUXPLEX_PORT", str(port))
|
|
os.environ.setdefault("MUXPLEX_AUTH", auth)
|
|
os.environ.setdefault("MUXPLEX_SESSION_TTL", str(session_ttl))
|
|
|
|
from muxplex.main import app # noqa: PLC0415
|
|
|
|
print(f" muxplex → http://{host}:{port}")
|
|
uvicorn.run(app, host=host, port=port, log_level="warning")
|
|
|
|
|
|
def doctor() -> None:
|
|
"""Run diagnostic checks and report system status."""
|
|
ok_mark = "\033[32m✓\033[0m" # green check
|
|
fail_mark = "\033[31m✗\033[0m" # red x
|
|
warn_mark = "\033[33m!\033[0m" # yellow warning
|
|
|
|
print("\nmuxplex doctor\n")
|
|
|
|
# Python version
|
|
py_version = platform.python_version()
|
|
py_ok = tuple(int(x) for x in py_version.split(".")[:2]) >= (3, 11)
|
|
print(
|
|
f" {ok_mark if py_ok else fail_mark} Python {py_version}"
|
|
+ ("" if py_ok else " (3.11+ required)")
|
|
)
|
|
|
|
# tmux
|
|
tmux_path = shutil.which("tmux")
|
|
if tmux_path:
|
|
try:
|
|
result = subprocess.run(
|
|
["tmux", "-V"], capture_output=True, text=True, timeout=5
|
|
)
|
|
tmux_version = result.stdout.strip()
|
|
print(f" {ok_mark} {tmux_version}")
|
|
except Exception:
|
|
print(f" {ok_mark} tmux (version unknown)")
|
|
else:
|
|
print(f" {fail_mark} tmux — not found")
|
|
if sys.platform == "darwin":
|
|
print(" Install: brew install tmux")
|
|
else:
|
|
print(" Install: sudo apt install tmux")
|
|
|
|
# ttyd
|
|
ttyd_path = shutil.which("ttyd")
|
|
if ttyd_path:
|
|
try:
|
|
result = subprocess.run(
|
|
["ttyd", "--version"], capture_output=True, text=True, timeout=5
|
|
)
|
|
ttyd_version = result.stdout.strip() or result.stderr.strip()
|
|
print(f" {ok_mark} ttyd {ttyd_version}")
|
|
except Exception:
|
|
print(f" {ok_mark} ttyd (version unknown)")
|
|
else:
|
|
print(f" {fail_mark} ttyd — not found")
|
|
if sys.platform == "darwin":
|
|
print(" Install: brew install ttyd")
|
|
else:
|
|
print(" Install: sudo apt install ttyd")
|
|
|
|
# muxplex version
|
|
try:
|
|
from importlib.metadata import version as pkg_version # noqa: PLC0415
|
|
|
|
muxplex_version = pkg_version("muxplex")
|
|
except Exception:
|
|
muxplex_version = "dev"
|
|
print(f" {ok_mark} muxplex {muxplex_version}")
|
|
|
|
# Settings file
|
|
from muxplex.settings import SETTINGS_PATH # noqa: PLC0415
|
|
|
|
if SETTINGS_PATH.exists():
|
|
print(f" {ok_mark} Settings: {SETTINGS_PATH}")
|
|
else:
|
|
print(
|
|
f" {warn_mark} Settings: {SETTINGS_PATH} (not yet created — will use defaults)"
|
|
)
|
|
|
|
# Auth status
|
|
pw_path = get_password_path()
|
|
if pam_available():
|
|
import pwd # noqa: PLC0415
|
|
|
|
username = pwd.getpwuid(os.getuid()).pw_name
|
|
print(f" {ok_mark} Auth: PAM available (user: {username})")
|
|
elif pw_path.exists():
|
|
print(f" {ok_mark} Auth: password file ({pw_path})")
|
|
elif os.environ.get("MUXPLEX_PASSWORD"):
|
|
print(f" {ok_mark} Auth: password (env var)")
|
|
else:
|
|
print(f" {warn_mark} Auth: no PAM, no password — will auto-generate on serve")
|
|
|
|
# tmux sessions (if tmux is available)
|
|
if tmux_path:
|
|
try:
|
|
result = subprocess.run(
|
|
["tmux", "list-sessions", "-F", "#{session_name}"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5,
|
|
)
|
|
if result.returncode == 0:
|
|
sessions = [s for s in result.stdout.strip().split("\n") if s]
|
|
print(f" {ok_mark} tmux sessions: {len(sessions)} active")
|
|
else:
|
|
print(f" {warn_mark} tmux server not running (no sessions)")
|
|
except Exception:
|
|
print(f" {warn_mark} tmux server not running")
|
|
|
|
# Platform + service status
|
|
print(f" {ok_mark} Platform: {sys.platform} ({platform.machine()})")
|
|
if sys.platform == "darwin":
|
|
plist = Path.home() / "Library" / "LaunchAgents" / "com.muxplex.plist"
|
|
if plist.exists():
|
|
print(f" {ok_mark} Service: launchd agent installed ({plist})")
|
|
else:
|
|
print(
|
|
f" {warn_mark} Service: not installed (run: muxplex install-service)"
|
|
)
|
|
else:
|
|
systemd_user = Path.home() / ".config" / "systemd" / "user" / "muxplex.service"
|
|
if systemd_user.exists():
|
|
print(f" {ok_mark} Service: systemd user unit installed ({systemd_user})")
|
|
elif _system_service_path.exists():
|
|
print(
|
|
f" {ok_mark} Service: systemd system unit installed ({_system_service_path})"
|
|
)
|
|
else:
|
|
print(
|
|
f" {warn_mark} Service: not installed (run: muxplex install-service)"
|
|
)
|
|
|
|
print() # trailing newline
|
|
|
|
|
|
def _check_dependencies() -> None:
|
|
"""Verify required external programs are installed.
|
|
|
|
Checks for tmux and ttyd. Prints a helpful error message and exits with
|
|
code 1 if any are missing.
|
|
"""
|
|
missing = []
|
|
if shutil.which("tmux") is None:
|
|
missing.append(("tmux", "sudo apt install tmux / brew install tmux"))
|
|
if shutil.which("ttyd") is None:
|
|
missing.append(("ttyd", "sudo apt install ttyd / brew install ttyd"))
|
|
|
|
if missing:
|
|
print("\n ERROR: Required dependencies not found:\n", file=sys.stderr)
|
|
for name, install_hint in missing:
|
|
print(f" {name}: {install_hint}", file=sys.stderr)
|
|
print(
|
|
"\n For details: https://github.com/bkrabach/muxplex#prerequisites\n",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
|
|
def _install_launchd(executable: str) -> None:
|
|
"""Install a macOS launchd agent plist to ~/Library/LaunchAgents/."""
|
|
import shutil
|
|
|
|
# Prefer the entry point script ('muxplex') so macOS shows the correct
|
|
# process name in Activity Monitor, launchctl list, and login items.
|
|
# Fall back to 'python -m muxplex' if the entry point isn't on PATH.
|
|
muxplex_bin = shutil.which("muxplex")
|
|
if muxplex_bin:
|
|
program_args = f""" <array>
|
|
<string>{muxplex_bin}</string>
|
|
</array>"""
|
|
else:
|
|
program_args = f""" <array>
|
|
<string>{executable}</string>
|
|
<string>-m</string>
|
|
<string>muxplex</string>
|
|
</array>"""
|
|
|
|
label = "com.muxplex"
|
|
plist = f"""<?xml version="1.0" encoding="UTF-8"?>
|
|
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
|
|
<plist version="1.0">
|
|
<dict>
|
|
<key>Label</key>
|
|
<string>{label}</string>
|
|
<key>ProgramArguments</key>
|
|
{program_args}
|
|
<key>RunAtLoad</key>
|
|
<true/>
|
|
<key>KeepAlive</key>
|
|
<true/>
|
|
<key>StandardOutPath</key>
|
|
<string>/tmp/muxplex.log</string>
|
|
<key>StandardErrorPath</key>
|
|
<string>/tmp/muxplex.err</string>
|
|
</dict>
|
|
</plist>
|
|
"""
|
|
path = Path.home() / "Library" / "LaunchAgents" / f"{label}.plist"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(plist)
|
|
print(f"Launch agent written to {path}")
|
|
print("Enable with:")
|
|
print(f" launchctl load {path}")
|
|
print("Disable with:")
|
|
print(f" launchctl unload {path}")
|
|
|
|
|
|
def _install_systemd(executable: str, *, system: bool = False) -> None:
|
|
"""Install a Linux systemd service unit file."""
|
|
_raw_path = os.environ.get("PATH", "/usr/local/bin:/usr/bin:/bin")
|
|
_safe_path = ":".join(p for p in _raw_path.split(":") if not p.startswith("/mnt/"))
|
|
_safe_path = _safe_path or "/usr/local/bin:/usr/bin:/bin"
|
|
|
|
unit = f"""\
|
|
[Unit]
|
|
Description=muxplex — web-based tmux session dashboard
|
|
After=network.target
|
|
|
|
[Service]
|
|
Type=simple
|
|
ExecStart={executable} -m muxplex
|
|
Restart=on-failure
|
|
RestartSec=5s
|
|
Environment=PATH={_safe_path}
|
|
|
|
[Install]
|
|
WantedBy={"multi-user.target" if system else "default.target"}
|
|
"""
|
|
|
|
if system:
|
|
path = _system_service_path
|
|
reload_cmd = (
|
|
"sudo systemctl daemon-reload && sudo systemctl enable --now muxplex"
|
|
)
|
|
else:
|
|
path = Path.home() / ".config" / "systemd" / "user" / "muxplex.service"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
reload_cmd = (
|
|
"systemctl --user daemon-reload && systemctl --user enable --now muxplex"
|
|
)
|
|
|
|
path.write_text(unit)
|
|
print(f"Service file written to {path}")
|
|
print(f"Enable with:\n {reload_cmd}")
|
|
|
|
|
|
def install_service(*, system: bool = False) -> None:
|
|
"""Install muxplex as a background service (launchd on macOS, systemd on Linux)."""
|
|
executable = sys.executable
|
|
|
|
if sys.platform == "darwin":
|
|
_install_launchd(executable)
|
|
else:
|
|
_install_systemd(executable, system=system)
|
|
|
|
|
|
def main() -> None:
|
|
"""CLI entry point."""
|
|
parser = argparse.ArgumentParser(
|
|
prog="muxplex",
|
|
description="muxplex — web-based tmux session dashboard",
|
|
)
|
|
parser.add_argument(
|
|
"--host", default="127.0.0.1", help="Bind host (default: 127.0.0.1)"
|
|
)
|
|
parser.add_argument("--port", type=int, default=8088, help="Port (default: 8088)")
|
|
parser.add_argument(
|
|
"--auth",
|
|
choices=["pam", "password"],
|
|
default="pam",
|
|
help="Authentication method: pam or password (default: pam)",
|
|
)
|
|
parser.add_argument(
|
|
"--session-ttl",
|
|
type=int,
|
|
default=604800,
|
|
dest="session_ttl",
|
|
help="Session TTL in seconds (default: 604800 = 7 days; 0 = browser session)",
|
|
)
|
|
|
|
sub = parser.add_subparsers(dest="command")
|
|
sub.add_parser("serve", help="Start the server (default)")
|
|
|
|
svc = sub.add_parser(
|
|
"install-service",
|
|
help="Install as a background service (systemd on Linux, launchd on macOS)",
|
|
)
|
|
svc.add_argument(
|
|
"--system", action="store_true", help="System-wide (requires sudo)"
|
|
)
|
|
|
|
sub.add_parser("show-password", help="Show the current muxplex password")
|
|
|
|
sub.add_parser(
|
|
"reset-secret", help="Regenerate signing secret (invalidates sessions)"
|
|
)
|
|
|
|
sub.add_parser("doctor", help="Check dependencies and system status")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "install-service":
|
|
install_service(system=args.system)
|
|
elif args.command == "show-password":
|
|
show_password()
|
|
elif args.command == "reset-secret":
|
|
reset_secret()
|
|
elif args.command == "doctor":
|
|
doctor()
|
|
else:
|
|
_check_dependencies()
|
|
serve(
|
|
host=args.host, port=args.port, auth=args.auth, session_ttl=args.session_ttl
|
|
)
|