From 4767b7965ab9b1ca15a5f4ad05f9b32cddc9e1cf Mon Sep 17 00:00:00 2001 From: jwinterm Date: Tue, 28 Jul 2026 16:03:40 -0400 Subject: [PATCH] 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. --- ampg/install_plan.py | 76 +++++++++++++++- ampg/renderers.py | 159 ++++++++++++++++++++++++++++++++- tests/test_gemini_build.py | 8 +- tests/test_gemtext_renderer.py | 34 ++++++- tests/test_install_plan.py | 92 +++++++++++++++++++ 5 files changed, 359 insertions(+), 10 deletions(-) diff --git a/ampg/install_plan.py b/ampg/install_plan.py index d9b38df..0cb597e 100644 --- a/ampg/install_plan.py +++ b/ampg/install_plan.py @@ -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 diff --git a/ampg/renderers.py b/ampg/renderers.py index e77460e..74dabd9 100644 --- a/ampg/renderers.py +++ b/ampg/renderers.py @@ -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: diff --git a/tests/test_gemini_build.py b/tests/test_gemini_build.py index c5a1036..1e50f9d 100644 --- a/tests/test_gemini_build.py +++ b/tests/test_gemini_build.py @@ -116,8 +116,12 @@ max_asset_bytes = 1024 self.assertFalse((root / "out/reticulum/app.js").exists()) self.assertFalse((root / "out/reticulum/styles.css").exists()) micron = (root / "out/reticulum/index.mu").read_text(encoding="utf-8") - self.assertIn("# Home", micron) - self.assertIn("=> about.mu About", micron) + # A .mu file must contain micron, not Gemtext. Nomad Network renders + # Gemtext markers as raw punctuation. + self.assertIn(">Home", micron) + self.assertNotIn("# Home", micron) + self.assertIn("`[About`about.mu]", micron) + self.assertNotIn("=> about.mu", micron) self.assertNotIn("bad", micron) diff --git a/tests/test_gemtext_renderer.py b/tests/test_gemtext_renderer.py index f423b13..5742604 100644 --- a/tests/test_gemtext_renderer.py +++ b/tests/test_gemtext_renderer.py @@ -47,11 +47,39 @@ class GemtextRendererTest(unittest.TestCase): rewrite_link=lambda href: href.replace(".html", ".mu"), ) - self.assertIn("# Hello", micron) - self.assertIn("Read the about page.", micron) - self.assertIn("=> about.mu about page", micron) + # Micron, not Gemtext: headings are `>` by depth and links are inline + # as `[label`url], not standalone "=> url label" lines. + self.assertIn(">Hello", micron) + self.assertNotIn("# Hello", micron) + self.assertIn("`[about page`about.mu]", micron) + self.assertNotIn("=> about.mu", micron) self.assertNotIn("ignored", micron) + def test_micron_escapes_content_that_would_become_markup(self): + html = ( + "

>Free Beer

" + "

price is 3 `BTC`

" + "

Sub

" + "
literal `backticks` stay
" + ) + + micron = render_micron(html) + + # A line starting with a structural character must not become a heading. + self.assertIn("\\>Free Beer", micron) + # Backticks in content must not open a formatting run. + self.assertIn("3 \\`BTC\\`", micron) + # Headings and literal blocks still render as real micron. + self.assertIn(">>Sub", micron) + self.assertIn("`=", micron) + self.assertIn("literal `backticks` stay", micron) + + def test_micron_emphasis_uses_toggle_pairs(self): + micron = render_micron("

a b c d

") + + self.assertIn("`!b`!", micron) + self.assertIn("`*d`*", micron) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_install_plan.py b/tests/test_install_plan.py index bf7bddb..9688d70 100644 --- a/tests/test_install_plan.py +++ b/tests/test_install_plan.py @@ -461,5 +461,97 @@ daemon_policy = "auto" self.assertIn("ExecStart=/bin/sh -lc 'exec tor", service.content) +class DaemonConfigInjectionTests(unittest.TestCase): + """torrc and tunnels.conf are line-oriented with no escape mechanism, so an + interpolated newline injects an arbitrary directive.""" + + def test_rejects_newline_in_tunnel_name(self): + from ampg.install_plan import _validate_config_token + + with self.assertRaises(ValueError): + _validate_config_token("web\nhost = attacker.example", "tunnel_name") + + def test_rejects_separator_characters(self): + from ampg.install_plan import _validate_config_token + + for value in ["a b", "../../etc/tor", "name=value", "a\rb", ""]: + with self.subTest(value=value): + with self.assertRaises(ValueError): + _validate_config_token(value, "keys_file") + + def test_accepts_ordinary_identifiers(self): + from ampg.install_plan import _validate_config_token + + self.assertEqual(_validate_config_token("smirk-web.dat", "keys_file"), "smirk-web.dat") + + def test_rejects_line_break_in_state_path(self): + from ampg.install_plan import _validate_config_path + + with self.assertRaises(ValueError): + _validate_config_path(Path("/var/lib/x\nDataDirectory /tmp/evil"), "state directory") + + +class SystemdHardeningTests(unittest.TestCase): + """A unit with no User= runs as whoever installs it, normally root.""" + + def _unit(self, **options) -> str: + import dataclasses + + from ampg.install_plan import _supervisor_content + + config = load_config( + _write_config( + Path(tempfile.mkdtemp()), + """ +[site.protocols.tor] +enabled = true +renderer = "gemtext" +daemon = "tor" +daemon_policy = "auto" +""", + ) + ) + site = config.sites[0] + base = site.protocols["tor"] + protocol = dataclasses.replace(base, options={**base.options, **options}) + return _supervisor_content( + config, + site, + protocol, + platform_by_name("linux-systemd"), + service_suffix="daemon", + command="exec tor -f /etc/tor/torrc", + ) + + def test_unit_warns_loudly_when_no_service_user_configured(self): + unit = self._unit() + + self.assertIn("WARNING: no service_user configured", unit) + self.assertNotIn("\nUser=", unit) + + def test_unit_sets_account_when_service_user_configured(self): + unit = self._unit(service_user="ampg-tor") + + self.assertIn("User=ampg-tor", unit) + self.assertIn("Group=ampg-tor", unit) + self.assertNotIn("WARNING: no service_user", unit) + + def test_unit_carries_baseline_hardening(self): + unit = self._unit(service_user="ampg-tor") + + for directive in ( + "NoNewPrivileges=true", + "PrivateTmp=true", + "RestrictSUIDSGID=true", + "CapabilityBoundingSet=", + ): + with self.subTest(directive=directive): + self.assertIn(directive, unit) + + def test_unit_rejects_injected_service_user(self): + with self.assertRaises(ValueError): + self._unit(service_user="root\nExecStartPre=/bin/evil") + + if __name__ == "__main__": unittest.main()