randogoth/svarbox
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
svarbox/scripts/dos-httpd

643 lines
20 KiB
Text
Raw Permalink Normal View History

http endpoint
2025-10-26 16:57:22 +02:00
#!/usr/bin/env python3
"""
Serve a lightweight web console backed by /usr/local/bin/dos-shell.
The server exposes:
- GET / → Single-page app with an xterm.js terminal
- GET /ws → WebSocket endpoint that proxies a PTY attached to dos-shell
- GET /healthz → Basic readiness probe
"""
import argparse
import asyncio
import base64
import errno
import fcntl
import json
import logging
import os
import pty
import pwd
import signal
import struct
import termios
from typing import Dict, Optional, Tuple
from aiohttp import web, WSMsgType
LOG = logging.getLogger("dos_httpd")
web feature
2025-10-26 22:21:50 +02:00
def _env_bool(name: str, default: bool) -> bool:
value = os.environ.get(name)
if value is None:
return default
value = value.strip().lower()
if value in {"1", "true", "yes", "on"}:
return True
if value in {"0", "false", "no", "off"}:
return False
return default
http endpoint
2025-10-26 16:57:22 +02:00
HTML_PAGE = """<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8" />
<title>SvarBox Web Console</title>
<meta name="viewport" content="width=device-width, initial-scale=1" />
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/xterm@5.3.0/css/xterm.css" />
<style>
DOS font
2025-10-26 17:02:43 +02:00
@font-face {
font-family: "DOSVGA";
src: url("/assets/WebPlus_IBM_VGA_8x16.woff") format("woff");
font-display: swap;
}
http endpoint
2025-10-26 16:57:22 +02:00
:root, body { background: #1e1e1e; color: #f0f0f0; height: 100%; }
body {
margin: 0;
display: flex;
flex-direction: column;
DOS font
2025-10-26 17:02:43 +02:00
font-family: "DOSVGA", "Fira Code", "Cascadia Mono", "Hack", monospace;
centered
2025-10-26 17:28:14 +02:00
line-height: 1;
http endpoint
2025-10-26 16:57:22 +02:00
}
header {
padding: 0.6rem 1rem;
background: #111;
font-size: 0.95rem;
letter-spacing: 0.04em;
text-transform: uppercase;
}
centered
2025-10-26 17:28:14 +02:00
#workspace {
http endpoint
2025-10-26 16:57:22 +02:00
flex: 1 1 auto;
centered
2025-10-26 17:28:14 +02:00
display: flex;
justify-content: center;
align-items: center;
padding: 32px 24px 40px;
box-sizing: border-box;
}
works
2025-10-26 20:41:23 +02:00
#terminal {
centered
2025-10-26 17:28:14 +02:00
display: inline-flex;
color fix
2025-10-26 21:03:48 +02:00
background: rgb(12 12 12);
centered
2025-10-26 17:28:14 +02:00
border: 1px solid #2b2b2b;
box-shadow: 0 10px 30px rgba(0, 0, 0, 0.45);
padding: 12px;
color fix
2025-10-26 21:03:48 +02:00
transform-origin: center center;
position: relative;
centered
2025-10-26 17:28:14 +02:00
box-sizing: content-box;
CRT
2025-10-26 21:51:57 +02:00
overflow: hidden;
}
.crt-display {
web feature
2025-10-26 22:21:50 +02:00
filter: saturate(1.05) contrast(1.08) brightness(0.95);
CRT
2025-10-26 21:51:57 +02:00
--crt-line-step: 1;
}
.crt-display::before {
content: "";
position: absolute;
inset: 0;
pointer-events: none;
background-image:
repeating-linear-gradient(
to bottom,
web feature
2025-10-26 22:21:50 +02:00
rgba(255, 255, 255, 0.06) 0px,
rgba(255, 255, 255, 0.06) calc(var(--crt-line-step) * 1px),
CRT
2025-10-26 21:51:57 +02:00
rgba(0, 0, 0, 0) calc(var(--crt-line-step) * 1px),
rgba(0, 0, 0, 0) calc(var(--crt-line-step) * 2px)
),
repeating-linear-gradient(
to right,
web feature
2025-10-26 22:21:50 +02:00
rgba(255, 255, 255, 0.025) 0px,
rgba(255, 255, 255, 0.025) calc(var(--crt-line-step) * 1px),
CRT
2025-10-26 21:51:57 +02:00
rgba(0, 0, 0, 0) calc(var(--crt-line-step) * 1px),
rgba(0, 0, 0, 0) calc(var(--crt-line-step) * 2px)
);
mix-blend-mode: screen;
opacity: 0.4;
animation: crt-scan 6s linear infinite;
}
.crt-display::after {
content: "";
position: absolute;
inset fix
2025-10-26 22:40:05 +02:00
inset: -0.1%;
CRT
2025-10-26 21:51:57 +02:00
pointer-events: none;
background: radial-gradient(circle at center, rgba(255, 255, 255, 0.08) 0%, rgba(0, 0, 0, 0.45) 70%, rgba(0, 0, 0, 0.75) 100%);
mix-blend-mode: multiply;
animation: crt-flicker 3s infinite;
}
@keyframes crt-scan {
0% { transform: translateY(-2%); }
100% { transform: translateY(2%); }
}
@keyframes crt-flicker {
0%, 100% { opacity: 0.35; }
45% { opacity: 0.45; }
50% { opacity: 0.3; }
55% { opacity: 0.4; }
90% { opacity: 0.33; }
centered
2025-10-26 17:28:14 +02:00
}
#terminal .xterm {
ypu
2025-10-26 20:41:35 +02:00
width: 100% !important;
height: 100% !important;
centered
2025-10-26 17:28:14 +02:00
}
#terminal .xterm .xterm-helper-textarea {
width: 0 !important;
height: 0 !important;
http endpoint
2025-10-26 16:57:22 +02:00
}
ypu
2025-10-26 20:41:35 +02:00
#terminal .xterm-viewport,
#terminal .xterm-screen,
#terminal .xterm-rows,
#terminal .xterm-scroll-area {
width: 640px !important;
height: 400px !important;
padding: 0 !important;
}
color fix
2025-10-26 21:03:48 +02:00
.xterm-dom-renderer-owner-1 .xterm-bg-0 {
background-color: rgb(14 14 14) !important;
}
http endpoint
2025-10-26 16:57:22 +02:00
</style>
</head>
<body>
<header>SvarBox DOS Console</header>
centered
2025-10-26 17:28:14 +02:00
<main id="workspace">
works
2025-10-26 20:41:23 +02:00
<div id="terminal"></div>
centered
2025-10-26 17:28:14 +02:00
</main>
http endpoint
2025-10-26 16:57:22 +02:00
<script type="module">
import { Terminal } from "https://cdn.jsdelivr.net/npm/xterm@5.3.0/+esm";
web feature
2025-10-26 22:21:50 +02:00
const CRT_ENABLED = __CRT_ENABLED__;
http endpoint
2025-10-26 16:57:22 +02:00
ypu
2025-10-26 20:41:35 +02:00
(async () => {
const COLS = 80;
const ROWS = 25;
const CHAR_HEIGHT = 16;
const VIEWPORT_WIDTH = 640;
const VIEWPORT_HEIGHT = 400;
inset fix
2025-10-26 22:40:05 +02:00
const BASE_VERTICAL_STRETCH = 1.2;
ypu
2025-10-26 20:41:35 +02:00
const TARGET_ROW_HEIGHT = VIEWPORT_HEIGHT / ROWS;
const terminalHost = document.getElementById("terminal");
web feature
2025-10-26 22:21:50 +02:00
if (CRT_ENABLED) {
terminalHost.classList.add("crt-display");
} else {
terminalHost.classList.remove("crt-display");
}
ypu
2025-10-26 20:41:35 +02:00
async function waitForDosvgaFont() {
if (!document.fonts || typeof document.fonts.load !== "function") {
return;
}
try {
await Promise.race([
document.fonts.load(`${CHAR_HEIGHT}px "DOSVGA"`),
new Promise((resolve) => setTimeout(resolve, 2000)),
]);
} catch (err) {
console.debug("DOSVGA font load race rejected", err);
}
http endpoint
2025-10-26 16:57:22 +02:00
}
ypu
2025-10-26 20:41:35 +02:00
function applyViewportClamp() {
terminalHost.style.width = `${VIEWPORT_WIDTH}px`;
terminalHost.style.height = `${VIEWPORT_HEIGHT}px`;
const viewport = terminalHost.querySelector(".xterm-viewport");
if (viewport) {
viewport.style.width = `${VIEWPORT_WIDTH}px`;
viewport.style.height = `${VIEWPORT_HEIGHT}px`;
viewport.style.overflow = "hidden";
}
const scrollArea = terminalHost.querySelector(".xterm-scroll-area");
if (scrollArea) {
scrollArea.style.width = `${VIEWPORT_WIDTH}px`;
scrollArea.style.height = `${VIEWPORT_HEIGHT}px`;
http endpoint
2025-10-26 16:57:22 +02:00
}
ypu
2025-10-26 20:41:35 +02:00
const screen = terminalHost.querySelector(".xterm-screen");
if (screen) {
screen.style.width = `${VIEWPORT_WIDTH}px`;
screen.style.height = `${VIEWPORT_HEIGHT}px`;
}
const rows = terminalHost.querySelector(".xterm-rows");
if (rows) {
rows.style.width = `${VIEWPORT_WIDTH}px`;
rows.style.height = `${VIEWPORT_HEIGHT}px`;
const rowDivs = rows.querySelectorAll("div");
rowDivs.forEach((div) => {
div.style.height = `${TARGET_ROW_HEIGHT}px`;
div.style.lineHeight = `${TARGET_ROW_HEIGHT}px`;
div.style.width = `${VIEWPORT_WIDTH}px`;
});
}
}
color fix
2025-10-26 21:03:48 +02:00
function updateScale() {
const maxWidth = window.innerWidth * 0.85;
const maxHeight = window.innerHeight * 0.85;
// reset transform to measure base stretched size
terminalHost.style.transform = `scaleY(${BASE_VERTICAL_STRETCH})`;
terminalHost.style.transformOrigin = "center center";
const rect = terminalHost.getBoundingClientRect();
const scaleForWidth = maxWidth / rect.width;
const scaleForHeight = maxHeight / rect.height;
const uniformScale = Math.min(scaleForWidth, scaleForHeight);
web feature
2025-10-26 22:21:50 +02:00
if (CRT_ENABLED) {
terminalHost.style.setProperty('--crt-line-step', (1 / uniformScale).toString());
} else {
terminalHost.style.removeProperty('--crt-line-step');
}
color fix
2025-10-26 21:03:48 +02:00
terminalHost.style.transform = `scale(${uniformScale}) scaleY(${BASE_VERTICAL_STRETCH})`;
}
ypu
2025-10-26 20:41:35 +02:00
await waitForDosvgaFont();
const term = new Terminal({
cursorBlink: true,
convertEol: true,
fontFamily: "DOSVGA, Fira Code, Cascadia Mono, Hack, monospace",
fontSize: CHAR_HEIGHT,
lineHeight: 1,
cols: COLS,
rows: ROWS,
theme: {
background: "#000000",
foreground: "#f5f5f5"
},
});
term.open(terminalHost);
term.resize(COLS, ROWS);
applyViewportClamp();
color fix
2025-10-26 21:03:48 +02:00
updateScale();
ypu
2025-10-26 20:41:35 +02:00
const wsProtocol = window.location.protocol === "https:" ? "wss" : "ws";
const socket = new WebSocket(`${wsProtocol}://${window.location.host}/ws`);
const textDecoder = new TextDecoder("utf-8", { fatal: false });
function decodeBase64ToString(b64) {
const binary = atob(b64);
const bytes = new Uint8Array(binary.length);
for (let i = 0; i < binary.length; i += 1) {
bytes[i] = binary.charCodeAt(i);
}
return textDecoder.decode(bytes);
http endpoint
2025-10-26 16:57:22 +02:00
}
ypu
2025-10-26 20:41:35 +02:00
socket.addEventListener("open", () => {
socket.send(JSON.stringify({ type: "resize", cols: term.cols, rows: term.rows }));
term.focus();
applyViewportClamp();
color fix
2025-10-26 21:03:48 +02:00
updateScale();
ypu
2025-10-26 20:41:35 +02:00
});
socket.addEventListener("message", (event) => {
try {
const payload = JSON.parse(event.data);
if (payload.type === "output") {
term.write(decodeBase64ToString(payload.data));
} else if (payload.type === "exit") {
term.write("\\r\\n[dos-shell exited]\\r\\n");
}
} catch (err) {
console.error("failed to process message", err);
}
});
socket.addEventListener("close", (event) => {
term.write(`\\r\\n[connection closed: ${event.code}]\\r\\n`);
});
socket.addEventListener("error", (event) => {
console.error("websocket error", event);
term.write("\\r\\n[websocket error]\\r\\n");
});
term.onData((data) => {
socket.send(JSON.stringify({ type: "input", data }));
});
http endpoint
2025-10-26 16:57:22 +02:00
ypu
2025-10-26 20:41:35 +02:00
term.onResize(({ cols, rows }) => {
socket.send(JSON.stringify({ type: "resize", cols, rows }));
applyViewportClamp();
color fix
2025-10-26 21:03:48 +02:00
updateScale();
ypu
2025-10-26 20:41:35 +02:00
});
http endpoint
2025-10-26 16:57:22 +02:00
ypu
2025-10-26 20:41:35 +02:00
term.onRender(() => {
applyViewportClamp();
color fix
2025-10-26 21:03:48 +02:00
updateScale();
ypu
2025-10-26 20:41:35 +02:00
});
http endpoint
2025-10-26 16:57:22 +02:00
color fix
2025-10-26 21:03:48 +02:00
window.addEventListener("resize", () => {
applyViewportClamp();
updateScale();
}, { passive: true });
ypu
2025-10-26 20:41:35 +02:00
})();
http endpoint
2025-10-26 16:57:22 +02:00
</script>
</body>
</html>
"""
def set_winsize(fd: int, rows: int, cols: int) -> None:
packed = struct.pack("HHHH", rows, cols, 0, 0)
fcntl.ioctl(fd, termios.TIOCSWINSZ, packed)
def spawn_shell(username: str, shell_path: str, env: Optional[Dict[str, str]] = None) -> Tuple[int, int]:
try:
pw_record = pwd.getpwnam(username)
except KeyError as exc:
raise RuntimeError(f"Unable to resolve user '{username}'") from exc
base_env = os.environ.copy()
if env:
base_env.update(env)
base_env["TERM"] = base_env.get("TERM", "xterm-256color")
base_env["HOME"] = pw_record.pw_dir
base_env["LOGNAME"] = username
base_env["USER"] = username
base_env["SHELL"] = shell_path
base_env.setdefault("PWD", pw_record.pw_dir)
pid, master_fd = pty.fork()
if pid == 0:
try:
os.chdir(pw_record.pw_dir)
except FileNotFoundError:
os.chdir("/")
try:
os.setsid()
except OSError:
pass
try:
os.umask(0o22)
except Exception:
pass
if os.getuid() == 0:
try:
os.initgroups(username, pw_record.pw_gid)
except PermissionError:
pass
os.setgid(pw_record.pw_gid)
os.setuid(pw_record.pw_uid)
os.environ.clear()
os.environ.update(base_env)
try:
os.execvpe(shell_path, [shell_path], os.environ)
except Exception as exc:
LOG.error("execvpe failed: %%s", exc, exc_info=True)
os._exit(1) # noqa: SLF001
os.set_blocking(master_fd, False)
return pid, master_fd
class ShellSession:
def __init__(self, username: str, shell_path: str, term_rows: int, term_cols: int):
self.username = username
self.shell_path = shell_path
self.term_rows = term_rows
self.term_cols = term_cols
self.pid: Optional[int] = None
self.fd: Optional[int] = None
self.loop = asyncio.get_running_loop()
self.ws: Optional[web.WebSocketResponse] = None
self._reader_active = False
self._closed = asyncio.Event()
async def start(self, ws: web.WebSocketResponse) -> None:
self.ws = ws
self.pid, self.fd = spawn_shell(self.username, self.shell_path)
set_winsize(self.fd, self.term_rows, self.term_cols)
self.loop.add_reader(self.fd, self._on_pty_ready)
self._reader_active = True
LOG.debug("Spawned dos-shell pid=%s fd=%s", self.pid, self.fd)
def _on_pty_ready(self) -> None:
if self.fd is None or self.ws is None or self.ws.closed:
return
try:
data = os.read(self.fd, 4096)
except OSError as exc:
if exc.errno in (errno.EIO, errno.EBADF):
data = b""
else:
LOG.warning("read error from PTY: exc=%s", exc)
return
if data:
payload = base64.b64encode(data).decode("ascii")
asyncio.create_task(self._send({"type": "output", "data": payload}))
return
LOG.debug("PTY produced zero bytes; closing websocket")
asyncio.create_task(self._send({"type": "exit"}))
asyncio.create_task(self.ws.close())
async def _send(self, message: Dict[str, object]) -> None:
if self.ws is None or self.ws.closed:
return
try:
await self.ws.send_str(json.dumps(message))
except ConnectionResetError:
LOG.debug("websocket reset during send")
except RuntimeError as exc:
LOG.debug("unable to send on websocket: %s", exc)
async def write(self, data: bytes) -> None:
if self.fd is None:
return
try:
await asyncio.to_thread(os.write, self.fd, data)
except OSError as exc:
LOG.debug("write failed: %s", exc)
def resize(self, rows: int, cols: int) -> None:
if self.fd is None:
return
rows = max(rows, 1)
cols = max(cols, 1)
try:
set_winsize(self.fd, rows, cols)
except OSError as exc:
LOG.debug("winsize update failed: %s", exc)
async def close(self) -> None:
if self._closed.is_set():
return
self._closed.set()
if self._reader_active and self.fd is not None:
try:
self.loop.remove_reader(self.fd)
except Exception:
pass
self._reader_active = False
if self.fd is not None:
try:
os.close(self.fd)
except OSError:
pass
self.fd = None
if self.pid:
for sig in (signal.SIGHUP, signal.SIGTERM, signal.SIGKILL):
try:
os.kill(self.pid, sig)
except ProcessLookupError:
break
await asyncio.sleep(0.1)
waited_pid, _ = await asyncio.to_thread(os.waitpid, self.pid, os.WNOHANG)
if waited_pid == self.pid:
break
try:
await asyncio.to_thread(os.waitpid, self.pid, 0)
except ChildProcessError:
pass
self.pid = None
async def index_handler(_request: web.Request) -> web.StreamResponse:
web feature
2025-10-26 22:21:50 +02:00
cfg: "AppConfig" = _request.app["config"]
page = HTML_PAGE.replace("__CRT_ENABLED__", "true" if cfg.crt_effect else "false")
return web.Response(text=page, content_type="text/html", headers={"Cache-Control": "no-store"})
http endpoint
2025-10-26 16:57:22 +02:00
async def health_handler(_request: web.Request) -> web.Response:
return web.Response(text="ok\n", content_type="text/plain")
DOS font
2025-10-26 17:02:43 +02:00
async def font_handler(request: web.Request) -> web.StreamResponse:
cfg: "AppConfig" = request.app["config"]
web feature
2025-10-26 22:21:50 +02:00
if not os.path.exists(cfg.font_path):
DOS font
2025-10-26 17:02:43 +02:00
raise web.HTTPNotFound()
return web.FileResponse(path=cfg.font_path, headers={"Cache-Control": "public, max-age=86400"})
http endpoint
2025-10-26 16:57:22 +02:00
async def websocket_handler(request: web.Request) -> web.WebSocketResponse:
ws = web.WebSocketResponse(heartbeat=30)
await ws.prepare(request)
cfg = request.app["config"]
session = ShellSession(cfg.user, cfg.shell_path, cfg.default_rows, cfg.default_cols)
await session.start(ws)
try:
async for msg in ws:
if msg.type == WSMsgType.TEXT:
try:
payload = json.loads(msg.data)
except json.JSONDecodeError:
continue
if payload.get("type") == "input":
await session.write(payload.get("data", "").encode("utf-8", "ignore"))
elif payload.get("type") == "resize":
rows = int(payload.get("rows", cfg.default_rows))
cols = int(payload.get("cols", cfg.default_cols))
session.resize(rows, cols)
elif msg.type == WSMsgType.ERROR:
LOG.warning("websocket error: %s", ws.exception())
break
finally:
await session.close()
return ws
class AppConfig:
web feature
2025-10-26 22:21:50 +02:00
FONT_PATH = "/usr/local/share/dos-httpd/WebPlus_IBM_VGA_8x16.woff"
def __init__(self, user: str, shell_path: str, host: str, port: int, rows: int, cols: int, crt_effect: bool):
http endpoint
2025-10-26 16:57:22 +02:00
self.user = user
self.shell_path = shell_path
self.host = host
self.port = port
self.default_rows = rows
self.default_cols = cols
web feature
2025-10-26 22:21:50 +02:00
self.font_path = self.FONT_PATH
self.crt_effect = crt_effect
http endpoint
2025-10-26 16:57:22 +02:00
def build_argument_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="SvarBox web console")
parser.add_argument("--host", default=os.environ.get("DOS_HTTP_HOST", "0.0.0.0"), help="Address to bind")
parser.add_argument("--port", type=int, default=int(os.environ.get("DOS_HTTP_PORT", "8080")), help="Port to bind")
parser.add_argument(
"--shell",
default=os.environ.get("DOS_HTTP_SHELL", "/usr/local/bin/dos-shell"),
help="Path to the DOS shell launcher",
)
parser.add_argument(
"--user",
default=os.environ.get("DOS_HTTP_USER", "dosuser"),
help="System user the PTY session should run as",
)
parser.add_argument("--rows", type=int, default=int(os.environ.get("DOS_HTTP_ROWS", "25")), help="Default rows")
parser.add_argument("--cols", type=int, default=int(os.environ.get("DOS_HTTP_COLS", "80")), help="Default columns")
parser.add_argument(
"--log-level",
default=os.environ.get("DOS_HTTP_LOG_LEVEL", "INFO"),
choices=["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"],
help="Log verbosity",
)
web feature
2025-10-26 22:21:50 +02:00
default_crt = _env_bool("DOS_HTTP_CRT", True)
parser.set_defaults(crt_effect=default_crt)
parser.add_argument("--crt", dest="crt_effect", action="store_true", help="Enable CRT overlay")
parser.add_argument("--no-crt", dest="crt_effect", action="store_false", help="Disable CRT overlay")
http endpoint
2025-10-26 16:57:22 +02:00
return parser
def configure_logging(level: str) -> None:
logging.basicConfig(level=getattr(logging, level.upper(), logging.INFO), format="%(asctime)s %(levelname)s %(message)s")
async def create_app(cfg: AppConfig) -> web.Application:
app = web.Application()
app["config"] = cfg
app.add_routes(
[
web.get("/", index_handler),
web.get("/healthz", health_handler),
DOS font
2025-10-26 17:02:43 +02:00
web.get("/assets/WebPlus_IBM_VGA_8x16.woff", font_handler),
http endpoint
2025-10-26 16:57:22 +02:00
web.get("/ws", websocket_handler),
]
)
return app
def main() -> None:
parser = build_argument_parser()
args = parser.parse_args()
configure_logging(args.log_level)
cfg = AppConfig(
user=args.user,
shell_path=args.shell,
host=args.host,
port=args.port,
rows=args.rows,
cols=args.cols,
web feature
2025-10-26 22:21:50 +02:00
crt_effect=args.crt_effect,
http endpoint
2025-10-26 16:57:22 +02:00
)
app = asyncio.run(create_app(cfg))
web.run_app(app, host=cfg.host, port=cfg.port, print=None, access_log=None)
if __name__ == "__main__":
main()
Reference in a new issue Copy permalink