import argparse
import json
import os
from pathlib import Path
import select
import signal
import socket
import subprocess
import sys
import tempfile
STOP_SIGNALS = (signal.SIGINT, signal.SIGTERM, signal.SIGHUP, signal.SIGTSTP)
SCRIPT = Path(__file__).resolve()
def system_tool(name):
for directory in ("/usr/bin", "/usr/sbin", "/bin", "/sbin"):
path = Path(directory) / name
if path.is_file() and os.access(path, os.X_OK):
return str(path)
raise RuntimeError(f"{name} is required (packages: sudo, kbd, util-linux)")
def vt_command(state, uid, gid, groups):
return [
system_tool("sudo"), "-n", "--", system_tool("openvt"), "-s", "-w", "--",
system_tool("setpriv"), f"--reuid={uid}", f"--regid={gid}",
"--groups=" + ",".join(map(str, sorted(set(groups) | {gid}))),
"--reset-env", "--no-new-privs", "--", system_tool("python3"),
str(SCRIPT), f"--_vt-child={state}",
]
class StopRequest:
def __init__(self):
self.received = []
self.saved = {}
def __enter__(self):
for number in STOP_SIGNALS:
self.saved[number] = signal.signal(number, self.handle)
return self
def handle(self, number, _frame):
self.received.append(number)
def take(self):
return self.received.pop() if self.received else 0
def __exit__(self, *_args):
for number, handler in self.saved.items():
signal.signal(number, handler)
def child_session(state):
if os.geteuid() == 0:
raise RuntimeError("the VT child must run as the invoking user")
config = json.loads((state / "config.json").read_text())
with open(config["log"], "ab", buffering=0) as log:
try:
with socket.socket(socket.AF_UNIX) as connection, StopRequest() as stop:
connection.connect(str(state / "control"))
return run_player(config, connection, stop, log)
except Exception as error:
log.write(f"medias-vt: {error}\n".encode())
return 1
def run_player(config, connection, stop, log):
player = subprocess.Popen(
config["command"], cwd=config["cwd"], env=config["environment"],
stdout=log, stderr=log,
)
try:
connected = True
while player.poll() is None:
number = stop.take()
if connected and select.select([connection], [], [], 0.05)[0]:
try:
message = connection.recv(64)
except ConnectionError:
message = b""
if not message:
connected = False
number = signal.SIGTERM
else:
number = message[-1]
if number in STOP_SIGNALS:
try:
player.send_signal(signal.SIGTERM if number == signal.SIGTSTP else number)
if number == signal.SIGTSTP:
player.send_signal(signal.SIGCONT)
except ProcessLookupError:
pass
if not connected:
try:
player.wait(timeout=0.05)
except subprocess.TimeoutExpired:
pass
status = player.returncode
status = status if status >= 0 else 128 - status
if connected:
try:
connection.sendall(f"{status}\n".encode())
except ConnectionError:
pass
return status
finally:
if player.poll() is None:
player.terminate()
player.wait()
def wait_session(helper, listener, stop):
connection = None
result = b""
pending = 0
closed = False
try:
while True:
helper_status = helper.poll()
pending = stop.take() or pending
if connection is not None and pending:
try:
connection.send(bytes([pending]))
except BlockingIOError:
pass
except ConnectionError:
connection.close()
connection, closed = None, True
else:
pending = 0
readers = [connection] if connection is not None else ([] if closed else [listener])
ready = select.select(readers, [], [], 0.05 if helper_status is None else 0)[0]
if connection is None and listener in ready:
connection, _ = listener.accept()
connection.setblocking(False)
continue
if connection is not None and connection in ready:
try:
message = connection.recv(64)
except ConnectionError:
message = b""
result += message
if message:
continue
connection.close()
connection, closed = None, True
if helper_status is not None:
break
finally:
if connection is not None:
connection.close()
if not result.strip():
raise RuntimeError(f"VT setup failed (status {helper.returncode}); see the log")
status = int(result.strip())
if helper.returncode not in (0, status):
raise RuntimeError(f"openvt failed (status {helper.returncode}); see the log")
return status
def launch(binary, arguments, log_path):
if os.geteuid() == 0:
raise RuntimeError("run medias-vt as your normal user; it invokes sudo for VT setup")
if not binary.is_file() or not os.access(binary, os.X_OK):
raise RuntimeError(f"{binary} is not executable; run cargo build --release")
log_path.parent.mkdir(parents=True, exist_ok=True)
with open(log_path, "ab", buffering=0) as log:
os.chmod(log.fileno(), 0o600)
log.write(b"\n=== medias-vt ===\n")
print(f"medias-vt: log: {log_path}", flush=True)
print("medias-vt: Ctrl-C stops medias and returns to the previous VT", flush=True)
subprocess.run([system_tool("sudo"), "-v"], check=True)
with tempfile.TemporaryDirectory(prefix="medias-vt-", dir="/tmp") as directory:
state = Path(directory)
config = {
"command": [str(binary), *arguments], "cwd": os.getcwd(),
"environment": dict(os.environ), "log": str(log_path),
}
(state / "config.json").write_text(json.dumps(config))
with socket.socket(socket.AF_UNIX) as listener, StopRequest() as stop:
listener.bind(str(state / "control"))
listener.listen(1)
command = vt_command(state, os.getuid(), os.getgid(), os.getgroups())
with subprocess.Popen(
command, stdin=subprocess.DEVNULL, stdout=log, stderr=log,
process_group=0,
) as helper:
try:
return wait_session(helper, listener, stop)
finally:
listener.close()
def main(arguments):
if len(arguments) == 1 and arguments[0].startswith("--_vt-child="):
return child_session(Path(arguments[0].split("=", 1)[1]))
parser = argparse.ArgumentParser(
description=__doc__, allow_abbrev=False,
epilog="Other arguments are passed to medias. Requires Python 3.11+, sudo, kbd, and util-linux.",
)
parser.add_argument("--binary", type=Path, default=SCRIPT.parent.parent / "target/release/medias")
state_home = Path(os.environ.get("XDG_STATE_HOME") or Path.home() / ".local/state")
parser.add_argument("--log", type=Path, default=state_home / "medias/console.log")
options, forwarded = parser.parse_known_args(arguments)
if forwarded[:1] == ["--"]:
forwarded.pop(0)
return launch(options.binary.resolve(), forwarded, options.log.expanduser().resolve())
if __name__ == "__main__":
try:
sys.exit(main(sys.argv[1:]))
except (OSError, RuntimeError, ValueError, subprocess.CalledProcessError) as error:
print(f"medias-vt: {error}", file=sys.stderr)
sys.exit(1)