| 1 | #!/usr/bin/env python3 |
| 2 | """brtwin — Brew Remote Tools for Windows.""" |
| 3 | |
| 4 | from __future__ import annotations |
| 5 | |
| 6 | import argparse |
| 7 | import base64 |
| 8 | import hashlib |
| 9 | import json |
| 10 | import os |
| 11 | import re |
| 12 | import socket |
| 13 | import ssl |
| 14 | import subprocess |
| 15 | import sys |
| 16 | import uuid |
| 17 | import urllib.error |
| 18 | import urllib.parse |
| 19 | import urllib.request |
| 20 | from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer |
| 21 | from pathlib import Path |
| 22 | |
| 23 | WIN_DIR = Path(__file__).resolve().parent.parent |
| 24 | LIB_DIR = Path(__file__).resolve().parent |
| 25 | SHARED = LIB_DIR / "shared" |
| 26 | STATE = Path.home() / ".brt" |
| 27 | VERSION_FILE = WIN_DIR / "version" |
| 28 | EMBED_VERSION = "1.5.12" |
| 29 | |
| 30 | |
| 31 | def read_version() -> str: |
| 32 | if VERSION_FILE.exists(): |
| 33 | return VERSION_FILE.read_text(encoding="utf-8").strip() |
| 34 | return EMBED_VERSION |
| 35 | |
| 36 | |
| 37 | def ok(msg: str) -> None: |
| 38 | print(f"\033[32m✓\033[0m {msg}") |
| 39 | |
| 40 | |
| 41 | def info(msg: str) -> None: |
| 42 | print(f"\033[36m●\033[0m {msg}") |
| 43 | |
| 44 | |
| 45 | def warn(msg: str) -> None: |
| 46 | print(f"\033[33m!\033[0m {msg}", file=sys.stderr) |
| 47 | |
| 48 | |
| 49 | def err(msg: str) -> None: |
| 50 | print(f"\033[31m✗\033[0m {msg}", file=sys.stderr) |
| 51 | sys.exit(1) |
| 52 | |
| 53 | |
| 54 | def 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 | |
| 68 | def 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 | |
| 74 | def 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 | |
| 80 | def 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 | |
| 86 | def 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 | |
| 102 | def cmd_uuid(_: argparse.Namespace) -> int: |
| 103 | print(str(uuid.uuid4())) |
| 104 | return 0 |
| 105 | |
| 106 | |
| 107 | def cmd_urlencode(args: argparse.Namespace) -> int: |
| 108 | print(urllib.parse.quote(args.text, safe="")) |
| 109 | return 0 |
| 110 | |
| 111 | |
| 112 | def cmd_urldecode(args: argparse.Namespace) -> int: |
| 113 | print(urllib.parse.unquote(args.text)) |
| 114 | return 0 |
| 115 | |
| 116 | |
| 117 | def 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 | |
| 126 | def 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 | |
| 134 | def _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 | |
| 143 | def 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 | |
| 154 | def cmd_dns_safe(args: argparse.Namespace) -> int: |
| 155 | return cmd_dns(args) |
| 156 | |
| 157 | |
| 158 | def 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 | |
| 166 | def 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 | |
| 175 | def 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 | |
| 190 | def 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 | |
| 198 | def 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 | |
| 216 | def _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 | |
| 223 | class _QuietHTTPRequestHandler(SimpleHTTPRequestHandler): |
| 224 | def log_message(self, format: str, *args) -> None: |
| 225 | pass |
| 226 | |
| 227 | |
| 228 | def 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 | |
| 239 | def 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 | |
| 257 | def 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 | |
| 267 | def 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 | |
| 284 | def 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 | |
| 295 | def 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 | |
| 316 | def cmd_version(_: argparse.Namespace) -> int: |
| 317 | print(f"brtwin {read_version()}") |
| 318 | return 0 |
| 319 | |
| 320 | |
| 321 | def 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 | |
| 332 | def 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 | |
| 338 | Usage: brtwin -<command> [args...] |
| 339 | |
| 340 | Networking |
| 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 | |
| 348 | Text & 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 | |
| 359 | Media & Search |
| 360 | -browse "query" Terminal web search |
| 361 | |
| 362 | System |
| 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 | |
| 368 | Remote 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 | |
| 376 | Install: {install_primary} |
| 377 | {install_secondary} |
| 378 | Remote access works across brt and brtwin. |
| 379 | """ |
| 380 | ) |
| 381 | |
| 382 | |
| 383 | def 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 | |
| 474 | if __name__ == "__main__": |
| 475 | raise SystemExit(main()) |
| 476 | |