"""md2star local GUI server, Overleaf-style Markdown editor with live PDF preview.
Launches a localhost-only HTTP server that fronts md2star's conversion
engine: the browser sends Markdown + a small JSON options blob;
the server invokes ``convert(fmt, ...)`` from :mod:`md2star.pipeline` against
a temp file and streams the resulting bytes back. The frontend renders
the PDF result inline via PDF.js (no DOCX/PPTX rendering in the browser;
those formats are simply offered as a download).
Endpoints
---------
* ``GET /`` → ``index.html`` (shipped under ``md2star/data/gui/``).
* ``GET /app.js`` → the frontend ES module.
* ``GET /favicon.svg`` → tiny inline SVG, no network calls.
* ``POST /render`` → run the converter, stream back the bytes.
* ``POST /shutdown`` → clean process exit (used by the in-page
"Quit server" button so users do not have
to find the terminal).
The server binds to ``127.0.0.1`` only (no LAN exposure) and uses
``ThreadingHTTPServer`` so a slow ``/render`` does not block fast
``/preview`` reads. There is no auth and no sandbox, it is intended
to run on the same machine as the user invoking it, just like Jupyter
or Vite's dev server.
Security note
-------------
Two different questions, and both have to be answered.
**What may a request touch?** Every ``/fs/*`` path is funnelled through
:func:`_safe_within_root`, which rejects ``..`` segments, absolute paths and
symlink escapes, so a request can never read or write outside the single
folder root the user opened.
**Who is asking?** Binding to loopback keeps the port off the network, but it
does not keep it away from *the user's own browser*: any page they have open
can post to ``127.0.0.1:8765``. A cross-origin POST with a CORS-safelisted
content type is a simple request, so there is no preflight to refuse, and an
attacker never has to read a reply — ``/fs/open`` with ``{"path": "~"}``
followed by ``/fs/delete`` is pure side effect. So every request first passes
:meth:`_Handler._caller_is_this_page`, which checks ``Origin`` (cross-site
calls) and ``Host`` (DNS rebinding) before a handler runs.
Both halves are unit-tested in ``tests/test_gui_security.py``, the second with
a live server and the exact headers a browser would send.
Author
------
`Warith Harchaoui <https://www.linkedin.com/in/warith-harchaoui/>`_
"""
from __future__ import annotations
import json
import os
import re
import shutil
import socket
import subprocess
import sys
import threading
import urllib.parse
import uuid
import webbrowser
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from importlib import resources
from pathlib import Path
from typing import Any
import os_helper as osh
from . import __version__, i18n
from .logging import get_logger
# Module logger — child of the package "md2star" logger. The GUI mostly speaks
# to the user via the browser, but server-side notices (e.g. a twin diagram
# pass degrading to PNGs) belong on stderr where the operator can see them.
logger = get_logger(__name__)
#: Every string in ``index.html`` this server will translate, listed rather
#: than discovered. A declared list is reviewable, is what the ``.po`` is
#: extracted from, and cannot quietly rewrite a class name or a bit of
#: JavaScript the way a clever scan eventually would. Format names, file names
#: and the product's own name are deliberately absent: they are the same in
#: every language. ``tests/test_gui_i18n.py`` checks each one still occurs in
#: the page, so the list cannot rot while the interface moves on.
#: How long the GUI waits for the optional AI lint before answering without
#: it. The CLI has no such limit on purpose: a conversion is not interactive,
#: and a model that takes two minutes to load is still worth waiting for there.
#: A browser is not — found when a resident vision model had to be swapped out
#: and a lint request blocked for over a minute with the editor spinning.
LINT_DEADLINE_S = 20.0
UI_STRINGS: tuple[str, ...] = (
"md2star — Markdown → DOCX / PPTX / PDF",
"Markdown editor",
"PDF preview",
"PDF preview will appear here.",
"Folder browser",
"Open a folder (sidebar will appear)",
"Close the folder browser",
"Create a new .md file",
"Delete selected .md files",
"Force-save the markdown now (auto-save runs every 4 s)",
"Download format",
"Toggle options",
"Light / Dark / Auto",
"AI syntax-lint the Markdown (repairs only, needs Ollama)",
"Activate automatic corrections",
"Custom reference templates",
"No custom templates — using the bundled defaults.",
"Load DOCX template",
"Load PPTX template",
"Bibliography heading",
"(comma separated)",
"(optional override)",
"e.g. Ada Lovelace, Charles Babbage",
"auto (e.g. 21 juin 2026)",
"Authors",
"Bibliography",
"Warnings",
"Render",
"Import",
"Open",
"Save",
"Clear",
"Twin",
"Theme",
"Date",
"Ready",
"and/or",
)
[docs]
def localise(html: str, accept_language: str | None) -> str:
"""Serve the page in the visitor's language, when there is a catalogue.
A browser's ``Accept-Language`` is a person stating a preference, so unlike
the CLI — which answers to ``MD2STAR_LANG`` and nothing else, see
:mod:`md2star.i18n` — the GUI honours it. Strings with no translation keep
their English source, so a partial catalogue gives a partly translated
interface rather than a broken one, and the ``lang`` attribute is updated
with them so a screen reader pronounces what it is actually reading.
"""
language = i18n.negotiate(accept_language)
if not language:
return html
catalogue = i18n.translator(language)
out = html.replace('<html lang="en"', f'<html lang="{language}"')
for source in UI_STRINGS:
translated = catalogue.gettext(source)
if translated != source:
out = out.replace(source, translated)
return out
# Per-server-process cache-busting tag. We splice it into every static
# asset URL in the served index.html so a stale browser is forced to
# re-fetch app.js / styles.css / fonts / vendor bundles on the next
# reload after a `md2star gui` restart. Without this, browsers happily
# serve months-old cached ES modules even though we send no-store —
# their in-memory module map is the culprit, and a query-string is the
# only reliable way to bust it short of versioned filenames.
_SESSION_CACHE_BUST = uuid.uuid4().hex[:8]
# Per-server-process directory where an in-session uploaded custom
# template lives. Lazy-created on first /template POST; wiped on
# /shutdown and (best-effort) at process exit.
_SESSION_DIR: Path | None = None
_SESSION_TEMPLATE_LOCK = threading.Lock()
def _session_dir() -> Path:
"""Return the per-process session tempdir, creating it on demand."""
global _SESSION_DIR
with _SESSION_TEMPLATE_LOCK:
if _SESSION_DIR is None:
_SESSION_DIR = Path(osh.make_temporary_directory(prefix="md2star-gui-session-"))
return _SESSION_DIR
def _session_template(fmt: str) -> Path | None:
"""Return the uploaded template path for *fmt* if one exists."""
if _SESSION_DIR is None or fmt not in ("docx", "pptx"):
return None
cand = _SESSION_DIR / f"template.{fmt}"
return cand if cand.exists() else None
def _session_template_status() -> dict[str, bool]:
"""Snapshot of which session templates are currently uploaded."""
return {
"docx": _session_template("docx") is not None,
"pptx": _session_template("pptx") is not None,
}
# ── Server-side draft auto-save ─────────────────────────────────────
# The editor POSTs to /draft every few seconds (after typing pauses).
# We persist to $XDG_CACHE_HOME/md2star/drafts/last.md so the user can
# come back after a browser crash, a `md2star gui` restart, or even a
# machine reboot and get their content back. This is the "safest"
# tier — localStorage is the in-browser fallback for offline; XDG is
# the durable copy.
from .cache import cache_dir as _cache_dir # noqa: E402 (avoid top-level cycle)
def _draft_path() -> Path:
return _cache_dir("drafts") / "last.md"
# ── Folder-browser session state ─────────────────────────────────────
# The user opens one folder per process; every subsequent /fs/* call is
# confined to its subtree. Set by POST /fs/open, cleared by POST
# /fs/close. The state is process-local — closing the browser tab does
# not auto-close the folder (the user might just be refreshing).
_FOLDER_ROOT: Path | None = None
_FOLDER_LOCK = threading.Lock()
def _set_folder_root(p: Path | None) -> None:
global _FOLDER_ROOT
with _FOLDER_LOCK:
_FOLDER_ROOT = p
def _folder_root() -> Path | None:
return _FOLDER_ROOT
def _safe_within_root(rel: str) -> Path | None:
"""Resolve *rel* against the open folder root, refusing escapes.
Returns the absolute Path on success, ``None`` if no folder is
open or if the resolution leaves the root (``..`` traversal, an
absolute path argument, symlink chicanery).
"""
root = _folder_root()
if root is None:
return None
# Reject absolute paths up front — the contract is "relative to
# the open folder". We check several shapes:
#
# * POSIX-absolute: ``/etc/passwd``
# * UNC / forward-slash root: ``\\server\share`` / ``\\?\C:\``
# * Windows drive-letter: ``C:\\Windows\\...`` (which os.path.isabs
# on POSIX returns False for — explicit check needed for
# defense-in-depth when the server is reached from a Windows
# client even though it's running on a POSIX host).
if rel.startswith("/") or rel.startswith("\\") or os.path.isabs(rel):
return None
if len(rel) >= 2 and rel[1] == ":" and rel[0].isalpha():
return None
candidate = (root / rel).resolve()
try:
candidate.relative_to(root.resolve())
except ValueError:
return None
return candidate
def _safe_stem(name: str) -> str:
"""Reduce a client-supplied filename to a safe bare stem (no path, no ext).
Twin mode writes ``<stem>.md`` directly into the open folder root, so the
stem must not carry directory separators or characters that could steer the
write outside it. We take the basename, drop the extension, and collapse
anything outside ``[word . - space]`` to ``_`` — a bare, root-relative name.
"""
# Normalise Windows separators first so the basename step can't be fooled.
base = Path(name.replace("\\", "/")).name
stem = Path(base).stem
cleaned = re.sub(r"[^\w.\- ]+", "_", stem).strip(" .")
# Cap the length so a pathological filename can't produce an unwieldy path.
return cleaned[:80]
def _applescript_literal(value: str) -> str:
r"""Quote *value* as an AppleScript string literal.
Interpolating a bare string into a script is how a prompt becomes code: a
``"`` ends the literal and whatever follows is AppleScript, which has
``do shell script``. Demonstrated against the un-escaped form — a prompt
of ``x" & (do shell script "…") & "`` ran the command. No caller passes
anything but a constant today, so nothing was reachable; this keeps that
from being the only thing standing between a future localised or
user-supplied prompt and arbitrary execution.
Backslash is escaped before the quote, or the escape we add would itself
be escaped. Newlines go too: a literal cannot span lines.
"""
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
return '"' + " ".join(escaped.split()) + '"'
def _native_folder_picker(prompt: str) -> Path | None:
"""Pop a native folder-chooser dialog. Returns None when unsupported / cancelled.
macOS uses ``osascript`` (built in). Linux tries ``zenity`` then
``kdialog``. Windows uses PowerShell's ``FolderBrowserDialog``.
When no backend is available we return None and the caller falls
back to the in-page text input.
"""
try:
if osh.macos():
# Quoted, never interpolated raw — see _applescript_literal.
script = f"POSIX path of (choose folder with prompt {_applescript_literal(prompt)})"
out = (
subprocess.check_output(
["osascript", "-e", script],
stderr=subprocess.DEVNULL,
timeout=300,
)
.decode("utf-8", errors="replace")
.strip()
)
return Path(out) if out else None
if osh.linux():
if shutil.which("zenity"):
out = (
subprocess.check_output(
["zenity", "--file-selection", "--directory", "--title", prompt],
stderr=subprocess.DEVNULL,
timeout=300,
)
.decode("utf-8", errors="replace")
.strip()
)
return Path(out) if out else None
if shutil.which("kdialog"):
out = (
subprocess.check_output(
["kdialog", "--getexistingdirectory", str(Path.home()), "--title", prompt],
stderr=subprocess.DEVNULL,
timeout=300,
)
.decode("utf-8", errors="replace")
.strip()
)
return Path(out) if out else None
if osh.windows():
# PowerShell one-liner so we don't need a separate .ps1 asset.
cmd = [
"powershell",
"-NoProfile",
"-Command",
"Add-Type -AssemblyName System.Windows.Forms;"
"$d = New-Object System.Windows.Forms.FolderBrowserDialog;"
f"$d.Description = '{prompt}';"
"if ($d.ShowDialog() -eq 'OK') { Write-Output $d.SelectedPath }",
]
out = (
subprocess.check_output(
cmd,
stderr=subprocess.DEVNULL,
timeout=300,
)
.decode("utf-8", errors="replace")
.strip()
)
return Path(out) if out else None
except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
return None
return None
# MIME map for outgoing converted files. The browser uses this to decide
# how to handle the response. HTML is the live-preview path; the others
# are the user-triggered downloads.
_OUTPUT_MIME: dict[str, str] = {
"html": "text/html; charset=utf-8",
"pdf": "application/pdf",
"docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
"pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
}
# Tiny inline favicon so the page does not 404 in the network panel.
# Drawn as a dark-blue document with a small "★" — readable at 16×16.
_FAVICON_SVG = b"""<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 32 32">
<rect x="5" y="3" width="22" height="26" rx="3" fill="#007AFF"/>
<path fill="#FFCC00" d="M16 10l1.8 4.4 4.7.3-3.6 3 1.2 4.5L16 19.8l-4.1 2.4 1.2-4.5-3.6-3 4.7-.3z"/>
</svg>"""
def _read_data(name: str) -> bytes:
"""Return the bytes of a file shipped under ``md2star/data/gui/<name>``."""
return resources.files("md2star.data.gui").joinpath(name).read_bytes()
def _content_type_for_static(path: str) -> str:
"""Map a URL path to an HTTP ``Content-Type`` for the static GUI files."""
if path.endswith(".html"):
return "text/html; charset=utf-8"
if path.endswith((".js", ".mjs")):
return "application/javascript; charset=utf-8"
if path.endswith(".css"):
return "text/css; charset=utf-8"
if path.endswith(".svg"):
return "image/svg+xml"
if path.endswith(".woff2"):
return "font/woff2"
if path.endswith(".woff"):
return "font/woff"
if path.endswith(".json"):
return "application/json"
if path.endswith(".txt"):
return "text/plain; charset=utf-8"
return "application/octet-stream"
def _sniff_ooxml_format(data: bytes) -> str | None:
"""Peek inside an OOXML zip and return "docx", "pptx", or None.
Both formats use the same zip envelope; the discriminator is
whether ``word/`` or ``ppt/`` entries are present in the central
directory. We scan only the first ~64 kB so a malformed body cannot
pin the server.
"""
import io
import zipfile
try:
with zipfile.ZipFile(io.BytesIO(data)) as zf:
names = zf.namelist()
except (zipfile.BadZipFile, OSError):
return None
if any(n.startswith("word/") for n in names):
return "docx"
if any(n.startswith("ppt/") for n in names):
return "pptx"
return None
def _read_vendor(rel_path: str) -> bytes | None:
r"""Read a file under ``md2star/data/gui/vendor/<rel_path>``.
Returns ``None`` for path-traversal attempts or for files that do not
exist. The vendored assets (PDF.js, Tailwind, CodeMirror bundle, fonts)
live there and the GUI references them via ``/vendor/...`` URLs so the
editor works fully offline.
Every segment is vetted, and separators are normalised first, because a
backslash *is* a separator on Windows: splitting on ``"/"`` alone leaves
``..\..\secret`` as one innocent-looking segment that the OS then walks
straight out of this directory. A drive-qualified segment is refused for
the same reason ``_safe_within_root`` refuses one — ``joinpath("C:")``
re-anchors the whole path on Windows. This mirrors the confinement
standard the rest of the module already holds itself to.
"""
if not rel_path:
return None
segments = rel_path.replace("\\", "/").split("/")
if any(seg in ("", ".", "..") or ":" in seg for seg in segments):
return None
rel_path = "/".join(segments)
try:
root = resources.files("md2star.data.gui").joinpath("vendor")
target = root.joinpath(*rel_path.split("/"))
return target.read_bytes()
except (FileNotFoundError, IsADirectoryError, OSError):
return None
#: Host names a loopback-bound server will answer to. A browser reaching
#: ``http://127.0.0.1:8765/`` sends exactly one of these; an attacker domain
#: that has been pointed at 127.0.0.1 (DNS rebinding) sends its own name and
#: is turned away here, before any handler runs.
_LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1", "[::1]"})
#: The address :func:`main` bound to. Only ever leaves ``"127.0.0.1"`` when the
#: operator passed ``--bind`` and read the banner, and that is exactly when the
#: Host allowlist has to stand down: a LAN client legitimately arrives with the
#: machine's own name or address in ``Host``.
_BIND_ADDRESS = "127.0.0.1"
def _host_name(value: str) -> str:
"""The host part of a ``Host``/``netloc`` value, port removed, lowercased.
IPv6 literals keep their brackets (``[::1]:8765`` → ``[::1]``), because
that is the form they are compared against — splitting on the last colon
would otherwise cut an address like ``::1`` in half.
"""
value = value.strip().lower()
if value.startswith("["):
end = value.find("]")
return value[: end + 1] if end != -1 else value
return value.rsplit(":", 1)[0] if ":" in value else value
def _single_line_header(value: str, limit: int = 4000) -> str:
"""Flatten *value* so it can only ever be one header value.
``BaseHTTPRequestHandler.send_header`` writes ``name: value\\r\\n`` without
validating anything, so a newline inside *value* does not get escaped — it
ends the header early and whatever follows is read as the next header, or,
after a blank line, as the response body. Encoding to ASCII does not help:
``\\r`` and ``\\n`` *are* ASCII, so they survive ``errors="replace"``.
Collapsing every run of whitespace to a single space makes the guarantee
structural rather than a question of what the captured text happens to
contain today: the result has no CR, LF or tab left to split on. The cap
keeps one verbose warning from pushing the real headers out of a response.
"""
return " ".join(value.encode("ascii", "replace").decode("ascii").split())[:limit]
class _Handler(BaseHTTPRequestHandler):
"""One request → one response. Threading is handled by the parent server."""
# Silence the default access log to keep the terminal readable when
# the user uses ``md2star gui`` interactively. Errors still print.
def log_message(self, format: str, *args: Any) -> None: # noqa: A002
return
def log_error(self, format: str, *args: Any) -> None: # noqa: A002
sys.stderr.write("md2star gui: " + (format % args) + "\n")
# ─────────────────────────────────────────────────────────────────
# Who is asking — the gate every request passes before any handler
# ─────────────────────────────────────────────────────────────────
def _caller_is_this_page(self) -> str:
"""Return "" when the request may proceed, else why it may not.
Path confinement answers *what* a request may touch; it cannot answer
*who* is asking, and for a server that writes files those are two
different questions. Without this gate, any page the user happens to
have open can drive the whole ``/fs/*`` API: a cross-origin ``POST``
with a CORS-safelisted content type is a *simple request*, so there is
no preflight to decline, and the attacker never needs to read a reply —
``/fs/open`` with ``{"path": "~"}`` followed by ``/fs/delete`` is all
side effect. Two independent checks close it:
**Origin**, when present, must be this very server. Browsers attach it
to every cross-origin request and the page itself cannot forge it, so
a mismatch is conclusive. Absent is allowed on purpose: ``curl`` and
scripted local use send no Origin, and a browser cannot be made to
omit one on a cross-origin call.
**Host** must name the loopback interface while we are bound to it.
This is the DNS-rebinding answer: an attacker domain resolved to
127.0.0.1 is same-origin as far as the browser is concerned — Origin
would match — but it still carries its own name in ``Host``. The check
stands down when the operator bound off-loopback, which is the one
case where a non-loopback Host is the legitimate one (and which the
startup banner already calls a security risk).
"""
host = self.headers.get("Host") or ""
if _BIND_ADDRESS == "127.0.0.1" and _host_name(host) not in _LOOPBACK_HOSTS:
return f"unexpected Host header {host!r}"
origin = (self.headers.get("Origin") or "").strip()
# "null" is what a sandboxed iframe or a file:// page sends. It is not
# this page, and treating it as "absent" would reopen the hole.
if origin and origin.lower() != "null":
parsed = urllib.parse.urlsplit(origin)
if parsed.scheme not in ("http", "https") or parsed.netloc.lower() != host.lower():
return f"cross-origin request from {origin!r}"
elif origin:
return "opaque origin"
return ""
def _reject_foreign_caller(self) -> bool:
"""Send 403 and return ``True`` when this request is not ours to serve."""
why = self._caller_is_this_page()
if not why:
return False
# Logged, not silent: a user whose GUI suddenly refuses a request
# deserves to find out why in the terminal, and a real attempt is
# something they should see.
logger.warning("gui: refused a request — %s", why)
self.send_error(403, f"refused: {why}")
return True
# ─────────────────────────────────────────────────────────────────
# GET — static assets only
# ─────────────────────────────────────────────────────────────────
def do_GET(self) -> None: # noqa: N802
# GET is gated too, not just the mutating verbs: `/fs/read` hands back
# the user's Markdown, and the Host half of the gate is what stops a
# rebound domain from reading it as same-origin.
if self._reject_foreign_caller():
return
# Strip the query string before path matching — every static
# asset has a `?v={{CACHE_BUST}}` appended for browser cache
# busting and `self.path` keeps the raw query string verbatim.
path = self.path.split("?", 1)[0]
if path in ("/", "/index.html"):
# Splice the per-server cache-bust tag into the served HTML
# so a stale browser is forced to re-fetch app.js / styles
# / fonts / vendor bundles on the next reload.
html = _read_data("index.html").replace(
b"{{CACHE_BUST}}", _SESSION_CACHE_BUST.encode("ascii")
)
html = localise(html.decode("utf-8"), self.headers.get("Accept-Language")).encode(
"utf-8"
)
return self._send_bytes(html, "text/html; charset=utf-8")
if path == "/app.js":
return self._send_bytes(_read_data("app.js"), "application/javascript; charset=utf-8")
if path == "/styles.css":
return self._send_bytes(_read_data("styles.css"), "text/css; charset=utf-8")
if path == "/favicon.svg":
return self._send_bytes(_FAVICON_SVG, "image/svg+xml")
if path == "/version":
return self._send_bytes(
json.dumps({"version": __version__}).encode("utf-8"),
"application/json",
)
if path == "/draft":
# Return the most recent on-disk draft so a returning user
# picks up where they left off. 204 when nothing is cached
# (first visit / cleared draft).
target = _draft_path()
if not target.exists():
self.send_response(204)
self.send_header("Content-Length", "0")
self.end_headers()
return
return self._send_bytes(target.read_bytes(), "text/markdown; charset=utf-8")
if path == "/template/status":
return self._send_bytes(
json.dumps({"session_status": _session_template_status()}).encode("utf-8"),
"application/json",
)
if path == "/example":
# Serve the bundled `assets/example.md` (mirrored into
# `md2star/data/example.md` so it ships inside the wheel —
# one level UP from `data/gui/` which is what _read_data
# resolves against). Default editor content when no
# localStorage / server draft is present.
try:
example = resources.files("md2star.data").joinpath("example.md").read_bytes()
return self._send_bytes(example, "text/markdown; charset=utf-8")
except (FileNotFoundError, ModuleNotFoundError, OSError):
self.send_response(204)
self.send_header("Content-Length", "0")
self.end_headers()
return
if path == "/fs/status":
return self._handle_fs_status()
if path == "/fs/list":
return self._handle_fs_list()
if path == "/fs/read":
return self._handle_fs_read()
if path.startswith("/vendor/"):
rel = path[len("/vendor/") :]
data = _read_vendor(rel)
if data is None:
return self.send_error(404, f"unknown vendor asset {rel!r}")
return self._send_bytes(data, _content_type_for_static(rel))
self.send_error(404, f"unknown path {self.path!r}")
# ─────────────────────────────────────────────────────────────────
# POST — actions
# ─────────────────────────────────────────────────────────────────
def do_POST(self) -> None: # noqa: N802
# Before anything is dispatched: is this request even ours? Everything
# below writes files, spawns a converter or stops the process.
if self._reject_foreign_caller():
return
# Every mutating action is a distinct POST path; dispatch on the path
# (query string stripped) to the matching handler. Unknown paths 404
# rather than silently succeeding.
path = self.path.split("?", 1)[0]
if path == "/render":
return self._handle_render()
if path == "/extract":
return self._handle_extract()
if path == "/lint":
return self._handle_lint()
if path == "/template":
return self._handle_template_upload()
if path == "/template/clear":
return self._handle_template_clear()
if path == "/draft":
return self._handle_draft_save()
# ``/fs/*`` is the folder-browser API; each verb is path-confined by
# _safe_within_root inside its handler (see the module security note).
if path == "/fs/open":
return self._handle_fs_open()
if path == "/fs/close":
return self._handle_fs_close()
if path == "/fs/save":
return self._handle_fs_save()
if path == "/fs/create":
return self._handle_fs_create()
if path == "/fs/delete":
return self._handle_fs_delete()
if path == "/shutdown":
return self._handle_shutdown()
self.send_error(404, f"unknown path {self.path!r}")
def _read_json(self) -> dict[str, Any]:
# Read exactly Content-Length bytes (0 → empty body → empty dict). A
# malformed body is a client error (400), not a server crash.
length = int(self.headers.get("Content-Length", "0") or "0")
if length <= 0:
return {}
raw = self.rfile.read(length).decode("utf-8", errors="replace")
try:
return json.loads(raw)
except json.JSONDecodeError as exc:
self.send_error(400, f"invalid JSON: {exc}")
return {}
def _handle_render(self) -> None:
# The browser posts the editor buffer + a format + an options blob;
# default to PDF (the live-preview format) when unspecified.
payload = self._read_json()
markdown: str = payload.get("markdown") or ""
fmt: str = (payload.get("format") or "pdf").lower()
opts: dict[str, Any] = payload.get("options") or {}
if fmt not in _OUTPUT_MIME:
return self.send_error(400, f"unknown format {fmt!r}")
# The engine is imported lazily so a thread that never converts does
# not pay the import cost of pandoc / lua-filter resolution.
from .pipeline import ConvertOptions, convert
# Each render runs in a throwaway folder (auto-removed on exit) so the
# user's disk stays clean and concurrent renders can't collide.
with osh.temporary_folder(prefix="md2star-gui-") as workdir:
md_path = Path(workdir) / "input.md"
out_path = Path(workdir) / f"output.{fmt}"
md_path.write_text(markdown, encoding="utf-8")
# Honor an in-session uploaded reference template
# (POST /template). PDF rendering shares the DOCX template
# because the PDF goes through DOCX → soffice. If absent,
# the engine's own resolver kicks in (source dir → XDG cache
# → deraison.ai → bundled). --bib still not exposed in
# the GUI; users who need bibliographies can run
# md2docx --bib from the CLI instead.
template_fmt = "docx" if fmt in ("docx", "pdf") else "pptx"
uploaded = _session_template(template_fmt)
# Translate the JSON options blob straight into the engine's typed
# options. This used to build a fake argv for the CLI to re-parse;
# a missing key now reads as the dataclass default instead of as an
# absent flag, which is the same thing said once rather than twice.
# Per-phase skips let the UI turn off e.g. mermaid or image embedding.
options = ConvertOptions(
input=str(md_path),
output=str(out_path),
author=str(opts["author"]) if opts.get("author") else None,
lang=str(opts["lang"]) if opts.get("lang") else None,
date=str(opts["date"]) if opts.get("date") else None,
bibliography_name=(
str(opts["bibliography_name"]) if opts.get("bibliography_name") else None
),
lint=bool(opts.get("lint")),
skip_phase=tuple(str(phase) for phase in opts.get("skip_phases") or []),
reference_doc=str(uploaded) if uploaded is not None else None,
)
# Capture the CLI's stdout/stderr so we can surface failures
# back to the browser instead of silently letting them drop
# into the server terminal.
stderr_buf = _StdRedirect(sys.stderr)
rc: int
try:
with stderr_buf:
rc = convert(fmt, options)
except SystemExit as exc: # a nested surface asking to stop
rc = int(exc.code or 1)
except Exception as exc: # noqa: BLE001
rc = 1
stderr_buf.captured += f"\nmd2star gui: unexpected error: {exc}\n"
# Failure (non-zero exit or no output file) → a JSON 500 carrying the
# captured stderr so the editor can show *why* the render failed.
if rc != 0 or not out_path.exists():
err_body = json.dumps(
{
"ok": False,
"exit_code": rc,
"stderr": stderr_buf.captured,
}
).encode("utf-8")
self.send_response(500)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(err_body)))
self.end_headers()
self.wfile.write(err_body)
return
# Success → stream the rendered bytes inline with the right MIME so
# PDF.js can preview a PDF and DOCX/PPTX download cleanly.
data = out_path.read_bytes()
self.send_response(200)
self.send_header("Content-Type", _OUTPUT_MIME[fmt])
self.send_header("Content-Length", str(len(data)))
self.send_header("Cache-Control", "no-store")
self.send_header(
"Content-Disposition",
f'inline; filename="md2star-output.{fmt}"',
)
# Forward any stderr from the CLI so the GUI can show "rendered
# OK, with warnings" — pulling the bytes through a header keeps
# the body a clean PDF/DOCX/PPTX stream.
# Flattened to a single line on the way out: this is free text
# reaching a header, and send_header validates nothing (see
# _single_line_header). The failure path above carries the same
# text in JSON, where the encoder escapes it already.
warned = _single_line_header(stderr_buf.captured)
if warned:
self.send_header("X-Md2star-Stderr", warned)
self.end_headers()
self.wfile.write(data)
# ─────────────────────────────────────────────────────────────────
# POST /template — upload an in-session reference docx
# ─────────────────────────────────────────────────────────────────
def _handle_template_upload(self) -> None:
"""Receive a raw ``.docx`` / ``.pptx`` body and save it as the session template.
The format is taken from the ``X-Md2star-Format`` header (which the
frontend infers from the file extension); we fall back to peeking
inside the OOXML zip if the header is missing or invalid.
"""
# Reject an empty body up front (nothing to save).
length = int(self.headers.get("Content-Length", "0") or "0")
if length <= 0:
return self.send_error(400, "empty upload")
# Cap at 50 MB — bigger than any reasonable Office template and
# protects the server from accidental upload floods.
if length > 50 * 1024 * 1024:
return self.send_error(413, "upload exceeds 50 MB cap")
# Every OOXML file is a zip, so a missing ``PK\x03\x04`` magic means the
# body is not a .docx/.pptx — reject before writing anything to disk.
data = self.rfile.read(length)
if not data.startswith(b"PK\x03\x04"):
return self.send_error(
415, "body does not look like a .docx/.pptx (no OOXML zip header)"
)
# Trust the frontend's format header, but verify by peeking inside the
# zip if it is missing or bogus (docx vs pptx changes where it is used).
fmt = (self.headers.get("X-Md2star-Format") or "").strip().lower()
if fmt not in ("docx", "pptx"):
fmt = _sniff_ooxml_format(data) or ""
if fmt not in ("docx", "pptx"):
return self.send_error(415, "could not determine format (expected .docx or .pptx)")
# Persist under the per-process session dir; the next /render for this
# format picks it up via _session_template.
target = _session_dir() / f"template.{fmt}"
target.write_bytes(data)
body = json.dumps(
{
"ok": True,
"format": fmt,
"bytes": len(data),
"session_status": _session_template_status(),
}
).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _handle_template_clear(self) -> None:
"""Delete every in-session uploaded template."""
fmt = ""
try:
payload = (
self._read_json() if int(self.headers.get("Content-Length", "0") or "0") > 0 else {}
)
fmt = (payload.get("format") or "").lower()
except Exception:
fmt = ""
# Clear just the named format, or both when none was specified.
cleared: list[str] = []
candidates = (fmt,) if fmt in ("docx", "pptx") else ("docx", "pptx")
for f in candidates:
p = _session_template(f)
if p is not None:
p.unlink(missing_ok=True)
cleared.append(f)
body = json.dumps(
{
"ok": True,
"cleared": cleared,
"session_status": _session_template_status(),
}
).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
# ─────────────────────────────────────────────────────────────────
# POST /extract — reverse direction: DOCX/PPTX/PDF → Markdown (Kreuzberg)
# ─────────────────────────────────────────────────────────────────
def _handle_extract(self) -> None:
"""Read an uploaded DOCX/PPTX/PDF body back into Markdown for the editor.
The frontend sends the raw file bytes with an ``X-Md2star-Ext`` header
(``.docx`` / ``.pptx`` / ``.pdf``, inferred from the picked file). Two
modes, selected by ``X-Md2star-Twin``:
* **text-only** (default) — hand the bytes to
:func:`md2star.reverse.to_markdown` and return the recovered Markdown
as JSON. Zero side effects; no folder need be open.
* **twin** (``X-Md2star-Twin: 1``) — call
:func:`md2star.reverse.to_markdown_twin`, which writes ``<stem>.md`` +
an ``assets/`` folder into the *open folder root* so scraped images
survive and their links resolve on re-render. With
``X-Md2star-Diagrams: 1`` and the ``[ai]`` stack present, node-and-edge
figures are re-authored as Mermaid; otherwise every image degrades to a
plain scraped PNG. Requires an open folder (assets need a home).
"""
from .reverse import (
SUPPORTED_REVERSE_EXTENSIONS,
ReverseUnavailable,
to_markdown,
to_markdown_twin,
)
length = int(self.headers.get("Content-Length", "0") or "0")
if length <= 0:
return self.send_error(400, "empty upload")
# Cap at 50 MB — larger than any document this GUI is meant to import.
if length > 50 * 1024 * 1024:
return self.send_error(413, "upload exceeds 50 MB cap")
# The extension decides the reader path *and* becomes part of a filename
# below, so it is matched whole against the supported set rather than
# sniffed with is_supported(), which only inspects the suffix and so
# accepts a value like "../../x.pdf". Nothing escapes today — the
# literal stem in front of it glues the "../" onto a directory that
# does not exist, and the write fails — but that is the prefix saving
# us, not the check, and the failure arrives as an unhandled
# FileNotFoundError. An equality test is correct by construction.
ext = (self.headers.get("X-Md2star-Ext") or "").strip().lower()
if ext and not ext.startswith("."):
ext = "." + ext
if ext not in SUPPORTED_REVERSE_EXTENSIONS:
return self.send_error(415, "expected a .docx, .pptx or .pdf upload")
# A header flag toggles the twin path; anything else is a plain text load.
def _truthy(h: str) -> bool:
return (self.headers.get(h) or "").strip().lower() in ("1", "true", "yes")
twin = _truthy("X-Md2star-Twin")
diagrams = _truthy("X-Md2star-Diagrams")
data = self.rfile.read(length)
# ── text-only path: no assets, no folder needed (v2.10 behaviour) ──
if not twin:
# One throwaway file per import (auto-removed with the folder), so a
# concurrent import can't collide and the disk stays clean.
with osh.temporary_folder(prefix="md2star-gui-extract-") as workdir:
src = Path(workdir) / f"upload{ext}"
src.write_bytes(data)
try:
markdown = to_markdown(src)
except ReverseUnavailable as exc:
# Optional [ocr] extra missing — 501 (not implemented on this
# install) with the exact install hint the exception carries.
return self.send_error(501, str(exc))
except (ValueError, FileNotFoundError) as exc:
return self.send_error(400, str(exc))
except RuntimeError as exc:
return self.send_error(500, str(exc))
return self._json_ok(
{
"ok": True,
"markdown": markdown,
"bytes": len(data),
"ext": ext,
"twin": False,
"assets": 0,
}
)
# ── twin path: scraped assets need a persistent, confined home ──
root = _folder_root()
if root is None:
# Nowhere safe to drop assets/ — steer the user to open a folder, and
# remind them plain Import still works for a text-only load.
return self.send_error(
409,
"Open a folder first (Folder > Open) so the twin's assets/ "
"can be written, or use plain Import for a text-only load.",
)
# Derive a safe, non-clobbering <stem>.md name inside the open folder.
stem = _safe_stem(self.headers.get("X-Md2star-Name") or "") or "twin"
final_stem, n = stem, 1
while (root / f"{final_stem}.md").exists():
final_stem, n = f"{stem}-{n}", n + 1
# Only build the AI handler when diagrams are requested AND the stack is
# live; otherwise fall through with None → every image stays a PNG.
handler = None
if diagrams:
from .reverse_diagrams import diagrams_available, make_diagram_handler
if diagrams_available(None):
handler = make_diagram_handler()
else:
logger.warning(
"md2star gui: --diagrams requested but the AI stack is "
"unavailable; keeping scraped images as PNGs."
)
with osh.temporary_folder(prefix="md2star-gui-extract-") as workdir:
src = Path(workdir) / f"{final_stem}{ext}"
src.write_bytes(data)
try:
md_path = to_markdown_twin(src, root, image_handler=handler)
except ReverseUnavailable as exc:
return self.send_error(501, str(exc))
except (ValueError, FileNotFoundError) as exc:
return self.send_error(400, str(exc))
except RuntimeError as exc:
return self.send_error(500, str(exc))
markdown = md_path.read_text(encoding="utf-8")
assets_dir = root / "assets"
n_assets = sum(1 for _ in assets_dir.iterdir()) if assets_dir.is_dir() else 0
return self._json_ok(
{
"ok": True,
"markdown": markdown,
"bytes": len(data),
"ext": ext,
"twin": True,
"assets": n_assets,
"filename": md_path.name,
}
)
# ─────────────────────────────────────────────────────────────────
# POST /lint — AI syntax-lint the editor's Markdown (Ollama, opt-in)
# ─────────────────────────────────────────────────────────────────
def _handle_lint(self) -> None:
"""Run the AI Markdown syntax-linter over the editor buffer, return the fix.
Mirrors the CLI ``--lint`` pass: :func:`lint_with_llm` sends the Markdown
to a local Ollama model for *syntax-only* repairs and returns the
original untouched on any failure (Ollama absent, model missing, network
error, or a length-guard trip). The response reports whether anything
actually changed so the UI can say "already clean" vs "fixed".
"""
payload = self._read_json()
markdown: str = payload.get("markdown") or ""
model = payload.get("model") or None
# lint_with_llm is self-guarding: it never raises and degrades to the
# input on any failure, so a missing Ollama simply yields "no change".
# What it has no answer for is a backend that is *slow* rather than
# absent — a local daemon holding a different model has to unload it
# and load this one first, which on a laptop is minutes, and the editor
# spins for all of it. The lint is explicitly optional, so waiting past
# a deadline is the one thing this endpoint must not do.
fixed, timed_out = self._lint_within_deadline(markdown, model)
body = json.dumps(
{
"ok": True,
"markdown": fixed,
"changed": fixed != markdown,
# The UI can tell "already clean" from "we gave up waiting",
# which are the same empty diff and very different news.
"timed_out": timed_out,
}
).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _lint_within_deadline(self, markdown: str, model: str | None) -> tuple[str, bool]:
"""Lint, or give up and hand back the buffer. Never blocks past the deadline.
A worker thread rather than a transport timeout, because the deadline
belongs to *this* surface: the CLI waits as long as a model needs (a
conversion is not interactive and a slow load is still worth having),
and only the browser cannot. The thread is left running when it is late
— it is a daemon, it will finish into a result nobody reads, and
killing a model mid-load would be worse than wasting it.
"""
from .preprocessing.lint import lint_with_llm
outcome: list[str] = []
worker = threading.Thread(
target=lambda: outcome.append(lint_with_llm(markdown, model=model)),
daemon=True,
)
worker.start()
worker.join(LINT_DEADLINE_S)
if outcome:
return outcome[0], False
logger.warning(
"gui: the AI lint did not answer within %.0fs; returning the buffer unchanged",
LINT_DEADLINE_S,
)
return markdown, True
# ─────────────────────────────────────────────────────────────────
# POST /draft + GET /draft — server-side editor auto-save
# ─────────────────────────────────────────────────────────────────
def _handle_draft_save(self) -> None:
"""Persist editor markdown to $XDG_CACHE_HOME/md2star/drafts/last.md."""
length = int(self.headers.get("Content-Length", "0") or "0")
if length < 0 or length > 5 * 1024 * 1024:
return self.send_error(413, "draft body too large (5 MB cap)")
data = self.rfile.read(length) if length else b""
# Atomic write so a crash mid-save can't leave a half-written
# file: write to a sibling tempfile, then rename.
target = _draft_path()
tmp = target.with_suffix(".tmp")
tmp.write_bytes(data)
tmp.replace(target)
body = json.dumps(
{
"ok": True,
"bytes": len(data),
"path": str(target),
}
).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
# ─────────────────────────────────────────────────────────────────
# /fs/* — folder browser (open / list / read / save / create /
# delete). Every path is confined to the user-chosen root.
# ─────────────────────────────────────────────────────────────────
def _query(self) -> dict[str, str]:
"""Parse the URL query string into a flat dict."""
q = self.path.split("?", 1)
if len(q) < 2:
return {}
return {
k: (v[0] if v else "")
for k, v in urllib.parse.parse_qs(q[1], keep_blank_values=True).items()
}
def _json_ok(self, payload: dict[str, Any]) -> None:
# Shared 200-JSON responder for the /fs/* verbs; no-store so the browser
# never caches a stale directory listing.
body = json.dumps(payload).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(body)
def _handle_fs_status(self) -> None:
# Report whether a folder is open (and which) so the UI can show the
# browser pane or the "open a folder" prompt.
root = _folder_root()
self._json_ok(
{
"open": root is not None,
"root": str(root) if root else None,
}
)
def _handle_fs_open(self) -> None:
"""Pick a folder (native dialog) or use a caller-supplied path."""
# A caller-supplied path wins (headless / scripted use); otherwise pop a
# native OS folder chooser.
payload = self._read_json()
supplied = (payload.get("path") or "").strip()
if supplied:
target = Path(supplied).expanduser()
if not target.exists() or not target.is_dir():
return self.send_error(404, f"not a directory: {target}")
else:
picked = _native_folder_picker("Pick a folder for md2star")
if picked is None:
return self.send_error(
501, 'no native folder picker available; supply { "path": "…" }'
)
target = picked
_set_folder_root(target.resolve())
return self._json_ok({"ok": True, "root": str(_folder_root())})
def _handle_fs_close(self) -> None:
_set_folder_root(None)
return self._json_ok({"ok": True})
def _handle_fs_list(self) -> None:
"""List one directory level inside the open root."""
root = _folder_root()
if root is None:
return self.send_error(409, "no folder open")
q = self._query()
rel = q.get("path", "")
target = root if rel in ("", ".") else _safe_within_root(rel)
if target is None or not target.exists() or not target.is_dir():
return self.send_error(404, f"not a directory: {rel!r}")
# Sort directories first, then files, each alphabetically — the ordering
# the file-tree UI expects. Paths are returned relative to the root so
# the client never learns the absolute filesystem layout.
dirs, files = [], []
try:
for entry in sorted(
target.iterdir(),
key=lambda e: (not e.is_dir(), e.name.lower()),
):
if entry.name.startswith("."):
continue # skip dot-files; cuts noise
rel_p = entry.relative_to(root).as_posix()
if entry.is_dir():
dirs.append({"name": entry.name, "path": rel_p})
elif entry.is_file():
files.append(
{
"name": entry.name,
"path": rel_p,
"size": entry.stat().st_size,
"is_md": entry.suffix.lower() in (".md", ".markdown"),
}
)
except PermissionError as exc:
return self.send_error(403, f"permission denied: {exc}")
return self._json_ok({"path": rel, "dirs": dirs, "files": files})
def _handle_fs_read(self) -> None:
# Confine the path, then only ever hand back Markdown — the editor has no
# business reading arbitrary binaries out of the opened folder.
q = self._query()
rel = q.get("path", "")
target = _safe_within_root(rel)
if target is None or not target.exists() or not target.is_file():
return self.send_error(404, f"not a file: {rel!r}")
if target.suffix.lower() not in (".md", ".markdown"):
return self.send_error(415, "only .md/.markdown files are readable")
data = target.read_bytes()
self.send_response(200)
self.send_header("Content-Type", "text/markdown; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(data)
def _handle_fs_save(self) -> None:
"""Write the request body to the file at ``path`` inside the open root."""
payload = self._read_json()
rel = (payload.get("path") or "").strip()
content = payload.get("content", "")
if not isinstance(content, str):
return self.send_error(400, "content must be a string")
target = _safe_within_root(rel)
if target is None:
return self.send_error(404, f"path is outside the open folder: {rel!r}")
if target.suffix.lower() not in (".md", ".markdown"):
return self.send_error(415, "only .md/.markdown files can be saved")
# Atomic write: stage to a sibling ``.tmp`` then rename, so a crash
# mid-write never truncates the user's real file.
osh.make_directory(str(target.parent))
tmp = target.with_suffix(target.suffix + ".tmp")
tmp.write_text(content, encoding="utf-8")
tmp.replace(target)
return self._json_ok({"ok": True, "path": rel, "bytes": len(content)})
def _handle_fs_create(self) -> None:
"""Create a new empty (or seeded) .md file at ``path``."""
payload = self._read_json()
rel = (payload.get("path") or "").strip()
seed = payload.get("seed", "")
target = _safe_within_root(rel)
if target is None:
return self.send_error(404, f"path is outside the open folder: {rel!r}")
if target.suffix.lower() not in (".md", ".markdown"):
return self.send_error(415, "filename must end in .md or .markdown")
if target.exists():
return self.send_error(409, f"file already exists: {rel!r}")
osh.make_directory(str(target.parent))
target.write_text(seed if isinstance(seed, str) else "", encoding="utf-8")
return self._json_ok({"ok": True, "path": rel})
def _handle_fs_delete(self) -> None:
"""Delete one or several .md files. Non-md paths are silently skipped."""
payload = self._read_json()
paths = payload.get("paths") or []
if not isinstance(paths, list):
return self.send_error(400, "paths must be a list")
# Delete is best-effort per path: anything that fails the confinement,
# existence, or .md-only checks is reported under ``skipped`` rather than
# aborting the whole batch — so one bad entry can't block the rest.
deleted, skipped = [], []
for raw in paths:
if not isinstance(raw, str):
skipped.append(raw)
continue
target = _safe_within_root(raw)
if target is None or not target.exists() or not target.is_file():
skipped.append(raw)
continue
if target.suffix.lower() not in (".md", ".markdown"):
skipped.append(raw) # safety: never delete non-md from here
continue
try:
target.unlink()
deleted.append(raw)
except OSError:
skipped.append(raw)
return self._json_ok(
{
"ok": True,
"deleted": deleted,
"skipped": skipped,
}
)
def _handle_shutdown(self) -> None:
# We schedule the actual shutdown on a side thread so the response
# can finish first — calling ``server.shutdown()`` from the same
# thread that's handling the request deadlocks ``ThreadingHTTPServer``.
body = b'{"ok":true}\n'
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
threading.Thread(
target=lambda: self.server.shutdown(), # type: ignore[union-attr]
daemon=True,
).start()
# ─────────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────────
def _send_bytes(self, data: bytes, content_type: str) -> None:
self.send_response(200)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(data)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(data)
class _StdRedirect:
"""Tiny context manager that captures ``sys.stderr`` writes into a string.
Used to wrap the synchronous ``convert`` call so any warnings the engine
emits (failed mermaid render, deraison.ai download fallback, …) can be
surfaced to the GUI instead of vanishing into the server terminal.
"""
def __init__(self, original: Any) -> None:
self.original = original
self.captured = ""
def __enter__(self) -> _StdRedirect:
sys.stderr = self # type: ignore[assignment]
return self
def __exit__(self, *_: Any) -> None:
sys.stderr = self.original
def write(self, chunk: str) -> int:
# Pass through to the real stderr so the developer launching the
# server still sees the messages live.
try:
self.original.write(chunk)
except Exception:
pass
self.captured += chunk
return len(chunk)
def flush(self) -> None:
try:
self.original.flush()
except Exception:
pass
# ─────────────────────────────────────────────────────────────────────
# Bind / port allocation
# ─────────────────────────────────────────────────────────────────────
def _pick_port(preferred: int) -> int:
"""Return *preferred* if free, otherwise the next free port the OS allocates."""
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind(("127.0.0.1", preferred))
return preferred
except OSError:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind(("127.0.0.1", 0))
return sock.getsockname()[1]
# ─────────────────────────────────────────────────────────────────────
# Public entry point — ``md2star gui [--port N] [--no-browser]``
# ─────────────────────────────────────────────────────────────────────
[docs]
def serve(*, port: int = 8765, no_browser: bool = False, bind: str = "127.0.0.1") -> int:
"""Serve the editor until Ctrl-C. The body of ``md2star gui``.
The flags arrive parsed: click owns this package's only command line (see
:mod:`md2star.cli`), so there is no argv to re-read here.
Parameters
----------
port : int, optional
Preferred port. On a loopback bind an in-use port falls back to the
next free one; on a non-loopback bind the request is honoured as given,
because an operator exposing a port on purpose means that port.
no_browser : bool, optional
Do not auto-open a browser. The URL is still printed.
bind : str, optional
Bind address. ``0.0.0.0`` exposes the server to the LAN with no
authentication, which is why it prints the banner below.
Returns
-------
int
``0`` once the server has been stopped cleanly.
"""
# Tell the request gate where we actually bound, so it knows whether the
# loopback Host allowlist applies (see _caller_is_this_page).
global _BIND_ADDRESS
_BIND_ADDRESS = bind
chosen = _pick_port(port) if bind == "127.0.0.1" else port
server = ThreadingHTTPServer((bind, chosen), _Handler)
url = f"http://{bind}:{chosen}/"
print(f"md2star {__version__} — GUI listening on {url}")
if bind != "127.0.0.1":
# LAN / 0.0.0.0 binding is a footgun. Loud multi-line banner so
# the operator cannot miss it. No prompt — we don't want to
# block scripted deployments — but the message names the exact
# risks so anyone reading their logs sees the issue.
banner = "═" * 64
print(
"\n"
+ banner
+ "\n ⚠ md2star GUI is bound to a NON-LOOPBACK address."
+ f"\n URL: {url}"
+ "\n ⚠ The server has NO AUTHENTICATION and runs the md2star CLI"
+ "\n against any markdown a client sends. Anyone who can reach"
+ "\n this port can write files inside the open folder root and"
+ "\n upload arbitrary DOCX/PPTX templates."
+ "\n ⚠ This is appropriate ONLY for trusted local-network use."
+ "\n For anything else, bind to 127.0.0.1 (the default) and"
+ "\n use SSH port-forwarding to reach the GUI remotely."
+ "\n"
+ banner
+ "\n",
file=sys.stderr,
)
print("Press Ctrl-C to stop (or use the in-page Quit button).")
if not no_browser:
threading.Timer(0.6, lambda: webbrowser.open(url)).start()
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nmd2star gui: stopped.")
finally:
server.server_close()
return 0
if __name__ == "__main__":
sys.exit(serve())