win/lib/brtwin.py

14.2 KB · 476 lines · python

1#!/usr/bin/env python3
2"""brtwin — Brew Remote Tools for Windows."""
3
4from __future__ import annotations
5
6import argparse
7import base64
8import hashlib
9import json
10import os
11import re
12import socket
13import ssl
14import subprocess
15import sys
16import uuid
17import urllib.error
18import urllib.parse
19import urllib.request
20from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
21from pathlib import Path
22
23WIN_DIR = Path(__file__).resolve().parent.parent
24LIB_DIR = Path(__file__).resolve().parent
25SHARED = LIB_DIR / "shared"
26STATE = Path.home() / ".brt"
27VERSION_FILE = WIN_DIR / "version"
28EMBED_VERSION = "1.5.12"
29
30
31def read_version() -> str:
32 if VERSION_FILE.exists():
33 return VERSION_FILE.read_text(encoding="utf-8").strip()
34 return EMBED_VERSION
35
36
37def ok(msg: str) -> None:
38 print(f"\033[32m✓\033[0m {msg}")
39
40
41def info(msg: str) -> None:
42 print(f"\033[36m●\033[0m {msg}")
43
44
45def warn(msg: str) -> None:
46 print(f"\033[33m!\033[0m {msg}", file=sys.stderr)
47
48
49def err(msg: str) -> None:
50 print(f"\033[31m✗\033[0m {msg}", file=sys.stderr)
51 sys.exit(1)
52
53
54def load_config() -> dict[str, str]:
55 cfg: dict[str, str] = {}
56 env_file = STATE / ".env"
57 if not env_file.exists():
58 return cfg
59 for line in env_file.read_text(encoding="utf-8").splitlines():
60 line = line.strip()
61 if not line or line.startswith("#") or "=" not in line:
62 continue
63 k, v = line.split("=", 1)
64 cfg[k.strip()] = v.strip().strip('"').strip("'")
65 return cfg
66
67
68def cmd_json(args: argparse.Namespace) -> int:
69 src = sys.stdin.read() if args.file == "-" else Path(args.file).read_text(encoding="utf-8")
70 print(json.dumps(json.loads(src), indent=2, ensure_ascii=False))
71 return 0
72
73
74def cmd_encode(args: argparse.Namespace) -> int:
75 data = Path(args.input).read_bytes() if Path(args.input).is_file() else args.input.encode()
76 print(base64.b64encode(data).decode())
77 return 0
78
79
80def cmd_decode(args: argparse.Namespace) -> int:
81 raw = Path(args.input).read_text(encoding="utf-8") if Path(args.input).is_file() else args.input
82 sys.stdout.buffer.write(base64.b64decode(raw.strip()))
83 return 0
84
85
86def cmd_hash(args: argparse.Namespace) -> int:
87 algo = args.algo.lower()
88 data = Path(args.input).read_bytes() if Path(args.input).is_file() else args.input.encode()
89 if algo == "md5":
90 print(hashlib.md5(data).hexdigest())
91 elif algo == "sha1":
92 print(hashlib.sha1(data).hexdigest())
93 elif algo == "sha256":
94 print(hashlib.sha256(data).hexdigest())
95 elif algo == "sha512":
96 print(hashlib.sha512(data).hexdigest())
97 else:
98 err(f"Unsupported algorithm: {algo}")
99 return 0
100
101
102def cmd_uuid(_: argparse.Namespace) -> int:
103 print(str(uuid.uuid4()))
104 return 0
105
106
107def cmd_urlencode(args: argparse.Namespace) -> int:
108 print(urllib.parse.quote(args.text, safe=""))
109 return 0
110
111
112def cmd_urldecode(args: argparse.Namespace) -> int:
113 print(urllib.parse.unquote(args.text))
114 return 0
115
116
117def cmd_pass(args: argparse.Namespace) -> int:
118 import secrets
119 import string
120
121 alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
122 print("".join(secrets.choice(alphabet) for _ in range(args.length)))
123 return 0
124
125
126def cmd_calc(args: argparse.Namespace) -> int:
127 expr = args.expr.replace(" ", "")
128 if not re.fullmatch(r"[0-9+\-*/().]+", expr):
129 err("Invalid expression")
130 print(eval(expr, {"__builtins__": {}}, {}))
131 return 0
132
133
134def _dns_socket(host: str, rtype: str) -> int:
135 if rtype.upper() == "A":
136 for _, _, _, _, addr in socket.getaddrinfo(host, None, socket.AF_INET):
137 print(addr[0])
138 return 0
139 err(f"Install dnspython for {rtype} lookups, or use A records only")
140 return 1
141
142
143def cmd_dns(args: argparse.Namespace) -> int:
144 try:
145 import dns.resolver # type: ignore
146 except ImportError:
147 return _dns_socket(args.host, args.rtype)
148 answers = dns.resolver.resolve(args.host, args.rtype)
149 for r in answers:
150 print(r)
151 return 0
152
153
154def cmd_dns_safe(args: argparse.Namespace) -> int:
155 return cmd_dns(args)
156
157
158def cmd_headers(args: argparse.Namespace) -> int:
159 req = urllib.request.Request(args.url, method="GET")
160 with urllib.request.urlopen(req, timeout=15) as resp:
161 for k, v in resp.headers.items():
162 print(f"{k}: {v}")
163 return 0
164
165
166def cmd_ssl(args: argparse.Namespace) -> int:
167 ctx = ssl.create_default_context()
168 with socket.create_connection((args.host, args.port), timeout=10) as sock:
169 with ctx.wrap_socket(sock, server_hostname=args.host) as ssock:
170 cert = ssock.getpeercert()
171 print(json.dumps(cert, indent=2))
172 return 0
173
174
175def cmd_ip(args: argparse.Namespace) -> int:
176 if args.mode in ("", "local", "check"):
177 hostname = socket.gethostname()
178 print(f"Host: {hostname}")
179 for info in socket.getaddrinfo(hostname, None):
180 print(info[4][0])
181 if args.mode in ("", "public", "check"):
182 try:
183 with urllib.request.urlopen("https://api.ipify.org", timeout=5) as r:
184 print(f"Public: {r.read().decode().strip()}")
185 except urllib.error.URLError as exc:
186 warn(str(exc))
187 return 0
188
189
190def cmd_ping(args: argparse.Namespace) -> int:
191 rc = subprocess.call(
192 ["ping", "-n", str(args.count), args.host],
193 shell=False,
194 )
195 return rc
196
197
198def cmd_portscan(args: argparse.Namespace) -> int:
199 host = args.host
200 ports = range(1, 1025) if args.ports == "common" else _parse_ports(args.ports)
201 open_ports = []
202 for port in ports:
203 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
204 s.settimeout(0.25)
205 try:
206 if s.connect_ex((host, port)) == 0:
207 open_ports.append(port)
208 finally:
209 s.close()
210 for p in open_ports:
211 print(p)
212 info(f"{len(open_ports)} open ports")
213 return 0
214
215
216def _parse_ports(spec: str) -> range:
217 if "-" in spec:
218 a, b = spec.split("-", 1)
219 return range(int(a), int(b) + 1)
220 return range(int(spec), int(spec) + 1)
221
222
223class _QuietHTTPRequestHandler(SimpleHTTPRequestHandler):
224 def log_message(self, format: str, *args) -> None:
225 pass
226
227
228def cmd_serve(args: argparse.Namespace) -> int:
229 os.chdir(args.directory)
230 server = ThreadingHTTPServer(("0.0.0.0", args.port), _QuietHTTPRequestHandler)
231 info(f"Serving {os.getcwd()} on port {args.port}")
232 try:
233 server.serve_forever()
234 except KeyboardInterrupt:
235 pass
236 return 0
237
238
239def cmd_killport(args: argparse.Namespace) -> int:
240 port = str(args.port)
241 out = subprocess.check_output(["netstat", "-ano"], text=True, errors="replace")
242 pids = set()
243 for line in out.splitlines():
244 if f":{port} " in line and "LISTENING" in line.upper():
245 parts = line.split()
246 if parts:
247 pids.add(parts[-1])
248 if not pids:
249 warn(f"Nothing listening on port {port}")
250 return 0
251 for pid in pids:
252 subprocess.call(["taskkill", "/F", "/PID", pid], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
253 ok(f"Killed PID {pid} on port {port}")
254 return 0
255
256
257def cmd_sysinfo(_: argparse.Namespace) -> int:
258 import platform
259
260 print(f" OS: {platform.system()} {platform.release()} ({platform.machine()})")
261 print(f" Python: {platform.python_version()}")
262 print(f" brtwin: {read_version()}")
263 print(f" Host: {socket.gethostname()}")
264 return 0
265
266
267def cmd_config(args: argparse.Namespace) -> int:
268 STATE.mkdir(parents=True, exist_ok=True)
269 env = STATE / ".env"
270 if args.action == "path":
271 print(env)
272 return 0
273 if args.action == "show":
274 print(env.read_text(encoding="utf-8") if env.exists() else "# empty")
275 return 0
276 if not env.exists():
277 env.write_text("# brtwin config\n", encoding="utf-8")
278 os.environ.setdefault("EDITOR", "notepad")
279 editor = os.environ.get("EDITOR", "notepad")
280 subprocess.call([editor, str(env)])
281 return 0
282
283
284def cmd_browse(args: argparse.Namespace) -> int:
285 browse = SHARED / "browse.py"
286 if not browse.exists():
287 err("browse.py not installed")
288 cfg = load_config()
289 env = os.environ.copy()
290 for k, v in cfg.items():
291 env.setdefault(k, v)
292 return subprocess.call([sys.executable, str(browse), args.query], env=env)
293
294
295def cmd_access(args: argparse.Namespace) -> int:
296 ps1 = LIB_DIR / "access.ps1"
297 if not ps1.exists():
298 err("access.ps1 not found")
299 subcmd = (args.access_cmd or "").strip()
300 if not subcmd:
301 err("Usage: brtwin -access <serve|connect|clients|...>")
302 cmd = [
303 "powershell",
304 "-NoProfile",
305 "-ExecutionPolicy",
306 "Bypass",
307 "-File",
308 str(ps1),
309 "-SubCmd",
310 subcmd,
311 *args.rest,
312 ]
313 return subprocess.call(cmd)
314
315
316def cmd_version(_: argparse.Namespace) -> int:
317 print(f"brtwin {read_version()}")
318 return 0
319
320
321def install_hints() -> tuple[str, str]:
322 base = os.environ.get("BRTWIN_BASE_URL") or os.environ.get("BRT_BASE_URL") or "https://brt.sushii.dev"
323 if sys.platform == "win32":
324 primary = f"irm {base}/win/install.ps1 | iex"
325 secondary = f"macOS / Linux: curl -fsSL {base}/install.sh | sh"
326 else:
327 primary = f"curl -fsSL {base}/install.sh | sh"
328 secondary = f"Windows: irm {base}/win/install.ps1 | iex"
329 return primary, secondary
330
331
332def usage() -> None:
333 ver = read_version()
334 install_primary, install_secondary = install_hints()
335 print(
336 f"""brtwin — Brew Remote Tools for Windows v{ver}
337
338Usage: brtwin -<command> [args...]
339
340Networking
341 -dns <host> [type] DNS lookup (A, AAAA, MX, TXT...)
342 -headers <url> HTTP response headers
343 -ssl <host> [port] SSL certificate info
344 -ip [local|public|check] IP information
345 -ping <host> [-c count] Ping host
346 -portscan <host> [ports] Scan TCP ports
347
348Text & Data
349 -json [file|-] Pretty-print JSON
350 -encode <string|file> Base64 encode
351 -decode <string|file> Base64 decode
352 -hash <algo> <input> Hash (md5, sha1, sha256, sha512)
353 -urlencode <string> URL-encode
354 -urldecode <string> URL-decode
355 -uuid Random UUID
356 -pass [length] Secure password
357 -calc <expression> Calculator
358
359Media & Search
360 -browse "query" Terminal web search
361
362System
363 -serve <port> [directory] Static file server
364 -killport <port> Kill process on port
365 -sysinfo System information
366 -config [show|path] Config file (~/.brt/.env)
367
368Remote Access (compatible with brt on macOS/Linux)
369 -access serve [-p port] Start relay + live console
370 -access connect <host:port> <token>
371 -access clients|disconnect|logs|clear-logs|status|stop|force-stop
372
373 -version Show version
374 -help Show this help
375
376Install: {install_primary}
377{install_secondary}
378 Remote access works across brt and brtwin.
379"""
380 )
381
382
383def main(argv: list[str] | None = None) -> int:
384 argv = list(argv or sys.argv[1:])
385 if not argv or argv[0] in ("-help", "-h", "help"):
386 usage()
387 return 0
388
389 flag = argv[0]
390 rest = argv[1:]
391
392 p = argparse.ArgumentParser(add_help=False)
393 sub: argparse.Namespace | None = None
394
395 if flag == "-json":
396 p.add_argument("file", nargs="?", default="-")
397 sub = p.parse_args(rest)
398 return cmd_json(sub)
399 if flag == "-encode":
400 p.add_argument("input")
401 return cmd_encode(p.parse_args(rest))
402 if flag == "-decode":
403 p.add_argument("input")
404 return cmd_decode(p.parse_args(rest))
405 if flag == "-hash":
406 p.add_argument("algo")
407 p.add_argument("input")
408 return cmd_hash(p.parse_args(rest))
409 if flag == "-uuid":
410 return cmd_uuid(p.parse_args(rest))
411 if flag == "-urlencode":
412 p.add_argument("text")
413 return cmd_urlencode(p.parse_args(rest))
414 if flag == "-urldecode":
415 p.add_argument("text")
416 return cmd_urldecode(p.parse_args(rest))
417 if flag == "-pass":
418 p.add_argument("length", nargs="?", type=int, default=20)
419 return cmd_pass(p.parse_args(rest))
420 if flag == "-calc":
421 p.add_argument("expr")
422 return cmd_calc(p.parse_args(rest))
423 if flag == "-dns":
424 p.add_argument("host")
425 p.add_argument("rtype", nargs="?", default="A")
426 return cmd_dns_safe(p.parse_args(rest))
427 if flag == "-headers":
428 p.add_argument("url")
429 return cmd_headers(p.parse_args(rest))
430 if flag == "-ssl":
431 p.add_argument("host")
432 p.add_argument("port", nargs="?", type=int, default=443)
433 return cmd_ssl(p.parse_args(rest))
434 if flag == "-ip":
435 p.add_argument("mode", nargs="?", default="")
436 return cmd_ip(p.parse_args(rest))
437 if flag == "-ping":
438 p.add_argument("host")
439 p.add_argument("-c", "--count", type=int, default=4)
440 return cmd_ping(p.parse_args(rest))
441 if flag == "-portscan":
442 p.add_argument("host")
443 p.add_argument("ports", nargs="?", default="common")
444 return cmd_portscan(p.parse_args(rest))
445 if flag == "-serve":
446 p.add_argument("port", type=int)
447 p.add_argument("directory", nargs="?", default=".")
448 return cmd_serve(p.parse_args(rest))
449 if flag == "-killport":
450 p.add_argument("port", type=int)
451 return cmd_killport(p.parse_args(rest))
452 if flag == "-sysinfo":
453 return cmd_sysinfo(p.parse_args(rest))
454 if flag == "-config":
455 p.add_argument("action", nargs="?", default="edit")
456 return cmd_config(p.parse_args(rest))
457 if flag in ("-browse", "-search"):
458 p.add_argument("query")
459 return cmd_browse(p.parse_args(rest))
460 if flag == "-access":
461 if not rest:
462 err("Usage: brtwin -access <serve|connect|...>")
463 p.add_argument("access_cmd")
464 p.add_argument("rest", nargs=argparse.REMAINDER)
465 sub = p.parse_args(rest)
466 return cmd_access(sub)
467 if flag in ("-version", "-v"):
468 return cmd_version(p.parse_args(rest))
469
470 err(f"Unknown command: {flag}\nRun 'brtwin -help' for usage.")
471 return 1
472
473
474if __name__ == "__main__":
475 raise SystemExit(main())
476