mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 15:41:41 +02:00
jwinterm
4767b7965a
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.
558 lines
18 KiB
Python
558 lines
18 KiB
Python
import contextlib
|
|
import io
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from ampg.approvals import ApprovalInput, approve_artifacts
|
|
from ampg.cli import main
|
|
from ampg.config import load_config
|
|
from ampg.install_plan import (
|
|
blocked_install_steps,
|
|
install_artifacts,
|
|
install_plan,
|
|
install_state_copies,
|
|
install_supervisor_actions,
|
|
write_install_artifacts,
|
|
)
|
|
from ampg.platforms import platform_by_name
|
|
from ampg.status import DaemonProbe
|
|
|
|
|
|
def _probe(*, installed: bool = False, running: bool = False):
|
|
def probe(adapter):
|
|
executable_path = f"/usr/bin/{adapter.executable}" if installed else None
|
|
return DaemonProbe(
|
|
installed=installed,
|
|
running=running,
|
|
executable_path=executable_path,
|
|
)
|
|
|
|
return probe
|
|
|
|
|
|
def _write_config(root: Path, protocol_block: str) -> Path:
|
|
source = root / "site"
|
|
source.mkdir()
|
|
(source / "index.html").write_text("<h1>Hello</h1>", encoding="utf-8")
|
|
config_path = root / "gateway.toml"
|
|
config_path.write_text(
|
|
f"""
|
|
[profiles.mobile-i2p]
|
|
protocols = ["i2p"]
|
|
platform = "android-termux"
|
|
|
|
[[site]]
|
|
id = "example"
|
|
domain = "example.test"
|
|
|
|
[site.source]
|
|
kind = "static-html"
|
|
path = "./site"
|
|
|
|
[site.outputs]
|
|
root = "./out"
|
|
plan_root = "./plan"
|
|
|
|
{protocol_block}
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
return config_path
|
|
|
|
|
|
class InstallPlanTest(unittest.TestCase):
|
|
def test_android_i2p_plan_includes_transport_and_loopback_packages(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config = load_config(
|
|
_write_config(
|
|
Path(tmp),
|
|
"""
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
)
|
|
|
|
steps = install_plan(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
self.assertEqual([], blocked_install_steps(steps))
|
|
package_commands = {
|
|
(step.target, step.command)
|
|
for step in steps
|
|
if step.stage == "package"
|
|
}
|
|
self.assertIn(("i2pd", "pkg install i2pd"), package_commands)
|
|
self.assertIn(("nginx", "pkg install nginx"), package_commands)
|
|
self.assertTrue(any(step.stage == "state" for step in steps))
|
|
self.assertTrue(
|
|
any(
|
|
step.stage == "supervisor"
|
|
and step.command == "sv-enable ampg-example-i2p"
|
|
for step in steps
|
|
)
|
|
)
|
|
|
|
def test_adopted_daemon_has_no_managed_install_steps(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config = load_config(
|
|
_write_config(
|
|
Path(tmp),
|
|
"""
|
|
[site.protocols.tor]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "tor"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
)
|
|
|
|
steps = install_plan(
|
|
config,
|
|
platform_provider=platform_by_name("linux-systemd"),
|
|
daemon_probe=_probe(installed=True, running=True),
|
|
)
|
|
|
|
self.assertEqual(1, len(steps))
|
|
self.assertEqual("adopt-existing", steps[0].action)
|
|
self.assertEqual("skipped", steps[0].status)
|
|
|
|
def test_unknown_platform_blocks_managed_install(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config = load_config(
|
|
_write_config(
|
|
Path(tmp),
|
|
"""
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
)
|
|
|
|
steps = install_plan(
|
|
config,
|
|
platform_provider=platform_by_name("unknown"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
self.assertEqual(1, len(steps))
|
|
self.assertEqual("blocked", steps[0].status)
|
|
self.assertIn("cannot manage", steps[0].message)
|
|
|
|
def test_install_plan_cli_uses_profile_defaults(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config_path = _write_config(
|
|
Path(tmp),
|
|
"""
|
|
[site.protocols.clearnet]
|
|
enabled = true
|
|
renderer = "clearnet"
|
|
daemon = "nginx"
|
|
daemon_policy = "adopt"
|
|
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout), patch("ampg.status.shutil.which", return_value=None):
|
|
status = main(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"install-plan",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
]
|
|
)
|
|
|
|
output = stdout.getvalue()
|
|
|
|
self.assertEqual(0, status)
|
|
self.assertIn("protocol=i2p", output)
|
|
self.assertIn("platform=android-termux", output)
|
|
self.assertIn('command="pkg install i2pd"', output)
|
|
self.assertIn("blocked=0", output)
|
|
self.assertNotIn("protocol=clearnet", output)
|
|
|
|
def test_managed_i2p_artifacts_include_daemon_loopback_and_supervisors(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config = load_config(
|
|
_write_config(
|
|
root,
|
|
"""
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
local_port = 19081
|
|
keys_file = "example-web.dat"
|
|
tunnel_name = "example-web"
|
|
""",
|
|
)
|
|
)
|
|
|
|
artifacts = install_artifacts(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
by_name = {artifact.path.name: artifact for artifact in artifacts}
|
|
self.assertEqual(
|
|
{
|
|
"i2pd-tunnels.conf",
|
|
"i2pd.run",
|
|
"nginx-loopback.conf",
|
|
"nginx-loopback.run",
|
|
},
|
|
set(by_name),
|
|
)
|
|
self.assertIn("[example-web]", by_name["i2pd-tunnels.conf"].content)
|
|
self.assertIn("type = http", by_name["i2pd-tunnels.conf"].content)
|
|
self.assertIn("inport = 80", by_name["i2pd-tunnels.conf"].content)
|
|
self.assertIn("keys = example-web.dat", by_name["i2pd-tunnels.conf"].content)
|
|
self.assertIn("listen 127.0.0.1:19081", by_name["nginx-loopback.conf"].content)
|
|
self.assertIn(f"mkdir -p {root / '.ampg/state/example/i2p'}", by_name["i2pd.run"].content)
|
|
self.assertIn(f"mkdir -p {root / '.ampg/state/example/i2p/data'}", by_name["i2pd.run"].content)
|
|
self.assertIn("exec i2pd", by_name["i2pd.run"].content)
|
|
self.assertIn("exec nginx", by_name["nginx-loopback.run"].content)
|
|
|
|
def test_write_install_artifacts_creates_review_files(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
config = load_config(
|
|
_write_config(
|
|
root,
|
|
"""
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
)
|
|
|
|
artifacts = write_install_artifacts(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
for artifact in artifacts:
|
|
self.assertTrue(artifact.path.exists())
|
|
self.assertTrue(
|
|
(
|
|
root
|
|
/ "plan/example/i2p/managed/android-termux/i2pd-tunnels.conf"
|
|
).exists()
|
|
)
|
|
|
|
def test_state_copy_plan_maps_review_artifacts_to_managed_state(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config = load_config(
|
|
_write_config(
|
|
root,
|
|
"""
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
)
|
|
write_install_artifacts(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
copies = install_state_copies(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
by_kind = {copy.kind: copy for copy in copies}
|
|
self.assertEqual({"daemon-config", "loopback-config"}, set(by_kind))
|
|
self.assertEqual("review", by_kind["daemon-config"].status)
|
|
self.assertIn("not approved", by_kind["daemon-config"].message)
|
|
self.assertEqual(
|
|
root / ".ampg/state/example/i2p/i2pd-tunnels.conf",
|
|
by_kind["daemon-config"].target,
|
|
)
|
|
self.assertEqual(
|
|
root / ".ampg/state/example/i2p/nginx-loopback.conf",
|
|
by_kind["loopback-config"].target,
|
|
)
|
|
self.assertIn("cp", by_kind["daemon-config"].command)
|
|
|
|
def test_supervisor_action_plan_uses_reviewed_supervisor_artifacts(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp).resolve()
|
|
config = load_config(
|
|
_write_config(
|
|
root,
|
|
"""
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
)
|
|
write_install_artifacts(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
actions = install_supervisor_actions(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
by_kind = {action.kind: action for action in actions}
|
|
self.assertEqual({"daemon-supervisor", "loopback-supervisor"}, set(by_kind))
|
|
self.assertEqual("review", by_kind["daemon-supervisor"].status)
|
|
self.assertIn("not approved", by_kind["daemon-supervisor"].message)
|
|
self.assertEqual("ampg-example-i2p-i2pd", by_kind["daemon-supervisor"].service)
|
|
self.assertEqual("sv-enable ampg-example-i2p-i2pd", by_kind["daemon-supervisor"].command)
|
|
self.assertEqual(
|
|
root / "plan/example/i2p/managed/android-termux/i2pd.run",
|
|
by_kind["daemon-supervisor"].source,
|
|
)
|
|
self.assertEqual(
|
|
"sv-enable ampg-example-i2p-nginx-loopback",
|
|
by_kind["loopback-supervisor"].command,
|
|
)
|
|
|
|
def test_approved_managed_artifacts_enable_copy_and_supervisor_plans(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config = load_config(
|
|
_write_config(
|
|
Path(tmp).resolve(),
|
|
"""
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
)
|
|
artifacts = write_install_artifacts(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
approve_artifacts(
|
|
config,
|
|
[
|
|
ApprovalInput(
|
|
site_id=artifact.site_id,
|
|
protocol=artifact.protocol,
|
|
platform=artifact.platform,
|
|
kind=artifact.kind,
|
|
path=artifact.path,
|
|
content=artifact.content,
|
|
)
|
|
for artifact in artifacts
|
|
],
|
|
)
|
|
|
|
copies = install_state_copies(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
actions = install_supervisor_actions(
|
|
config,
|
|
platform_provider=platform_by_name("android-termux"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
self.assertTrue(all(copy.status == "planned" for copy in copies))
|
|
self.assertTrue(all(action.status == "planned" for action in actions))
|
|
|
|
def test_install_plan_cli_can_write_managed_artifacts(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
config_path = _write_config(
|
|
root,
|
|
"""
|
|
[site.protocols.i2p]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "i2pd"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
|
|
stdout = io.StringIO()
|
|
with contextlib.redirect_stdout(stdout):
|
|
status = main(
|
|
[
|
|
"--config",
|
|
str(config_path),
|
|
"install-plan",
|
|
"--profile",
|
|
"mobile-i2p",
|
|
"--write-artifacts",
|
|
]
|
|
)
|
|
|
|
output = stdout.getvalue()
|
|
self.assertEqual(0, status)
|
|
self.assertIn("AMPG_INSTALL_ARTIFACT", output)
|
|
self.assertIn("kind=daemon-config", output)
|
|
self.assertTrue(
|
|
(
|
|
root
|
|
/ "plan/example/i2p/managed/android-termux/i2pd-tunnels.conf"
|
|
).exists()
|
|
)
|
|
|
|
def test_systemd_supervisor_wraps_start_command_in_shell(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config = load_config(
|
|
_write_config(
|
|
Path(tmp),
|
|
"""
|
|
[site.protocols.tor]
|
|
enabled = true
|
|
renderer = "privacy-html"
|
|
daemon = "tor"
|
|
daemon_policy = "auto"
|
|
""",
|
|
)
|
|
)
|
|
|
|
artifacts = install_artifacts(
|
|
config,
|
|
platform_provider=platform_by_name("linux-systemd"),
|
|
daemon_probe=_probe(installed=False, running=False),
|
|
)
|
|
|
|
service = next(artifact for artifact in artifacts if artifact.path.name == "tor.service")
|
|
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()
|