import os, shutil, html, urllib.parse, pty, select, termios, struct, fcntl, signal, threading, json, time from flask import Flask, request, redirect, send_file, Response from werkzeug.middleware.dispatcher import DispatcherMiddleware from werkzeug.serving import run_simple app = Flask(__name__) app.config["APPLICATION_ROOT"] = "/fm" SECRET_HINTS = ("TOKEN", "KEY", "SECRET", "PASSWORD", "AUTH", "COOKIE") def is_sensitive(name): return any(h in name.upper() for h in SECRET_HINTS) def mask_secrets(text): out = [] for line in text.splitlines(keepends=True): if any(h in line.upper() for h in SECRET_HINTS) and ("=" in line or ":" in line): out.append(line) else: out.append(line) return "".join(out) # ================================================================= # PERSISTENT PTY SHELL with scrollback (reattachable). Starts at "/". # ================================================================= SCROLLBACK_LIMIT = 200_000 # chars of history kept per shell class PtyShell: def __init__(self, name): self.name = name self.created = time.time() self.pid, self.fd = pty.fork() if self.pid == 0: os.environ["TERM"] = "xterm-256color" os.chdir("/") os.execvp("bash", ["bash", "-i"]) else: flags = fcntl.fcntl(self.fd, fcntl.F_GETFL) fcntl.fcntl(self.fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) self._set_size(40, 120) self.history = "" # full scrollback (for reattach) self.pending = {} # per-viewer unread cursor: viewer_id -> index into history self.lock = threading.Lock() threading.Thread(target=self._reader, daemon=True).start() def _set_size(self, rows, cols): try: fcntl.ioctl(self.fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0)) except Exception: pass def _reader(self): while True: try: r, _, _ = select.select([self.fd], [], [], 0.5) if r: data = os.read(self.fd, 65536) if not data: break with self.lock: self.history += data.decode(errors="replace") if len(self.history) > SCROLLBACK_LIMIT: self.history = self.history[-SCROLLBACK_LIMIT:] except OSError: break def write(self, text): try: os.write(self.fd, text.encode()) except OSError: pass def snapshot(self): # full history โ€” used when (re)attaching to replay what happened with self.lock: return mask_secrets(self.history), len(self.history) def read_from(self, cursor): # incremental output since a given cursor position with self.lock: new = self.history[cursor:] return mask_secrets(new), len(self.history) def alive(self): try: pid, _ = os.waitpid(self.pid, os.WNOHANG) return pid == 0 except OSError: return False def kill(self): try: os.kill(self.pid, signal.SIGKILL) except OSError: pass # --- Session manager: named shells that outlive browser tabs --- _shells = {} # name -> PtyShell _counter = {"n": 0} _slock = threading.Lock() def create_shell(): with _slock: _counter["n"] += 1 name = f"Terminal {_counter['n']}" _shells[name] = PtyShell(name) return name def get_shell(name): sh = _shells.get(name) if sh and not sh.alive(): del _shells[name] return None return sh def list_shells(): # prune dead ones, return live session names for n in list(_shells.keys()): if not _shells[n].alive(): del _shells[n] return sorted(_shells.keys(), key=lambda n: _shells[n].created) PAGE = """Box
๐Ÿ“ Files๐Ÿ–ฅ๏ธ Terminals โ™ป๏ธ Reload๐ŸŽฎ GPU test mode ยท ephemeral
{body}""" # ---------------- WINDOWS-11-STYLE TERMINAL WORKSPACE ---------------- @app.route("/term") def term_page(): body = """
Terminals โ€” sessions live on the box and survive closing the browser. Reopen this page anytime to reattach. Paste: Ctrl/Cmd+V or right-click ยท Copy: select then Ctrl/Cmd+C
""" return Response(PAGE.format(body=body)) @app.route("/term_list") def term_list(): return Response(json.dumps({"sessions": list_shells()}), mimetype="application/json") @app.route("/term_new", methods=["POST"]) def term_new(): return Response(json.dumps({"name": create_shell()}), mimetype="application/json") @app.route("/term_close", methods=["POST"]) def term_close(): name = request.get_json(force=True).get("name") sh = _shells.get(name) if sh: sh.kill() # SIGKILL the bash process _shells.pop(name, None) # remove from the session list return Response(json.dumps({"ok": True}), mimetype="application/json") @app.route("/term_snapshot") def term_snapshot(): sh = get_shell(request.args.get("name", "")) if not sh: return Response(json.dumps({"data": "", "cursor": 0, "gone": True}), mimetype="application/json") data, cursor = sh.snapshot() return Response(json.dumps({"data": data, "cursor": cursor}), mimetype="application/json") @app.route("/term_in", methods=["POST"]) def term_in(): d = request.get_json(force=True) sh = get_shell(d.get("name", "")) if sh: sh.write(d.get("data", "")) return Response(json.dumps({"ok": bool(sh)}), mimetype="application/json") @app.route("/term_out") def term_out(): sh = get_shell(request.args.get("name", "")) if not sh: return Response(json.dumps({"gone": True}), mimetype="application/json") cursor = int(request.args.get("cursor", 0)) data, newcur = sh.read_from(cursor) return Response(json.dumps({"data": data, "cursor": newcur}), mimetype="application/json") # ---------------- FILE MANAGER (unchanged) ---------------- @app.route("/") def browse(): path = os.path.abspath(request.args.get("path", "/")) if not os.path.exists(path): return Response(PAGE.format(body=f"

Not found: {html.escape(path)}

"), status=404) if os.path.isfile(path): return redirect(f"/fm/view?path={urllib.parse.quote(path)}") parent = os.path.dirname(path.rstrip("/")) or "/" rows = f'.. (up)' try: entries = sorted(os.listdir(path)) except PermissionError: return Response(PAGE.format(body=f"

Permission denied: {html.escape(path)}

"), status=403) for name in entries: full = os.path.join(path, name); q = urllib.parse.quote(full) try: size = os.path.getsize(full) if os.path.isfile(full) else "" except OSError: size = "?" if os.path.isdir(full): link = f'๐Ÿ“ {html.escape(name)}/' actions = f'delete' else: link = f'๐Ÿ“„ {html.escape(name)}' actions = (f'download | ' f'edit | ' f'delete') rows += f"{link}{size}{actions}" body = f"""
Path: {html.escape(path)}
{rows}
NameSizeActions
""" return Response(PAGE.format(body=body)) @app.route("/view") def view(): path = os.path.abspath(request.args.get("path", "")) if not os.path.isfile(path): return redirect("/fm/") if is_sensitive(os.path.basename(path)): content = "*** masked (sensitive filename) ***" else: try: with open(path, "r", errors="replace") as f: content = mask_secrets(f.read(200000)) except Exception as e: content = f"[cannot read as text: {e}]" q = urllib.parse.quote(path) body = (f'
{html.escape(path)} | ' f'edit | ' f'download | ' f'back
' f"
{html.escape(content)}
") return Response(PAGE.format(body=body)) @app.route("/edit") def edit(): path = os.path.abspath(request.args.get("path", "")) try: with open(path, "r", errors="replace") as f: content = f.read() except Exception as e: content = f"[cannot open: {e}]" body = f"""
Editing: {html.escape(path)}

cancel
""" return Response(PAGE.format(body=body)) @app.route("/save", methods=["POST"]) def save(): path = os.path.abspath(request.form["path"]) with open(path, "w") as f: f.write(request.form["content"]) return redirect(f"/fm/view?path={urllib.parse.quote(path)}") @app.route("/download") def download(): path = os.path.abspath(request.args.get("path", "")) if is_sensitive(os.path.basename(path)): return Response("masked (sensitive filename)", status=403) return send_file(path, as_attachment=True) @app.route("/mkdir", methods=["POST"]) def mkdir(): base = os.path.abspath(request.form["path"]) os.makedirs(os.path.join(base, request.form["name"]), exist_ok=True) return redirect(f"/fm/?path={urllib.parse.quote(base)}") @app.route("/touch", methods=["POST"]) def touch(): base = os.path.abspath(request.form["path"]) name = request.form.get("name", "").strip() if name: full = os.path.join(base, name) if not os.path.exists(full): open(full, "a").close() return redirect(f"/fm/edit?path={urllib.parse.quote(full)}") return redirect(f"/fm/?path={urllib.parse.quote(base)}") @app.route("/upload", methods=["POST"]) def upload(): base = os.path.abspath(request.form["path"]) f = request.files.get("file") if f and f.filename: f.save(os.path.join(base, f.filename)) return redirect(f"/fm/?path={urllib.parse.quote(base)}") @app.route("/delete") def delete(): path = os.path.abspath(request.args.get("path", "")) parent = os.path.dirname(path) try: shutil.rmtree(path) if os.path.isdir(path) else os.remove(path) except Exception as e: return Response(PAGE.format(body=f"

Delete failed: {e}

"), status=500) return redirect(f"/fm/?path={urllib.parse.quote(parent)}") if __name__ == "__main__": wrapped = DispatcherMiddleware(Flask("empty"), {"/fm": app}) run_simple("0.0.0.0", 9001, wrapped, threaded=True)