github-such-software/adaptive-multi-protocol-gateway
1
0
Fork 0
mirror of https://github.com/Such-Software/adaptive-multi-protocol-gateway.git synced 2026-10-17 07:31:32 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add site onboarding config generator

Browse Source
This commit is contained in:
jwinterm
2026-07-06 17:55:19 -04:00
parent 0426ca673c
commit db5d9737c0
6 changed files with 557 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+12 -10
View File
@@ -45,21 +45,22 @@ The default for selected non-clearnet protocols is `auto`.
## Target workflow
```sh
ampg init site wownero \
python3 -m ampg --config gateway.toml init site wownero \
--domain wownero.org \
--source ../wownero.org-website \
--source-kind static-html
--preset full
ampg enable wownero tor gemini i2p reticulum
ampg plan
ampg build
ampg apply --dry-run
python3 -m ampg --config gateway.toml build
python3 -m ampg --config gateway.toml doctor
python3 -m ampg --config gateway.toml install-plan --profile vps-full --write-artifacts
python3 -m ampg --config gateway.toml approvals list --profile vps-full
python3 -m ampg --config gateway.toml apply --dry-run --profile vps-full
```
`plan` prints rendered output paths, daemon decisions, port bindings, config diffs, and
operator actions. The current `apply --dry-run` prints the activation sequence and a
preflight verdict without changing services. Live apply support will perform only
approved changes.
`init site` writes a readable `gateway.toml` with selected transports enabled and common
transports left as one-line toggles. The current `apply --dry-run` prints the activation
sequence and a preflight verdict without changing services. Live apply support will
perform only approved changes.
## Interaction tiers
@@ -102,6 +103,7 @@ python3 -m ampg docs generate --check
The current implementation is dependency-free Python:
```sh
python3 -m ampg --config gateway.toml init site wownero --domain wownero.org --source ../wownero.org-website --preset full
python3 -m ampg --config examples/wownero.gateway.toml plan
python3 -m ampg --config examples/wownero.gateway.toml plan --write-artifacts
python3 -m ampg --config examples/wownero.gateway.toml status
ampg/cli.py
+96
View File
@@ -49,6 +49,7 @@ from .install_plan import (
write_install_artifacts,
)
from .manifest import write_fixture_manifests
from .onboarding import PRESETS, init_site_config, parse_protocol_filters
from .plan import plan_artifacts, plan_gateway, write_plan_artifacts
from .platforms import PLATFORM_NAMES, platform_by_name
from .preview import PreviewServers, preview_endpoints, write_preview_fixture_manifests
@@ -72,6 +73,51 @@ def main(argv: list[str] | None = None) -> int:
help="Path to gateway TOML config.",
)
subcommands = parser.add_subparsers(dest="command", required=True)
init_parser = subcommands.add_parser(
"init",
help="Create a gateway config for an existing site.",
)
init_subcommands = init_parser.add_subparsers(dest="init_command", required=True)
init_site = init_subcommands.add_parser(
"site",
help="Create gateway.toml for an existing static HTML site.",
)
init_site.add_argument("site_id", help="Stable site id used in output/state paths.")
init_site.add_argument("--domain", required=True, help="Canonical clearnet domain.")
init_site.add_argument("--source", required=True, type=Path, help="Existing site source tree.")
init_site.add_argument(
"--source-kind",
default="static-html",
choices=("static-html",),
help="Source adapter kind.",
)
init_site.add_argument(
"--preset",
choices=sorted(PRESETS),
default="full",
help="Transport preset to enable when --protocol is omitted.",
)
init_site.add_argument(
"--protocol",
action="append",
default=[],
help="Protocol to enable. May be repeated or comma-separated.",
)
init_site.add_argument(
"--output-root",
type=Path,
help="Generated output root. Defaults to ./dist/<site>.",
)
init_site.add_argument(
"--plan-root",
type=Path,
help="Generated review artifact root. Defaults to ./dist/ampg-plan.",
)
init_site.add_argument(
"--force",
action="store_true",
help="Replace an existing config file.",
)
plan_parser = subcommands.add_parser("plan", help="Print a dry-run plan.")
_add_target_selection(plan_parser)
plan_parser.add_argument(
@@ -316,6 +362,8 @@ def main(argv: list[str] | None = None) -> int:
args = parser.parse_args(argv)
try:
if args.command == "init":
return _cmd_init(args)
if args.command == "docs":
return _cmd_docs(args)
if args.command == "route-manifest":
@@ -355,6 +403,54 @@ def main(argv: list[str] | None = None) -> int:
return 1
def _cmd_init(args) -> int:
if args.init_command == "site":
result = init_site_config(
config_path=args.config,
site_id=args.site_id,
domain=args.domain,
source=args.source,
source_kind=args.source_kind,
preset=args.preset,
protocols=parse_protocol_filters(args.protocol),
output_root=args.output_root,
plan_root=args.plan_root,
force=args.force,
)
protocol_text = ",".join(result.protocols)
print(
"AMPG_INIT_SITE "
f"status=written "
f"path=\"{_quote(str(result.config_path))}\" "
f"site={result.site_id} "
f"domain={result.domain} "
f"protocols={protocol_text}"
)
for profile in result.profiles:
print(f"AMPG_INIT_PROFILE name={profile}")
for protocol in ("clearnet", "tor", "i2p", "gemini"):
enabled = protocol in result.protocols
print(
"AMPG_INIT_TOGGLE "
f"protocol={protocol} "
f"enabled={_bool(enabled)} "
f"field=\"[site.protocols.{protocol}].enabled\""
)
config_arg = str(result.config_path)
profile_arg = result.profiles[0] if result.profiles else "-"
next_commands = [
f"python3 -m ampg --config {config_arg} build",
f"python3 -m ampg --config {config_arg} doctor",
f"python3 -m ampg --config {config_arg} install-plan --profile {profile_arg} --write-artifacts",
f"python3 -m ampg --config {config_arg} approvals list --profile {profile_arg}",
f"python3 -m ampg --config {config_arg} apply --dry-run --profile {profile_arg}",
]
for index, command in enumerate(next_commands, start=1):
print(f"AMPG_INIT_NEXT step={index} command=\"{_quote(command)}\"")
return 0
return 1
def _cmd_apply(config, args) -> int:
if not args.dry_run:
print(
ampg/onboarding.py
+277
View File
@@ -0,0 +1,277 @@
from __future__ import annotations
from dataclasses import dataclass
import json
import os
from pathlib import Path
COMMON_PROTOCOLS = ("clearnet", "tor", "i2p", "gemini")
PRESETS = {
"full": ("clearnet", "tor", "i2p"),
"privacy": ("tor", "i2p"),
"i2p-only": ("i2p",),
"tor-only": ("tor",),
"clearnet-only": ("clearnet",),
"gemini-only": ("gemini",),
}
@dataclass(frozen=True)
class InitSiteResult:
config_path: Path
site_id: str
domain: str
protocols: tuple[str, ...]
profiles: tuple[str, ...]
def init_site_config(
*,
config_path: Path,
site_id: str,
domain: str,
source: Path,
source_kind: str,
preset: str,
protocols: tuple[str, ...],
output_root: Path | None = None,
plan_root: Path | None = None,
force: bool = False,
) -> InitSiteResult:
if source_kind != "static-html":
raise ValueError("init currently supports --source-kind static-html")
selected_protocols = _selected_protocols(preset=preset, protocols=protocols)
_validate_site_id(site_id)
resolved_config = config_path.expanduser().resolve()
if resolved_config.exists() and not force:
raise FileExistsError(f"config already exists: {resolved_config}; pass --force to replace")
resolved_source = source.expanduser().resolve()
if not resolved_source.exists():
raise FileNotFoundError(f"source path does not exist: {resolved_source}")
if not resolved_source.is_dir():
raise NotADirectoryError(f"source path is not a directory: {resolved_source}")
base_dir = resolved_config.parent
resolved_output_root = (output_root or Path(f"./dist/{site_id}")).expanduser()
resolved_plan_root = (plan_root or Path("./dist/ampg-plan")).expanduser()
content = render_site_config(
site_id=site_id,
domain=domain,
source_path=_config_path_text(resolved_source, base_dir),
source_kind=source_kind,
output_root=_config_path_text(_resolve_for_config(resolved_output_root, base_dir), base_dir),
plan_root=_config_path_text(_resolve_for_config(resolved_plan_root, base_dir), base_dir),
selected_protocols=selected_protocols,
)
resolved_config.parent.mkdir(parents=True, exist_ok=True)
resolved_config.write_text(content, encoding="utf-8")
return InitSiteResult(
config_path=resolved_config,
site_id=site_id,
domain=domain,
protocols=selected_protocols,
profiles=_profile_names(selected_protocols),
)
def render_site_config(
*,
site_id: str,
domain: str,
source_path: str,
source_kind: str,
output_root: str,
plan_root: str,
selected_protocols: tuple[str, ...],
) -> str:
lines = [
"# Generated by AMPG. Toggle transports with enabled = true/false.",
"",
"[gateway]",
'state_dir = "./.ampg/state"',
'cache_dir = "./.ampg/cache"',
'run_dir = "./.ampg/run"',
'user = "ampg"',
"",
]
lines.extend(_profile_sections(selected_protocols))
lines.extend(
[
"[[site]]",
f"id = {_toml_string(site_id)}",
f"domain = {_toml_string(domain)}",
"",
"[site.source]",
f"kind = {_toml_string(source_kind)}",
f"path = {_toml_string(source_path)}",
f"canonical_url = {_toml_string('https://' + domain)}",
"",
"[site.outputs]",
f"root = {_toml_string(output_root)}",
f"plan_root = {_toml_string(plan_root)}",
"",
"[site.interactions]",
'default_tier = "static"',
'deny_routes = ["/admin/*", "/internal/*", "/webhooks/*"]',
"",
]
)
for protocol in COMMON_PROTOCOLS:
lines.extend(_protocol_section(site_id, protocol, enabled=protocol in selected_protocols))
return "\n".join(lines).rstrip() + "\n"
def parse_protocol_filters(raw_protocols: list[str] | tuple[str, ...]) -> tuple[str, ...]:
values: list[str] = []
for raw_protocol in raw_protocols:
values.extend(protocol.strip() for protocol in raw_protocol.split(",") if protocol.strip())
return tuple(values)
def _selected_protocols(*, preset: str, protocols: tuple[str, ...]) -> tuple[str, ...]:
selected = protocols or PRESETS[preset]
invalid = [protocol for protocol in selected if protocol not in COMMON_PROTOCOLS]
if invalid:
names = ", ".join(COMMON_PROTOCOLS)
raise ValueError(f"unsupported init protocol(s): {', '.join(invalid)}; choose from {names}")
deduped = tuple(dict.fromkeys(selected))
if not deduped:
raise ValueError("at least one protocol must be selected")
return deduped
def _profile_sections(selected_protocols: tuple[str, ...]) -> list[str]:
lines: list[str] = []
profiles = _profile_names(selected_protocols)
if "vps-full" in profiles:
lines.extend(
[
"[profiles.vps-full]",
'description = "Server deployment for selected transports."',
f"protocols = {_toml_array(selected_protocols)}",
'platform = "linux-systemd"',
"write_artifacts = true",
"",
]
)
if "tor-i2p" in profiles:
lines.extend(
[
"[profiles.tor-i2p]",
'description = "Privacy-network deployment without clearnet ingress."',
'protocols = ["tor", "i2p"]',
"write_artifacts = true",
"",
]
)
if "mobile-i2p" in profiles:
lines.extend(
[
"[profiles.mobile-i2p]",
'description = "User-space I2P-only deployment for Android/Termux-style hosts."',
'protocols = ["i2p"]',
'platform = "android-termux"',
"write_artifacts = true",
"",
]
)
return lines
def _profile_names(selected_protocols: tuple[str, ...]) -> tuple[str, ...]:
names = ["vps-full"]
if "tor" in selected_protocols and "i2p" in selected_protocols:
names.append("tor-i2p")
if "i2p" in selected_protocols:
names.append("mobile-i2p")
return tuple(names)
def _protocol_section(site_id: str, protocol: str, *, enabled: bool) -> list[str]:
if protocol == "clearnet":
return [
"[site.protocols.clearnet]",
f"enabled = {_toml_bool(enabled)}",
'renderer = "clearnet"',
'daemon = "nginx"',
'daemon_policy = "adopt"',
"",
]
if protocol == "tor":
return [
"[site.protocols.tor]",
f"enabled = {_toml_bool(enabled)}",
'renderer = "privacy-html"',
'daemon = "tor"',
'daemon_policy = "auto"',
"local_port = 18080",
f"hidden_service_dir = {_toml_string('./.ampg/state/' + site_id + '/tor/hidden-service')}",
'onion_location = "auto"',
"max_asset_bytes = 1048576",
'script_policy = "strip"',
"",
]
if protocol == "i2p":
return [
"[site.protocols.i2p]",
f"enabled = {_toml_bool(enabled)}",
'renderer = "privacy-html"',
'daemon = "i2pd"',
'daemon_policy = "auto"',
"local_port = 18081",
f"keys_file = {_toml_string(site_id + '-web.dat')}",
f"tunnel_name = {_toml_string(site_id + '-web')}",
"max_asset_bytes = 1048576",
'script_policy = "strip"',
"",
]
if protocol == "gemini":
return [
"[site.protocols.gemini]",
f"enabled = {_toml_bool(enabled)}",
'renderer = "gemtext"',
'daemon = "agate"',
'daemon_policy = "auto"',
"port = 1965",
"max_asset_bytes = 1048576",
"",
]
raise ValueError(f"unsupported protocol {protocol!r}")
def _validate_site_id(site_id: str) -> None:
if not site_id:
raise ValueError("site id must not be empty")
allowed = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-")
if any(char not in allowed for char in site_id):
raise ValueError("site id may only contain letters, numbers, underscores, and hyphens")
def _resolve_for_config(path: Path, base_dir: Path) -> Path:
if path.is_absolute():
return path.resolve()
return (base_dir / path).resolve()
def _config_path_text(path: Path, base_dir: Path) -> str:
relative = os.path.relpath(path, base_dir)
if Path(relative).parts.count("..") <= 2:
if not relative.startswith("."):
return f"./{relative}"
return relative
return str(path)
def _toml_string(value: str) -> str:
return json.dumps(value)
def _toml_array(values: tuple[str, ...]) -> str:
return "[" + ", ".join(_toml_string(value) for value in values) + "]"
def _toml_bool(value: bool) -> str:
return "true" if value else "false"
docs/configuration.md
+30
View File
@@ -5,6 +5,36 @@
AMPG is configured from one TOML file. The file declares global paths, sites, source
adapters, render targets, and daemon policy.
## Create A Config
For an existing static HTML site, start with `init site`:
```sh
python3 -m ampg --config gateway.toml init site example \
--domain example.org \
--source ../example.org \
--preset full
```
Presets are concise transport bundles:
- `full`: clearnet, Tor, and I2P.
- `privacy`: Tor and I2P.
- `i2p-only`, `tor-only`, `clearnet-only`, or `gemini-only`: one transport.
Use `--protocol` to select exact transports instead of a preset:
```sh
python3 -m ampg --config gateway.toml init site example \
--domain example.org \
--source ../example.org \
--protocol i2p
```
The generated config keeps common transports as `enabled = true/false` toggles and
creates practical profiles such as `vps-full`, `tor-i2p`, and `mobile-i2p` when they
apply.
## Minimal Wownero pilot config
```toml
docs/wownero-integration.md
+1
View File
@@ -149,6 +149,7 @@ daemon_policy = "auto"
## Local Commands
```sh
python3 -m ampg --config gateway.toml init site wownero --domain wownero.org --source ../wownero.org-website --preset full
python3 -m ampg --config examples/wownero.gateway.toml plan
python3 -m ampg --config examples/wownero.gateway.toml build
python3 -m ampg --config examples/wownero.gateway.toml manifest
tests/test_onboarding.py
+141
View File
@@ -0,0 +1,141 @@
import contextlib
import io
import tempfile
import unittest
from pathlib import Path
from ampg.cli import main
from ampg.config import load_config
def _source(root: Path) -> Path:
source = root / "site"
source.mkdir()
(source / "index.html").write_text("<h1>Hello</h1>", encoding="utf-8")
return source
def _run_cli(args: list[str]) -> tuple[int, str, str]:
stdout = io.StringIO()
stderr = io.StringIO()
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
status = main(args)
return status, stdout.getvalue(), stderr.getvalue()
class OnboardingTest(unittest.TestCase):
def test_init_site_writes_loadable_gateway_config(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve()
config_path = root / "gateway.toml"
status, output, error = _run_cli(
[
"--config",
str(config_path),
"init",
"site",
"example",
"--domain",
"example.test",
"--source",
str(_source(root)),
"--preset",
"privacy",
]
)
config = load_config(config_path)
self.assertEqual(0, status, error)
self.assertIn("AMPG_INIT_SITE status=written", output)
self.assertIn("AMPG_INIT_PROFILE name=tor-i2p", output)
self.assertIn("AMPG_INIT_TOGGLE protocol=clearnet enabled=false", output)
self.assertIn("AMPG_INIT_TOGGLE protocol=tor enabled=true", output)
self.assertEqual("example", config.sites[0].id)
self.assertTrue(config.sites[0].protocols["tor"].enabled)
self.assertTrue(config.sites[0].protocols["i2p"].enabled)
self.assertFalse(config.sites[0].protocols["clearnet"].enabled)
self.assertFalse(config.sites[0].protocols["gemini"].enabled)
def test_init_site_protocol_override_can_make_i2p_only_config(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve()
config_path = root / "gateway.toml"
status, output, error = _run_cli(
[
"--config",
str(config_path),
"init",
"site",
"example",
"--domain",
"example.test",
"--source",
str(_source(root)),
"--protocol",
"i2p",
]
)
config = load_config(config_path)
self.assertEqual(0, status, error)
self.assertIn("protocols=i2p", output)
self.assertTrue(config.sites[0].protocols["i2p"].enabled)
self.assertFalse(config.sites[0].protocols["tor"].enabled)
self.assertIn("mobile-i2p", config.profiles)
def test_init_site_refuses_overwrite_without_force(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve()
config_path = root / "gateway.toml"
source = _source(root)
args = [
"--config",
str(config_path),
"init",
"site",
"example",
"--domain",
"example.test",
"--source",
str(source),
]
self.assertEqual(0, _run_cli(args)[0])
status, _, error = _run_cli(args)
self.assertEqual(1, status)
self.assertIn("config already exists", error)
def test_init_site_output_can_build_immediately(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp).resolve()
config_path = root / "gateway.toml"
status, _, error = _run_cli(
[
"--config",
str(config_path),
"init",
"site",
"example",
"--domain",
"example.test",
"--source",
str(_source(root)),
"--protocol",
"i2p",
]
)
self.assertEqual(0, status, error)
status, output, error = _run_cli(["--config", str(config_path), "build"])
self.assertEqual(0, status, error)
self.assertIn("AMPG_BUILD site=example protocol=i2p", output)
if __name__ == "__main__":
unittest.main()