fix: emit real micron, and validate values interpolated into daemon configs

render_micron was a one-line alias to the Gemtext renderer, so every .mu file
AMPG produced was Gemtext. A Nomad Network browser renders those markers as
raw punctuation. Micron is a different markup: headings are > by depth, links
are inline as `[label`url] rather than standalone lines, emphasis is a toggle
pair, and literal blocks are fenced with `=.

Content is now escaped as well. Micron's control character is the backtick, and
>, <, #, and - are structural at column zero, so a vendor named ">Free Beer"
would have become a section heading on the published page. The line-start check
applies to the emitted line rather than each text chunk, since whether a
character is structural depends on it landing at column zero.

Separately, torrc and tunnels.conf are line-oriented with no quoting or escape
mechanism, so a newline in an interpolated value injects an arbitrary directive:
a tunnel_name of "web\nhost = attacker.example" silently repoints the tunnel.
tunnel_name and keys_file are now validated at the point of interpolation.
_validate_site_id covers the init path only, so a config loaded from disk never
passed through it. There is nothing to escape with here, so these reject rather
than sanitize.

Generated systemd units had no User= and no hardening, so an operator following
the install steps as root got a root-run daemon. Units now carry User=/Group=
when a service account is configured, a visible warning when not, and a baseline
of hardening directives chosen to not break a daemon whose state directory may
live anywhere.

Two tests asserted Gemtext markers inside a .mu file and so had encoded the bug.
This commit is contained in:
jwinterm
2026-07-28 16:03:40 -04:00
parent bb63585098
commit 4767b7965a
5 changed files with 359 additions and 10 deletions
+72 -4
View File
@@ -626,6 +626,43 @@ def _supervisor_start_command(provider: PlatformProvider, service: str) -> str:
return "-"
# Daemon config formats (torrc, i2pd tunnels.conf, systemd units) are
# line-oriented and have NO quoting or escape mechanism, so a single newline in
# an interpolated value injects an arbitrary directive: a `tunnel_name` of
# "web\nhost = attacker.example" silently repoints the tunnel. Because there is
# nothing to escape WITH, these validators reject rather than sanitize.
#
# `_validate_site_id` in onboarding.py covers the `init` path only. A config
# loaded from disk never passes through it, so validation also has to happen
# here, at the point of interpolation.
_CONFIG_TOKEN_ALLOWED = frozenset(
"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-."
)
def _validate_config_token(value: str, field: str) -> str:
"""Return `value` if it is safe to interpolate into a daemon config line."""
if not value:
raise ValueError(f"{field} must not be empty")
if len(value) > 128:
raise ValueError(f"{field} must be 128 characters or fewer")
rejected = sorted({char for char in value if char not in _CONFIG_TOKEN_ALLOWED})
if rejected:
raise ValueError(
f"{field} may only contain letters, numbers, and _-. characters; "
f"rejected {''.join(rejected)!r}"
)
return value
def _validate_config_path(path: Path, field: str) -> Path:
"""Return `path` if it cannot break out of its config line."""
text = str(path)
if "\n" in text or "\r" in text:
raise ValueError(f"{field} must not contain a line break")
return path
def _state_dir(
config: GatewayConfig,
provider: PlatformProvider,
@@ -673,7 +710,9 @@ def _daemon_config_content(
protocol: ProtocolConfig,
provider: PlatformProvider,
) -> str:
state_dir = _state_dir(config, provider, site, protocol)
state_dir = _validate_config_path(
_state_dir(config, provider, site, protocol), "state directory"
)
if protocol.name == "tor":
port = int(protocol.options.get("local_port", 18080))
hidden_service_dir = state_dir / "hidden-service"
@@ -686,8 +725,12 @@ Log notice file {state_dir / "tor.log"}
if protocol.name == "i2p":
port = int(protocol.options.get("local_port", 18081))
inport = int(protocol.options.get("inport", 80))
key_file = protocol.options.get("keys_file", f"{site.id}-web.dat")
tunnel_name = protocol.options.get("tunnel_name", f"{site.id}-web")
key_file = _validate_config_token(
protocol.options.get("keys_file", f"{site.id}-web.dat"), "keys_file"
)
tunnel_name = _validate_config_token(
protocol.options.get("tunnel_name", f"{site.id}-web"), "tunnel_name"
)
return f"""# Generated by AMPG. Review before installing.
[{tunnel_name}]
type = http
@@ -801,6 +844,23 @@ def _supervisor_content(
) -> str:
service_name = _artifact_service_name(site, protocol, service_suffix)
if provider.name == "linux-systemd":
# A unit with no User= runs as whoever installs it, which is normally
# root. Emit User=/Group= when the operator has configured a service
# account, and say so loudly in the generated file otherwise, since the
# whole point of the plan/approve flow is that a human reads this.
service_user = protocol.options.get("service_user")
if service_user:
service_user = _validate_config_token(service_user, "service_user")
service_group = _validate_config_token(
protocol.options.get("service_group", service_user), "service_group"
)
account_lines = f"User={service_user}\nGroup={service_group}\n"
else:
account_lines = (
"# WARNING: no service_user configured, so this unit runs as the\n"
"# user that installs it (normally root). Set `service_user` in the\n"
"# protocol options, or add User=/Group= below before installing.\n"
)
return f"""# Generated by AMPG. Review before installing.
[Unit]
Description=AMPG {protocol.name} {service_suffix} for {site.id}
@@ -809,10 +869,18 @@ Wants=network-online.target
[Service]
Type=simple
WorkingDirectory={_state_dir(config, provider, site, protocol)}
{account_lines}WorkingDirectory={_state_dir(config, provider, site, protocol)}
ExecStart=/bin/sh -lc '{_shell_single_quote(command)}'
Restart=on-failure
RestartSec=5
# Baseline hardening. Kept to directives that cannot break a daemon whose
# state directory may live anywhere (so no ProtectSystem/ProtectHome here);
# tighten per deployment once the state path is known.
NoNewPrivileges=true
PrivateTmp=true
RestrictSUIDSGID=true
RestrictRealtime=true
CapabilityBoundingSet=
[Install]
WantedBy=multi-user.target
+158 -1
View File
@@ -272,8 +272,165 @@ def render_gemtext(html: str, rewrite_link: Callable[[str], str] | None = None)
return GemtextRenderer(rewrite_link=rewrite_link).render(html)
MICRON_STRUCTURAL_PREFIXES = (">", "<", "#", "-")
def micron_escape(text: str) -> str:
"""Escape inline text so micron renders it literally.
Micron's control character is the backtick, so an unescaped one in content
would open a formatting or link run. Escapes with a preceding backslash.
Line-leading structural characters are handled separately at line emission
(see `_escape_line_start`), because whether a character is structural
depends on it landing at column 0, not on which chunk it arrived in.
"""
return text.replace("\\", "\\\\").replace("`", "\\`")
def _escape_line_start(line: str) -> str:
"""Neutralize a structural character at column 0 of an emitted line.
`>` is a heading, `<` resets heading depth, `#` is a comment, and `-` is a
divider. Without this, a vendor named ">Free Beer" silently becomes a
section heading on the published page.
"""
if line[:1] in MICRON_STRUCTURAL_PREFIXES:
return "\\" + line
return line
class MicronRenderer(HTMLParser):
"""Render HTML into micron, the markup Nomad Network nodes serve.
Micron is NOT Gemtext. Headings are `>` / `>>` / `>>>` by depth, links are
inline as ```[label`url]`` rather than standalone lines, emphasis is a
toggle pair, and literal blocks are fenced with ```=``. Reference: the
``TOPIC_MARKUP`` section of NomadNet's in-app guide, which is the only
specification that exists.
Emitting Gemtext into a `.mu` file (which this function used to do) produces
a page that a Nomad Network browser renders as raw punctuation.
"""
def __init__(self, rewrite_link: Callable[[str], str] | None = None) -> None:
super().__init__(convert_charrefs=True)
self.rewrite_link = rewrite_link or (lambda href: href)
self.lines: list[str] = []
self.text_parts: list[str] = []
self.skip_stack: list[str] = []
self.heading_level: int | None = None
self.anchor_stack: list[tuple[str, list[str]]] = []
self.in_pre = False
self.pre_parts: list[str] = []
def render(self, html: str) -> str:
self.feed(html)
self.close()
self._flush_block()
return _squash_blank_lines(self.lines)
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
tag = tag.lower()
attrs_dict = {name.lower(): value for name, value in attrs}
if tag in ACTIVE_TAGS or tag in {"nav", "noscript", "svg"}:
self.skip_stack.append(tag)
return
if self.skip_stack:
return
if tag in {"p", "div", "section", "article", "header", "footer"}:
self._flush_block()
elif tag in {"h1", "h2", "h3", "h4", "h5", "h6"}:
self._flush_block()
self.heading_level = min(int(tag[1]), 3)
elif tag == "li":
self._flush_block()
elif tag == "br":
self._append_text(" ")
elif tag == "hr":
self._flush_block()
self.lines.append("-")
elif tag in {"strong", "b"}:
self.text_parts.append("`!")
elif tag in {"em", "i"}:
self.text_parts.append("`*")
elif tag == "a":
href = attrs_dict.get("href")
if href:
self.anchor_stack.append((href, []))
elif tag == "img":
src = attrs_dict.get("src")
alt = attrs_dict.get("alt") or "image"
if src:
self._emit_link(self.rewrite_link(src), f"Image: {_normalize_text(alt)}")
elif tag == "pre":
self._flush_block()
self.in_pre = True
self.pre_parts = []
def handle_endtag(self, tag: str) -> None:
tag = tag.lower()
if self.skip_stack:
if self.skip_stack[-1] == tag:
self.skip_stack.pop()
return
if tag in {"h1", "h2", "h3", "h4", "h5", "h6"}:
self._flush_block()
self.heading_level = None
elif tag in {"p", "div", "section", "article", "header", "footer", "li"}:
self._flush_block(prefix=" - " if tag == "li" else "")
elif tag in {"strong", "b"}:
self.text_parts.append("`!")
elif tag in {"em", "i"}:
self.text_parts.append("`*")
elif tag == "a" and self.anchor_stack:
href, parts = self.anchor_stack.pop()
label = _normalize_text("".join(parts)) or href
self._emit_link(self.rewrite_link(href), label)
elif tag == "pre" and self.in_pre:
# Literal block: content is emitted verbatim, so it must NOT be escaped.
self.lines.append("`=")
self.lines.extend("".join(self.pre_parts).strip("\n").splitlines())
self.lines.append("`=")
self.lines.append("")
self.in_pre = False
self.pre_parts = []
def handle_data(self, data: str) -> None:
if self.skip_stack:
return
if self.in_pre:
self.pre_parts.append(data)
return
self._append_text(data)
def _append_text(self, data: str) -> None:
if self.anchor_stack:
self.anchor_stack[-1][1].append(data)
return
self.text_parts.append(micron_escape(data))
def _emit_link(self, href: str, label: str) -> None:
# In-page anchors have no meaning on a Nomad node; drop the link, keep the text.
if href.startswith("#"):
self.text_parts.append(micron_escape(label))
return
self.text_parts.append(f"`[{micron_escape(label)}`{micron_escape(href)}]")
def _flush_block(self, prefix: str = "") -> None:
text = _normalize_text("".join(self.text_parts))
if text:
if self.heading_level is not None:
self.lines.append(f"{'>' * self.heading_level}{text}")
elif prefix:
self.lines.append(f"{prefix}{text}")
else:
self.lines.append(_escape_line_start(text))
self.lines.append("")
self.text_parts = []
def render_micron(html: str, rewrite_link: Callable[[str], str] | None = None) -> str:
return GemtextRenderer(rewrite_link=rewrite_link).render(html)
return MicronRenderer(rewrite_link=rewrite_link).render(html)
def _format_attrs(attrs: list[tuple[str, str | None]]) -> str: