adaptive-multi-protocol-gateway/ampg/onboarding.py

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 = "auto"',
            "",
        ]
    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"