DownloadRawCopy

1	#!/usr/bin/env python3
2	"""Secure terminal relay for brt -access serve/connect."""
3
4	from __future__ import annotations
5
6	import argparse
7	import hmac
8	import json
9	import os
10	import re
11	import secrets
12	import shutil
13	import signal
14	import socket
15	import string
16	import struct
17	import sys
18	import threading
19	import time
20	from dataclasses import dataclass, field
21	from datetime import datetime, timezone
22	from pathlib import Path
23	from typing import Any
24
25	IS_WINDOWS = sys.platform == "win32"
26
27	import select
28
29	if IS_WINDOWS:
30	import subprocess
31	else:
32	import pty
33	import termios
34	import tty
35
36	MSG_AUTH = 0x01
37	MSG_AUTH_OK = 0x02
38	MSG_AUTH_FAIL = 0x03
39	MSG_DATA = 0x04
40	MSG_CLOSE = 0x05
41
42	DEFAULT_AUDIT_LOG = str(Path.home() / ".brt" / "access.audit.log")
43	DEFAULT_CLIENT_INFO = str(Path.home() / ".brt" / "access.client")
44	DEFAULT_ADMIN_SOCK = (
45	"tcp:127.0.0.1:8767"
46	if IS_WINDOWS
47	else str(Path.home() / ".brt" / "access.admin.sock")
48	)
49	SESSION_ID_ALPHABET = string.ascii_letters + string.digits
50	MAX_COMMAND_LOG_LEN = 4096
51
52	# ANSI colors (bettercap-style dashboard)
53	C_RESET = "\033[0m"
54	C_DIM = "\033[2m"
55	C_BOLD = "\033[1m"
56	C_CYAN = "\033[36m"
57	C_GREEN = "\033[32m"
58	C_YELLOW = "\033[33m"
59	C_RED = "\033[31m"
60	C_MAGENTA = "\033[35m"
61	C_WHITE = "\033[97m"
62	C_BG = "\033[40m"
63
64
65	def send_msg(sock: socket.socket, msg_type: int, data: bytes = b"") -> None:
66	sock.sendall(struct.pack("!BH", msg_type, len(data)) + data)
67
68
69	def recv_exact(sock: socket.socket, n: int) -> bytes:
70	buf = b""
71	while len(buf) < n:
72	chunk = sock.recv(n - len(buf))
73	if not chunk:
74	raise ConnectionError("Connection closed")
75	buf += chunk
76	return buf
77
78
79	def recv_msg(sock: socket.socket) -> tuple[int, bytes]:
80	header = recv_exact(sock, 3)
81	msg_type, length = struct.unpack("!BH", header)
82	data = recv_exact(sock, length) if length else b""
83	return msg_type, data
84
85
86	def utc_now_iso() -> str:
87	dt = datetime.now(timezone.utc)
88	return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond // 1000:03d}Z"
89
90
91	def verify_token(provided: str, expected: str) -> bool:
92	return hmac.compare_digest(provided.encode(), expected.encode())
93
94
95	def default_audit_log_path() -> str:
96	return os.environ.get("BRT_ACCESS_AUDIT_LOG", DEFAULT_AUDIT_LOG)
97
98
99	def default_client_info_path() -> str:
100	return os.environ.get("BRT_ACCESS_CLIENT_INFO", DEFAULT_CLIENT_INFO)
101
102
103	def write_client_info(
104	endpoint: str,
105	token: str,
106	pid: int,
107	session_id: str = "",
108	) -> None:
109	path = Path(default_client_info_path())
110	path.parent.mkdir(parents=True, exist_ok=True)
111	payload = {
112	"endpoint": endpoint,
113	"token": token,
114	"pid": pid,
115	"session_id": session_id,
116	"started": utc_now_iso(),
117	}
118	path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
119
120
121	def read_client_info() -> dict[str, Any] | None:
122	path = Path(default_client_info_path())
123	if not path.is_file():
124	return None
125	try:
126	data = json.loads(path.read_text(encoding="utf-8"))
127	except (json.JSONDecodeError, OSError):
128	return None
129	return data if isinstance(data, dict) else None
130
131
132	def clear_client_info() -> None:
133	path = Path(default_client_info_path())
134	try:
135	path.unlink(missing_ok=True)
136	except OSError:
137	pass
138
139
140	def _pid_alive(pid: int) -> bool:
141	if pid <= 0:
142	return False
143	try:
144	os.kill(pid, 0)
145	return True
146	except OSError:
147	return False
148
149
150	def leave_client() -> int:
151	info = read_client_info()
152	if not info:
153	print("error: not connected to any access server", file=sys.stderr)
154	print(
155	"hint: type exit in the remote shell, or connect first with brt -access connect",
156	file=sys.stderr,
157	)
158	return 1
159
160	pid = int(info.get("pid") or 0)
161	session_id = str(info.get("session_id") or "")
162	endpoint = str(info.get("endpoint") or "")
163
164	if not _pid_alive(pid):
165	clear_client_info()
166	print("Disconnected (session was already closed)")
167	return 0
168
169	try:
170	os.kill(pid, signal.SIGTERM)
171	except OSError as exc:
172	clear_client_info()
173	print(f"error: could not disconnect: {exc}", file=sys.stderr)
174	return 1
175
176	for _ in range(30):
177	if not _pid_alive(pid):
178	break
179	time.sleep(0.1)
180
181	if _pid_alive(pid):
182	try:
183	os.kill(pid, getattr(signal, "SIGKILL", signal.SIGTERM))
184	except OSError:
185	pass
186
187	clear_client_info()
188	sid_msg = f" (session {session_id})" if session_id else ""
189	print(f"Disconnected from {endpoint}{sid_msg}")
190	return 0
191
192
193	def clear_serve_logs(audit_log: str | None = None, *, quiet: bool = False) -> int:
194	audit = audit_log or default_audit_log_path()
195	state_dir = Path.home() / ".brt"
196	cleared: list[str] = []
197	for path in (Path(audit), state_dir / "access.stdout", state_dir / "access.log"):
198	path.parent.mkdir(parents=True, exist_ok=True)
199	path.write_text("", encoding="utf-8")
200	cleared.append(str(path))
201	if not quiet:
202	print(f"Cleared {len(cleared)} log file(s)", flush=True)
203	return 0
204
205
206	def default_admin_sock_path() -> str:
207	return os.environ.get("BRT_ACCESS_ADMIN_SOCK", DEFAULT_ADMIN_SOCK)
208
209
210	def _default_shell() -> str:
211	if IS_WINDOWS:
212	for candidate in (
213	os.environ.get("BRT_SHELL"),
214	shutil.which("pwsh"),
215	shutil.which("powershell"),
216	os.environ.get("COMSPEC"),
217	):
218	if candidate:
219	return candidate
220	return "cmd.exe"
221	return os.environ.get("SHELL", "/bin/bash")
222
223
224	def _admin_is_tcp(spec: str) -> bool:
225	return spec.startswith("tcp:")
226
227
228	def _admin_open_connection(spec: str) -> socket.socket:
229	if _admin_is_tcp(spec):
230	host, port_str = spec[4:].rsplit(":", 1)
231	sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
232	sock.settimeout(2.0)
233	sock.connect((host, int(port_str)))
234	return sock
235	path = Path(spec)
236	if not path.exists():
237	raise FileNotFoundError(spec)
238	if not hasattr(socket, "AF_UNIX"):
239	raise OSError("Unix admin socket not supported on this platform")
240	sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
241	sock.settimeout(2.0)
242	sock.connect(str(path))
243	return sock
244
245
246	def _admin_bind_server(spec: str) -> socket.socket:
247	if _admin_is_tcp(spec):
248	host, port_str = spec[4:].rsplit(":", 1)
249	server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
250	server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
251	server.bind((host, int(port_str)))
252	server.listen(8)
253	return server
254	path = Path(spec)
255	path.parent.mkdir(parents=True, exist_ok=True)
256	if path.exists():
257	path.unlink()
258	if not hasattr(socket, "AF_UNIX"):
259	raise OSError("Unix admin socket not supported on this platform")
260	server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
261	server.bind(str(path))
262	server.listen(8)
263	return server
264
265
266	def _admin_close_server(spec: str, server: socket.socket) -> None:
267	try:
268	server.close()
269	except OSError:
270	pass
271	if not _admin_is_tcp(spec):
272	path = Path(spec)
273	if path.exists():
274	path.unlink()
275
276
277	def new_session_id(taken: set[str]) -> str:
278	for _ in range(200):
279	sid = "".join(secrets.choice(SESSION_ID_ALPHABET) for _ in range(6))
280	if sid not in taken:
281	return sid
282	raise RuntimeError("could not allocate session id")
283
284
285	class AccessSession:
286	def __init__(self, session_id: str, remote: str) -> None:
287	self.session_id = session_id
288	self.remote = remote
289	self.started = time.perf_counter()
290	self.last_action = self.started
291	self.bytes_in = 0
292	self.bytes_out = 0
293	self.command_buffer = ""
294	self.command_count = 0
295	self.shell_pid: int | None = None
296	self.shell = _default_shell()
297
298
299	@dataclass
300	class ActiveSessionHandle:
301	session_id: str
302	remote: str
303	started: float
304	shell_pid: int | None = None
305	conn: socket.socket | None = None
306	master_fd: int | None = None
307	proc: Any = None
308	stop: threading.Event = field(default_factory=threading.Event)
309
310
311	class SessionRegistry:
312	def __init__(self) -> None:
313	self._lock = threading.Lock()
314	self._sessions: dict[str, ActiveSessionHandle] = {}
315
316	def ids(self) -> set[str]:
317	with self._lock:
318	return set(self._sessions)
319
320	def register(self, handle: ActiveSessionHandle) -> None:
321	with self._lock:
322	self._sessions[handle.session_id] = handle
323
324	def unregister(self, session_id: str) -> None:
325	with self._lock:
326	self._sessions.pop(session_id, None)
327
328	def get(self, session_id: str) -> ActiveSessionHandle | None:
329	with self._lock:
330	return self._sessions.get(session_id)
331
332	def list_clients(self) -> list[dict[str, Any]]:
333	now = time.perf_counter()
334	with self._lock:
335	out = []
336	for sid, handle in sorted(self._sessions.items()):
337	uptime_ms = int((now - handle.started) * 1000)
338	out.append(
339	{
340	"id": sid,
341	"remote": handle.remote,
342	"shell_pid": handle.shell_pid,
343	"uptime_ms": uptime_ms,
344	}
345	)
346	return out
347
348	def disconnect(self, session_id: str, audit: AccessAuditLog | None = None) -> bool:
349	handle = self.get(session_id)
350	if not handle:
351	return False
352	handle.stop.set()
353	if handle.conn is not None:
354	try:
355	send_msg(handle.conn, MSG_CLOSE)
356	handle.conn.shutdown(socket.SHUT_RDWR)
357	except OSError:
358	pass
359	if handle.shell_pid:
360	try:
361	os.kill(handle.shell_pid, signal.SIGTERM)
362	except ProcessLookupError:
363	pass
364	if handle.proc is not None:
365	try:
366	handle.proc.terminate()
367	except OSError:
368	pass
369	if audit is not None:
370	audit.log(
371	"session_disconnect",
372	level="warn",
373	session=session_id,
374	remote=handle.remote,
375	reason="admin_disconnect",
376	)
377	return True
378
379
380	class AccessAuditLog:
381	def __init__(self, path: str) -> None:
382	self.path = Path(path)
383	self.path.parent.mkdir(parents=True, exist_ok=True)
384	self.server_started = time.perf_counter()
385	self._lock = threading.Lock()
386
387	def _write(self, record: dict) -> None:
388	line = json.dumps(record, ensure_ascii=False)
389	with self._lock:
390	with open(self.path, "a", encoding="utf-8") as fh:
391	fh.write(line + "\n")
392
393	def log(self, event: str, level: str = "info", **fields) -> None:
394	record = {"ts": utc_now_iso(), "level": level, "event": event}
395	record.update(fields)
396	self._write(record)
397
398	def session_log(
399	self,
400	session: AccessSession,
401	event: str,
402	level: str = "info",
403	**fields,
404	) -> None:
405	now = time.perf_counter()
406	delay_ms = int((now - session.last_action) * 1000)
407	since_session_ms = int((now - session.started) * 1000)
408	session.last_action = now
409	record = {
410	"ts": utc_now_iso(),
411	"level": level,
412	"event": event,
413	"session": session.session_id,
414	"remote": session.remote,
415	"delay_ms": delay_ms,
416	"since_session_ms": since_session_ms,
417	}
418	record.update(fields)
419	self._write(record)
420
421	def server_start(self, host: str, port: int, pid: int, admin_sock: str) -> None:
422	self.log(
423	"server_start",
424	host=host,
425	port=port,
426	pid=pid,
427	audit_log=str(self.path),
428	admin_sock=admin_sock,
429	)
430
431	def server_stop(self, reason: str) -> None:
432	uptime_ms = int((time.perf_counter() - self.server_started) * 1000)
433	self.log("server_stop", reason=reason, uptime_ms=uptime_ms)
434
435
436	def _sanitize_command(text: str) -> str:
437	text = text.replace("\r", "").replace("\n", " ").strip()
438	text = re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", text)
439	if len(text) > MAX_COMMAND_LOG_LEN:
440	text = text[:MAX_COMMAND_LOG_LEN] + "…"
441	return text
442
443
444	def _log_client_input(session: AccessSession, audit: AccessAuditLog, data: bytes) -> None:
445	session.bytes_in += len(data)
446	chunk = data.decode("utf-8", errors="replace")
447	for ch in chunk:
448	if ch in ("\n", "\r"):
449	line = _sanitize_command(session.command_buffer)
450	session.command_buffer = ""
451	if not line:
452	continue
453	session.command_count += 1
454	audit.session_log(
455	session,
456	"user_command",
457	action="command",
458	command=line,
459	command_index=session.command_count,
460	)
461	elif ch in ("\x7f", "\b"):
462	session.command_buffer = session.command_buffer[:-1]
463	elif ch == "\x03":
464	audit.session_log(session, "user_input", action="interrupt", key="^C")
465	elif ch == "\x1a":
466	audit.session_log(session, "user_input", action="suspend", key="^Z")
467	elif ch == "\x04":
468	audit.session_log(session, "user_input", action="eof", key="^D")
469	elif ch.isprintable() or ch == "\t":
470	session.command_buffer += ch
471
472
473	def set_raw(enabled: bool):
474	if IS_WINDOWS or not sys.stdin.isatty():
475	return None
476	fd = sys.stdin.fileno()
477	old = termios.tcgetattr(fd)
478	if enabled:
479	tty.setraw(fd)
480	else:
481	termios.tcsetattr(fd, termios.TCSADRAIN, old)
482	return old
483
484
485	def relay_io_windows(sock: socket.socket) -> None:
486	stop = threading.Event()
487
488	def _reader() -> None:
489	while not stop.is_set():
490	try:
491	msg_type, data = recv_msg(sock)
492	if msg_type == MSG_CLOSE:
493	break
494	if msg_type == MSG_DATA:
495	sys.stdout.buffer.write(data)
496	sys.stdout.buffer.flush()
497	except (ConnectionError, OSError):
498	break
499
500	thread = threading.Thread(target=_reader, daemon=True)
501	thread.start()
502	try:
503	while True:
504	data = sys.stdin.buffer.read(1)
505	if not data:
506	break
507	send_msg(sock, MSG_DATA, data)
508	finally:
509	stop.set()
510	thread.join(timeout=0.5)
511
512
513	def relay_io(sock: socket.socket, master_fd=None) -> None:
514	if IS_WINDOWS and master_fd is None:
515	relay_io_windows(sock)
516	return
517	if IS_WINDOWS:
518	raise RuntimeError("Windows relay_io requires pipe mode")
519	old_attrs = set_raw(True)
520	try:
521	while True:
522	rlist = [sock]
523	if master_fd is not None:
524	rlist.append(master_fd)
525	else:
526	rlist.append(sys.stdin.fileno())
527	readable, _, _ = select.select(rlist, [], [], 0.5)
528	if sock in readable:
529	msg_type, data = recv_msg(sock)
530	if msg_type == MSG_CLOSE:
531	break
532	if msg_type == MSG_DATA:
533	os.write(sys.stdout.fileno(), data)
534	if master_fd is not None and master_fd in readable:
535	try:
536	data = os.read(master_fd, 4096)
537	if not data:
538	break
539	send_msg(sock, MSG_DATA, data)
540	except OSError:
541	break
542	if master_fd is None and sys.stdin.fileno() in readable:
543	data = os.read(sys.stdin.fileno(), 4096)
544	if not data:
545	break
546	send_msg(sock, MSG_DATA, data)
547	finally:
548	if old_attrs:
549	termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, old_attrs)
550
551
552	def relay_io_server_process(
553	sock: socket.socket,
554	proc: subprocess.Popen,
555	session: AccessSession,
556	audit: AccessAuditLog,
557	stop: threading.Event,
558	) -> str:
559	reason = "disconnect"
560	write_lock = threading.Lock()
561
562	def _stdout_reader() -> None:
563	nonlocal reason
564	assert proc.stdout is not None
565	while not stop.is_set():
566	try:
567	data = proc.stdout.read(4096)
568	if not data:
569	reason = "shell_exit"
570	stop.set()
571	break
572	session.bytes_out += len(data)
573	send_msg(sock, MSG_DATA, data)
574	except OSError as exc:
575	audit.session_log(
576	session,
577	"io_error",
578	level="error",
579	direction="stdout_read",
580	error=str(exc),
581	)
582	reason = "io_error"
583	stop.set()
584	break
585
586	reader = threading.Thread(target=_stdout_reader, daemon=True)
587	reader.start()
588	try:
589	while not stop.is_set():
590	try:
591	msg_type, data = recv_msg(sock)
592	except (ConnectionError, OSError) as exc:
593	audit.session_log(
594	session,
595	"io_error",
596	level="error",
597	direction="socket",
598	error=str(exc),
599	)
600	reason = "io_error"
601	break
602	if msg_type == MSG_CLOSE:
603	reason = "client_close"
604	break
605	if msg_type == MSG_DATA:
606	_log_client_input(session, audit, data)
607	if proc.stdin is not None:
608	with write_lock:
609	proc.stdin.write(data)
610	proc.stdin.flush()
611	finally:
612	stop.set()
613	reader.join(timeout=0.5)
614	if stop.is_set() and reason == "disconnect":
615	reason = "admin_disconnect"
616	return reason
617
618
619	def relay_io_server(
620	sock: socket.socket,
621	master_fd: int,
622	session: AccessSession,
623	audit: AccessAuditLog,
624	stop: threading.Event,
625	) -> str:
626	if IS_WINDOWS:
627	raise RuntimeError("relay_io_server is Unix-only")
628	reason = "disconnect"
629	try:
630	while not stop.is_set():
631	readable, _, _ = select.select([sock, master_fd], [], [], 0.25)
632	if stop.is_set():
633	reason = "admin_disconnect"
634	break
635	if sock in readable:
636	msg_type, data = recv_msg(sock)
637	if msg_type == MSG_CLOSE:
638	reason = "client_close"
639	break
640	if msg_type == MSG_DATA:
641	_log_client_input(session, audit, data)
642	os.write(master_fd, data)
643	if master_fd in readable:
644	try:
645	data = os.read(master_fd, 4096)
646	if not data:
647	reason = "shell_exit"
648	break
649	session.bytes_out += len(data)
650	send_msg(sock, MSG_DATA, data)
651	except OSError as exc:
652	audit.session_log(
653	session,
654	"io_error",
655	level="error",
656	direction="pty_read",
657	error=str(exc),
658	)
659	reason = "io_error"
660	break
661	except (ConnectionError, OSError) as exc:
662	audit.session_log(
663	session,
664	"io_error",
665	level="error",
666	direction="socket",
667	error=str(exc),
668	)
669	reason = "io_error"
670	if stop.is_set() and reason == "disconnect":
671	reason = "admin_disconnect"
672	return reason
673
674
675	def handle_client(
676	conn: socket.socket,
677	addr,
678	token: str,
679	audit: AccessAuditLog,
680	registry: SessionRegistry,
681	) -> None:
682	remote = f"{addr[0]}:{addr[1]}"
683	session_id = new_session_id(registry.ids())
684	session = AccessSession(session_id, remote)
685	handle = ActiveSessionHandle(session_id, remote, session.started, conn=conn)
686	connect_started = time.perf_counter()
687
688	audit.session_log(session, "client_connect", action="connect")
689	print(f"[+] Connection from {remote} (session {session_id})", file=sys.stderr)
690
691	pid = 0
692	master_fd: int | None = None
693	proc: subprocess.Popen | None = None
694	try:
695	auth_started = time.perf_counter()
696	msg_type, data = recv_msg(conn)
697	if msg_type != MSG_AUTH:
698	send_msg(conn, MSG_AUTH_FAIL, b"Expected auth")
699	audit.session_log(
700	session,
701	"auth_fail",
702	level="warn",
703	reason="expected_auth",
704	auth_delay_ms=int((time.perf_counter() - auth_started) * 1000),
705	)
706	return
707
708	provided = data.decode("utf-8", errors="replace")
709	auth_delay_ms = int((time.perf_counter() - auth_started) * 1000)
710	if not verify_token(provided, token):
711	send_msg(conn, MSG_AUTH_FAIL, b"Invalid token")
712	audit.session_log(
713	session,
714	"auth_fail",
715	level="warn",
716	reason="invalid_token",
717	auth_delay_ms=auth_delay_ms,
718	)
719	print(f"[!] Auth failed from {remote}", file=sys.stderr)
720	return
721
722	send_msg(
723	conn,
724	MSG_AUTH_OK,
725	json.dumps(
726	{"message": "Authenticated", "session_id": session_id},
727	).encode(),
728	)
729	audit.session_log(
730	session,
731	"auth_ok",
732	auth_delay_ms=auth_delay_ms,
733	connect_delay_ms=int((time.perf_counter() - connect_started) * 1000),
734	)
735	print(f"[+] Authenticated {remote} (session {session_id})", file=sys.stderr)
736
737	if IS_WINDOWS:
738	shell_cmd = [session.shell]
739	if session.shell.lower().endswith(("cmd.exe", "cmd")):
740	shell_cmd = [session.shell]
741	elif "powershell" in session.shell.lower() or session.shell.lower().endswith("pwsh"):
742	shell_cmd = [session.shell, "-NoLogo"]
743	proc = subprocess.Popen(
744	shell_cmd,
745	stdin=subprocess.PIPE,
746	stdout=subprocess.PIPE,
747	stderr=subprocess.STDOUT,
748	text=False,
749	bufsize=0,
750	creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
751	)
752	session.shell_pid = proc.pid
753	handle.shell_pid = proc.pid
754	handle.proc = proc
755	registry.register(handle)
756	audit.session_log(
757	session,
758	"shell_spawn",
759	shell=session.shell,
760	shell_pid=proc.pid,
761	)
762	reason = relay_io_server_process(conn, proc, session, audit, handle.stop)
763	exit_code = proc.poll()
764	pid = proc.pid
765	else:
766	pid, master_fd = pty.fork()
767	if pid == 0:
768	os.execvp(session.shell, [session.shell, "-l"])
769	sys.exit(1)
770
771	session.shell_pid = pid
772	handle.shell_pid = pid
773	handle.master_fd = master_fd
774	registry.register(handle)
775
776	audit.session_log(session, "shell_spawn", shell=session.shell, shell_pid=pid)
777
778	reason = relay_io_server(conn, master_fd, session, audit, handle.stop)
779
780	exit_code = None
781	try:
782	os.kill(pid, signal.SIGTERM)
783	_, status = os.waitpid(pid, 0)
784	if os.WIFEXITED(status):
785	exit_code = os.WEXITSTATUS(status)
786	elif os.WIFSIGNALED(status):
787	exit_code = -os.WTERMSIG(status)
788	except (ProcessLookupError, ChildProcessError):
789	pass
790
791	duration_ms = int((time.perf_counter() - session.started) * 1000)
792	audit.session_log(
793	session,
794	"session_end",
795	reason=reason,
796	duration_ms=duration_ms,
797	bytes_in=session.bytes_in,
798	bytes_out=session.bytes_out,
799	commands=session.command_count,
800	shell_pid=pid,
801	exit_code=exit_code,
802	)
803	print(
804	f"[-] Session {session_id} ended ({reason}, {duration_ms}ms)",
805	file=sys.stderr,
806	)
807	except (ConnectionError, OSError) as exc:
808	audit.session_log(session, "session_error", level="error", error=str(exc))
809	print(f"[!] Connection error ({remote}): {exc}", file=sys.stderr)
810	finally:
811	registry.unregister(session_id)
812	if proc is not None:
813	try:
814	if proc.poll() is None:
815	proc.terminate()
816	except OSError:
817	pass
818	if master_fd is not None:
819	try:
820	os.close(master_fd)
821	except OSError:
822	pass
823	conn.close()
824
825
826	def _admin_handle_connection(conn: socket.socket, registry: SessionRegistry, audit: AccessAuditLog) -> None:
827	with conn:
828	try:
829	raw = conn.recv(65536).decode("utf-8", errors="replace").strip()
830	req = json.loads(raw) if raw else {}
831	cmd = req.get("cmd", "")
832	if cmd == "list":
833	resp = {"ok": True, "clients": registry.list_clients()}
834	elif cmd == "disconnect":
835	sid = str(req.get("id", ""))
836	ok = registry.disconnect(sid, audit)
837	resp = (
838	{"ok": True, "id": sid}
839	if ok
840	else {"ok": False, "error": f"unknown session {sid}"}
841	)
842	else:
843	resp = {"ok": False, "error": f"unknown cmd {cmd}"}
844	conn.sendall((json.dumps(resp) + "\n").encode())
845	except (json.JSONDecodeError, OSError):
846	conn.sendall(b'{"ok":false,"error":"bad request"}\n')
847
848
849	def admin_server_loop(
850	admin_spec: str,
851	registry: SessionRegistry,
852	audit: AccessAuditLog,
853	stop: threading.Event,
854	) -> None:
855	server = _admin_bind_server(admin_spec)
856	server.settimeout(0.5)
857
858	while not stop.is_set():
859	try:
860	conn, _ = server.accept()
861	except socket.timeout:
862	continue
863	except OSError:
864	break
865	threading.Thread(
866	target=_admin_handle_connection,
867	args=(conn, registry, audit),
868	daemon=True,
869	).start()
870
871	_admin_close_server(admin_spec, server)
872
873
874	def admin_request(admin_spec: str, payload: dict) -> dict:
875	try:
876	sock = _admin_open_connection(admin_spec)
877	except FileNotFoundError:
878	return {"ok": False, "error": "access server not running"}
879	except OSError as exc:
880	return {"ok": False, "error": str(exc)}
881	try:
882	with sock:
883	sock.sendall((json.dumps(payload) + "\n").encode())
884	data = sock.recv(65536).decode("utf-8", errors="replace").strip()
885	return json.loads(data) if data else {"ok": False, "error": "empty response"}
886	except (OSError, json.JSONDecodeError) as exc:
887	return {"ok": False, "error": str(exc)}
888
889
890	def _fmt_bytes(n: int) -> str:
891	if n < 1024:
892	return f"{n}B"
893	if n < 1024 * 1024:
894	return f"{n // 1024}K"
895	return f"{n / (1024 * 1024):.1f}M"
896
897
898	def _fmt_uptime(ms: int) -> str:
899	sec = max(0, ms // 1000)
900	if sec < 60:
901	return f"{sec}s"
902	mins, sec = divmod(sec, 60)
903	if mins < 60:
904	return f"{mins}m{sec:02d}s"
905	hrs, mins = divmod(mins, 60)
906	return f"{hrs}h{mins:02d}m"
907
908
909	def _event_detail(rec: dict) -> str:
910	event = rec.get("event", "")
911	if event == "server_start":
912	host = rec.get("host", "")
913	port = rec.get("port", "")
914	if host != "" and port != "":
915	return f"{host}:{port}"
916	if event == "user_command":
917	return rec.get("command", "")
918	if event in ("auth_fail", "session_end", "session_disconnect"):
919	return rec.get("reason", "") or rec.get("error", "")
920	if event == "user_input":
921	return rec.get("key", rec.get("action", ""))
922	for key in ("command", "error", "shell", "host", "reason"):
923	if rec.get(key):
924	return str(rec[key])
925	return ""
926
927
928	def _level_color(level: str) -> str:
929	if level == "error":
930	return C_RED
931	if level == "warn":
932	return C_YELLOW
933	if level == "info":
934	return C_GREEN
935	return C_WHITE
936
937
938	def _event_color(event: str) -> str:
939	if event in ("user_command", "user_input"):
940	return C_CYAN
941	if event in ("auth_fail", "session_error", "io_error"):
942	return C_RED
943	if event in ("session_end", "session_disconnect"):
944	return C_YELLOW
945	if event in ("auth_ok", "client_connect", "shell_spawn"):
946	return C_GREEN
947	return C_WHITE
948
949
950	def _parse_audit_line(line: str) -> dict | None:
951	line = line.strip()
952	if not line:
953	return None
954	try:
955	return json.loads(line)
956	except json.JSONDecodeError:
957	return None
958
959
960	def _short_ts(ts: str) -> str:
961	if "T" in ts:
962	return ts.split("T", 1)[1].replace("Z", "")[:12]
963	return ts[:12]
964
965
966	def _draw_dashboard(
967	cols: int,
968	rows: int,
969	clients: list[dict],
970	events: list[dict],
971	audit_path: str,
972	server_online: bool,
973	*,
974	endpoint: str = "",
975	token: str = "",
976	status_msg: str = "",
977	cmd_input: str = "",
978	interactive: bool = False,
979	) -> None:
980	out: list[str] = []
981	title = f"{C_BOLD}{C_CYAN}━━ brt access — live audit ━━{C_RESET}"
982	if endpoint:
983	title += f" {C_GREEN}{endpoint}{C_RESET}"
984	out.append(title)
985	if token:
986	tok_show = token if len(token) <= 20 else f"{token[:8]}…{token[-4:]}"
987	out.append(
988	f"{C_DIM}token {C_WHITE}{tok_show}{C_RESET} · "
989	f"{audit_path} · "
990	f"{'online' if server_online else 'offline'}{C_RESET}"
991	)
992	else:
993	out.append(
994	f"{C_DIM}{audit_path} · "
995	f"{'server online' if server_online else 'server offline'}"
996	f"{' · q quit' if not interactive else ''}{C_RESET}"
997	)
998
999	client_line = f"{C_BOLD} Connected ({len(clients)}){C_RESET} "
1000	if clients:
1001	parts = []
1002	for c in clients[:8]:
1003	parts.append(f"{C_MAGENTA}{c['id']}{C_RESET}@{c['remote']}")
1004	client_line += " ".join(parts)
1005	if len(clients) > 8:
1006	client_line += f" {C_DIM}+{len(clients) - 8} more{C_RESET}"
1007	else:
1008	client_line += f"{C_DIM}(none){C_RESET}"
1009	out.append(client_line)
1010	out.append("")
1011
1012	out.append(f"{C_BOLD} Events{C_RESET}")
1013	out.append(
1014	f"{C_DIM} {'TIME':<12} {'ID':<6} {'EVENT':<16} {'DETAIL':<28} +ms{C_RESET}"
1015	)
1016
1017	footer_lines = 2 if interactive else 0
1018	view_h = max(3, rows - len(out) - footer_lines - 1)
1019	for rec in events[-view_h:]:
1020	ts = _short_ts(str(rec.get("ts", "")))
1021	sid = str(rec.get("session", ""))[:6]
1022	event = str(rec.get("event", ""))[:16]
1023	detail = _event_detail(rec)[: max(8, cols - 58)]
1024	delay = rec.get("delay_ms", "")
1025	delay_s = str(delay) if delay != "" else ""
1026	lc = _level_color(str(rec.get("level", "info")))
1027	ec = _event_color(event)
1028	out.append(
1029	f" {C_DIM}{ts:<12}{C_RESET} "
1030	f"{C_MAGENTA}{sid:<6}{C_RESET} "
1031	f"{ec}{event:<16}{C_RESET} "
1032	f"{lc}{detail:<28}{C_RESET} "
1033	f"{C_DIM}{delay_s}{C_RESET}"
1034	)
1035
1036	body_lines = rows - footer_lines - 1
1037	while len(out) < body_lines:
1038	out.append("")
1039
1040	if interactive:
1041	hint = "connect · stop · disconnect <id> · clear · help · quit"
1042	if status_msg:
1043	out.append(f"{C_YELLOW}{status_msg[: cols - 1]}{C_RESET}")
1044	else:
1045	out.append(f"{C_DIM}{hint[: cols - 1]}{C_RESET}")
1046	prompt = f"{C_BOLD}>{C_RESET} {cmd_input}"
1047	out.append(prompt)
1048
1049	screen = "\033[H\033[2J\033[?25l" + "\n".join(out[: rows - 1])
1050	if interactive:
1051	screen += "\033[?25h"
1052	prompt_row = rows
1053	cursor_col = len(f"> {cmd_input}") + 1
1054	screen += f"\033[{prompt_row};{cursor_col}H"
1055	else:
1056	screen += "\033[?25h"
1057	sys.stdout.write(screen)
1058	sys.stdout.flush()
1059
1060
1061	def _copy_to_clipboard(text: str) -> bool:
1062	import subprocess
1063
1064	candidates: list[list[str]] = []
1065	if IS_WINDOWS:
1066	candidates.append(
1067	[
1068	"powershell",
1069	"-NoProfile",
1070	"-Command",
1071	"Set-Clipboard -Value ([Console]::In.ReadToEnd())",
1072	]
1073	)
1074	if shutil.which("clip"):
1075	candidates.insert(0, ["clip"])
1076	if shutil.which("pbcopy"):
1077	candidates.append(["pbcopy"])
1078	if shutil.which("wl-copy"):
1079	candidates.append(["wl-copy"])
1080	if shutil.which("xclip"):
1081	candidates.append(["xclip", "-selection", "clipboard"])
1082	if shutil.which("xsel"):
1083	candidates.append(["xsel", "--clipboard", "--input"])
1084
1085	data = text.encode("utf-8")
1086	for cmd in candidates:
1087	try:
1088	if IS_WINDOWS and cmd[0] == "clip":
1089	subprocess.run(cmd, input=data, check=True, timeout=2)
1090	elif IS_WINDOWS and "Set-Clipboard" in " ".join(cmd):
1091	subprocess.run(
1092	[
1093	"powershell",
1094	"-NoProfile",
1095	"-Command",
1096	f"Set-Clipboard -Value {json.dumps(text)}",
1097	],
1098	check=True,
1099	timeout=2,
1100	)
1101	else:
1102	subprocess.run(cmd, input=data, check=True, timeout=2)
1103	return True
1104	except (OSError, subprocess.SubprocessError):
1105	continue
1106	return False
1107
1108
1109	def _connect_command(endpoint: str, token: str) -> str:
1110	cli = os.environ.get("BRT_CONNECT_CLI")
1111	if not cli:
1112	cli = "brtwin" if IS_WINDOWS else "brt"
1113	return f"{cli} -access connect {endpoint} {token}"
1114
1115
1116	def _copy_connect_command(endpoint: str, token: str) -> tuple[str, bool]:
1117	if not endpoint or not token:
1118	return "Connect info not available", False
1119	cmd = _connect_command(endpoint, token)
1120	if _copy_to_clipboard(cmd):
1121	return "Copied connect command to clipboard", True
1122	return cmd, False
1123
1124
1125	def _stop_server_pid(pid: int) -> bool:
1126	if pid <= 0:
1127	return False
1128	try:
1129	os.kill(pid, signal.SIGTERM)
1130	except ProcessLookupError:
1131	return True
1132	except OSError:
1133	return False
1134	for _ in range(30):
1135	time.sleep(0.1)
1136	try:
1137	os.kill(pid, 0)
1138	except ProcessLookupError:
1139	return True
1140	try:
1141	os.kill(pid, signal.SIGKILL)
1142	except ProcessLookupError:
1143	pass
1144	return True
1145
1146
1147	def _exec_console_command(
1148	line: str,
1149	admin_sock: str,
1150	server_pid: int | None,
1151	owns_server: bool,
1152	endpoint: str = "",
1153	token: str = "",
1154	audit_log: str = "",
1155	console_state: dict[str, bool] | None = None,
1156	) -> tuple[str, int | None]:
1157	raw = line.strip()
1158	if not raw:
1159	return "", None
1160
1161	parts = raw.split()
1162	cmd = parts[0].lower()
1163
1164	if cmd in ("quit", "q", "exit"):
1165	if owns_server and server_pid:
1166	_stop_server_pid(server_pid)
1167	return "Server stopped", 0
1168	return "", 0
1169
1170	if cmd == "stop":
1171	if server_pid:
1172	_stop_server_pid(server_pid)
1173	return "Server stopped", 0
1174	return "No server process", None
1175
1176	if cmd in ("help", "?"):
1177	return "connect · stop · disconnect <id> · clear · quit", None
1178
1179	if cmd in ("clear", "clear-logs", "cls"):
1180	clear_serve_logs(audit_log or default_audit_log_path(), quiet=True)
1181	if console_state is not None:
1182	console_state["clear_logs"] = True
1183	return "Logs cleared", None
1184
1185	if cmd == "connect":
1186	msg, _ok = _copy_connect_command(endpoint, token)
1187	return msg, None
1188
1189	session_id = ""
1190	if cmd in ("disconnect", "dc", "kick"):
1191	if len(parts) < 2:
1192	return "Usage: disconnect <6-char-id>", None
1193	session_id = parts[1]
1194	elif re.fullmatch(r"[A-Za-z0-9]{6}", parts[0]):
1195	session_id = parts[0]
1196
1197	if session_id:
1198	if not re.fullmatch(r"[A-Za-z0-9]{6}", session_id):
1199	return "Session id must be 6 alphanumeric characters", None
1200	resp = admin_request(admin_sock, {"cmd": "disconnect", "id": session_id})
1201	if resp.get("ok"):
1202	return f"Disconnected {session_id}", None
1203	return f"Error: {resp.get('error', 'disconnect failed')}", None
1204
1205	return f"Unknown command: {raw} (try help)", None
1206
1207
1208	def _read_stdin_keys(timeout: float) -> list[bytes]:
1209	chunks: list[bytes] = []
1210	if IS_WINDOWS:
1211	import msvcrt
1212
1213	deadline = time.perf_counter() + timeout
1214	while time.perf_counter() < deadline or chunks:
1215	if msvcrt.kbhit():
1216	chunks.append(msvcrt.getch())
1217	deadline = time.perf_counter() + 0.05
1218	elif chunks:
1219	break
1220	else:
1221	time.sleep(0.03)
1222	return chunks
1223	try:
1224	fd = sys.stdin.fileno()
1225	except (ValueError, OSError):
1226	return chunks
1227	while True:
1228	readable, _, _ = select.select([fd], [], [], timeout if not chunks else 0.0)
1229	if not readable:
1230	break
1231	ch = os.read(fd, 64)
1232	if not ch:
1233	break
1234	chunks.append(ch)
1235	timeout = 0.0
1236	return chunks
1237
1238
1239	def _apply_input_keys(keys: list[bytes], buf: str) -> tuple[str, bool]:
1240	"""Return (new_buffer, submit)."""
1241	submit = False
1242	for chunk in keys:
1243	for byte in chunk:
1244	if byte in (10, 13):
1245	submit = True
1246	elif byte in (127, 8):
1247	buf = buf[:-1]
1248	elif byte == 3:
1249	if buf:
1250	buf = ""
1251	else:
1252	submit = True
1253	elif 32 <= byte <= 126:
1254	buf += chr(byte)
1255	return buf, submit
1256
1257
1258	def _enable_windows_vt() -> None:
1259	if not IS_WINDOWS:
1260	return
1261	try:
1262	import ctypes
1263
1264	kernel32 = ctypes.windll.kernel32
1265	handle = kernel32.GetStdHandle(-11)
1266	mode = ctypes.c_uint()
1267	if kernel32.GetConsoleMode(handle, ctypes.byref(mode)):
1268	kernel32.SetConsoleMode(handle, mode.value | 0x0004)
1269	except (OSError, AttributeError):
1270	pass
1271
1272
1273	def watch_logs(
1274	audit_log: str,
1275	admin_sock: str,
1276	follow: bool = True,
1277	*,
1278	interactive: bool = False,
1279	server_pid: int | None = None,
1280	endpoint: str = "",
1281	token: str = "",
1282	owns_server: bool = False,
1283	) -> int:
1284	path = Path(audit_log)
1285	if not path.exists():
1286	path.parent.mkdir(parents=True, exist_ok=True)
1287	path.touch()
1288
1289	events: list[dict] = []
1290	pos = 0
1291	if path.stat().st_size > 0:
1292	with open(path, encoding="utf-8") as fh:
1293	for line in fh:
1294	rec = _parse_audit_line(line)
1295	if rec:
1296	events.append(rec)
1297	events = events[-500:]
1298	pos = path.stat().st_size
1299
1300	cmd_input = ""
1301	status_msg = ""
1302	copied_connect = False
1303	console_state: dict[str, bool] = {"clear_logs": False}
1304	old_tty = None
1305	if interactive and sys.stdin.isatty():
1306	if IS_WINDOWS:
1307	_enable_windows_vt()
1308	else:
1309	old_tty = termios.tcgetattr(sys.stdin.fileno())
1310	tty.setcbreak(sys.stdin.fileno())
1311
1312	exit_code = 0
1313	try:
1314	while True:
1315	cols = shutil.get_terminal_size(fallback=(100, 30)).columns
1316	rows = shutil.get_terminal_size(fallback=(100, 30)).lines
1317
1318	if server_pid and server_pid > 0:
1319	try:
1320	os.kill(server_pid, 0)
1321	except ProcessLookupError:
1322	status_msg = "Server exited"
1323	_draw_dashboard(
1324	cols,
1325	rows,
1326	[],
1327	events,
1328	audit_log,
1329	False,
1330	endpoint=endpoint,
1331	token=token,
1332	status_msg=status_msg,
1333	cmd_input=cmd_input,
1334	interactive=interactive,
1335	)
1336	exit_code = 0
1337	break
1338
1339	admin = admin_request(admin_sock, {"cmd": "list"})
1340	clients = admin.get("clients", []) if admin.get("ok") else []
1341	server_online = admin.get("ok", False)
1342
1343	with open(path, encoding="utf-8") as fh:
1344	fh.seek(pos)
1345	for line in fh:
1346	rec = _parse_audit_line(line)
1347	if rec:
1348	events.append(rec)
1349	pos = fh.tell()
1350	events = events[-500:]
1351
1352	if (
1353	interactive
1354	and owns_server
1355	and endpoint
1356	and token
1357	and not copied_connect
1358	):
1359	msg, copied_connect = _copy_connect_command(endpoint, token)
1360	if copied_connect:
1361	status_msg = msg
1362
1363	_draw_dashboard(
1364	cols,
1365	rows,
1366	clients,
1367	events,
1368	audit_log,
1369	server_online,
1370	endpoint=endpoint,
1371	token=token,
1372	status_msg=status_msg,
1373	cmd_input=cmd_input,
1374	interactive=interactive,
1375	)
1376
1377	if not follow:
1378	break
1379
1380	keys = _read_stdin_keys(0.35)
1381	if interactive:
1382	if keys and not cmd_input:
1383	for chunk in keys:
1384	if b"\x03" in chunk:
1385	if owns_server and server_pid:
1386	_stop_server_pid(server_pid)
1387	exit_code = 0
1388	follow = False
1389	break
1390	if not follow:
1391	break
1392	cmd_input, submit = _apply_input_keys(keys, cmd_input)
1393	if submit:
1394	status_msg, done = _exec_console_command(
1395	cmd_input,
1396	admin_sock,
1397	server_pid,
1398	owns_server,
1399	endpoint=endpoint,
1400	token=token,
1401	audit_log=audit_log,
1402	console_state=console_state,
1403	)
1404	cmd_input = ""
1405	if console_state.get("clear_logs"):
1406	events.clear()
1407	pos = 0
1408	console_state["clear_logs"] = False
1409	if done is not None:
1410	exit_code = done
1411	break
1412	elif keys:
1413	for chunk in keys:
1414	if b"q" in chunk or b"Q" in chunk or b"\x03" in chunk:
1415	exit_code = 0
1416	follow = False
1417	break
1418	finally:
1419	if old_tty:
1420	termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, old_tty)
1421	sys.stdout.write("\033[?25h\n")
1422	sys.stdout.flush()
1423	return exit_code
1424
1425
1426	def print_clients(admin_sock: str) -> int:
1427	resp = admin_request(admin_sock, {"cmd": "list"})
1428	if not resp.get("ok"):
1429	print(f"error: {resp.get('error', 'unknown')}", file=sys.stderr)
1430	return 1
1431	clients = resp.get("clients", [])
1432	if not clients:
1433	print("No connected sessions.")
1434	return 0
1435	print(f"{'ID':<6} {'REMOTE':<24} {'UPTIME':<8} PID")
1436	for c in clients:
1437	print(
1438	f"{c['id']:<6} {c['remote']:<24} "
1439	f"{_fmt_uptime(c.get('uptime_ms', 0)):<8} {c.get('shell_pid') or '-'}"
1440	)
1441	return 0
1442
1443
1444	def disconnect_client(admin_sock: str, session_id: str) -> int:
1445	if not re.fullmatch(r"[A-Za-z0-9]{6}", session_id):
1446	print("error: session id must be 6 alphanumeric characters", file=sys.stderr)
1447	return 1
1448	resp = admin_request(admin_sock, {"cmd": "disconnect", "id": session_id})
1449	if not resp.get("ok"):
1450	print(f"error: {resp.get('error', 'disconnect failed')}", file=sys.stderr)
1451	return 1
1452	print(f"Disconnected session {session_id}")
1453	return 0
1454
1455
1456	def serve(host: str, port: int, token: str, audit_log: str, admin_sock: str) -> None:
1457	audit = AccessAuditLog(audit_log)
1458	registry = SessionRegistry()
1459	admin_stop = threading.Event()
1460
1461	audit.server_start(host, port, os.getpid(), admin_sock)
1462	print(f"BRT_ACCESS_AUDIT_LOG={audit_log}", flush=True)
1463	print(f"BRT_ACCESS_ADMIN_SOCK={admin_sock}", flush=True)
1464
1465	admin_thread = threading.Thread(
1466	target=admin_server_loop,
1467	args=(admin_sock, registry, audit, admin_stop),
1468	daemon=True,
1469	)
1470	admin_thread.start()
1471
1472	server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1473	server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1474	server.bind((host, port))
1475	server.listen(16)
1476
1477	def shutdown(sig, frame):
1478	sig_name = signal.Signals(sig).name if hasattr(signal, "Signals") else str(sig)
1479	print(f"\n[-] Shutting down ({sig_name})", file=sys.stderr)
1480	admin_stop.set()
1481	audit.server_stop(reason=sig_name)
1482	server.close()
1483	sys.exit(0)
1484
1485	signal.signal(signal.SIGINT, shutdown)
1486	signal.signal(signal.SIGTERM, shutdown)
1487
1488	while True:
1489	try:
1490	conn, addr = server.accept()
1491	thread = threading.Thread(
1492	target=handle_client,
1493	args=(conn, addr, token, audit, registry),
1494	daemon=True,
1495	)
1496	thread.start()
1497	except OSError:
1498	admin_stop.set()
1499	audit.server_stop(reason="socket_error")
1500	break
1501
1502
1503	def connect(endpoint: str, token: str) -> None:
1504	if ":" in endpoint:
1505	host, port_str = endpoint.rsplit(":", 1)
1506	port = int(port_str)
1507	else:
1508	host = endpoint
1509	port = 8765
1510
1511	write_client_info(endpoint, token, os.getpid())
1512
1513	sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
1514
1515	def _disconnect_on_term(signum, frame) -> None:
1516	try:
1517	send_msg(sock, MSG_CLOSE)
1518	except OSError:
1519	pass
1520	clear_client_info()
1521	sys.exit(0)
1522
1523	if hasattr(signal, "SIGTERM"):
1524	signal.signal(signal.SIGTERM, _disconnect_on_term)
1525
1526	try:
1527	sock.connect((host, port))
1528	send_msg(sock, MSG_AUTH, token.encode())
1529
1530	msg_type, data = recv_msg(sock)
1531	if msg_type != MSG_AUTH_OK:
1532	print(f"Authentication failed: {data.decode()}", file=sys.stderr)
1533	sys.exit(1)
1534
1535	session_id = ""
1536	try:
1537	payload = json.loads(data.decode("utf-8"))
1538	if isinstance(payload, dict):
1539	session_id = str(payload.get("session_id") or "")
1540	except json.JSONDecodeError:
1541	pass
1542
1543	write_client_info(endpoint, token, os.getpid(), session_id)
1544	hint = f"Connected to {endpoint}"
1545	if session_id:
1546	hint += f" (session {session_id})"
1547	hint += ". Type exit to disconnect, or run `brt -access leave` from another terminal."
1548	print(hint, file=sys.stderr)
1549
1550	relay_io(sock)
1551	finally:
1552	clear_client_info()
1553	try:
1554	send_msg(sock, MSG_CLOSE)
1555	except OSError:
1556	pass
1557	sock.close()
1558
1559
1560	def main() -> int:
1561	parser = argparse.ArgumentParser()
1562	sub = parser.add_subparsers(dest="mode", required=True)
1563
1564	s = sub.add_parser("serve")
1565	s.add_argument("--host", default="0.0.0.0")
1566	s.add_argument("--port", type=int, default=8765)
1567	s.add_argument("--token", default="")
1568	s.add_argument("--audit-log", default=default_audit_log_path())
1569	s.add_argument("--admin-sock", default=default_admin_sock_path())
1570
1571	c = sub.add_parser("connect")
1572	c.add_argument("endpoint")
1573	c.add_argument("token")
1574
1575	w = sub.add_parser("watch-logs")
1576	w.add_argument("--audit-log", default=default_audit_log_path())
1577	w.add_argument("--admin-sock", default=default_admin_sock_path())
1578	w.add_argument("--no-follow", action="store_true")
1579	w.add_argument("--interactive", action="store_true")
1580	w.add_argument("--server-pid", type=int, default=0)
1581	w.add_argument("--endpoint", default="")
1582	w.add_argument("--token", default="")
1583	w.add_argument("--owns-server", action="store_true")
1584
1585	cl = sub.add_parser("clients")
1586	cl.add_argument("--admin-sock", default=default_admin_sock_path())
1587
1588	d = sub.add_parser("disconnect")
1589	d.add_argument("session_id", nargs="?")
1590	d.add_argument("--admin-sock", default=default_admin_sock_path())
1591
1592	sub.add_parser("leave")
1593
1594	lg = sub.add_parser("clear-logs")
1595	lg.add_argument("--audit-log", default=default_audit_log_path())
1596
1597	args = parser.parse_args()
1598
1599	if args.mode == "serve":
1600	token = args.token or os.urandom(24).hex()
1601	print(f"BRT_ACCESS_TOKEN={token}", flush=True)
1602	print(f"BRT_ACCESS_ENDPOINT=0.0.0.0:{args.port}", flush=True)
1603	serve(args.host, args.port, token, args.audit_log, args.admin_sock)
1604	return 0
1605	if args.mode == "connect":
1606	connect(args.endpoint, args.token)
1607	return 0
1608	if args.mode == "watch-logs":
1609	return watch_logs(
1610	args.audit_log,
1611	args.admin_sock,
1612	follow=not args.no_follow,
1613	interactive=args.interactive,
1614	server_pid=args.server_pid or None,
1615	endpoint=args.endpoint,
1616	token=args.token,
1617	owns_server=args.owns_server,
1618	)
1619	if args.mode == "clients":
1620	return print_clients(args.admin_sock)
1621	if args.mode == "leave":
1622	return leave_client()
1623	if args.mode == "disconnect":
1624	if args.session_id:
1625	return disconnect_client(args.admin_sock, args.session_id)
1626	return leave_client()
1627	if args.mode == "clear-logs":
1628	return clear_serve_logs(args.audit_log)
1629	return 1
1630
1631
1632	if __name__ == "__main__":
1633	sys.exit(main())
1634