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

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path

from .config import GatewayConfig, ProtocolConfig, SiteConfig


@dataclass(frozen=True)
class PlanLine:
    site_id: str
    protocol: str
    renderer: str
    output_root: Path
    daemon: str
    daemon_policy: str
    action: str
    artifacts: tuple[Path, ...]


@dataclass(frozen=True)
class PlanArtifact:
    path: Path
    content: str


def plan_gateway(config: GatewayConfig) -> list[PlanLine]:
    lines: list[PlanLine] = []
    for site in config.sites:
        for protocol in site.protocols.values():
            if not protocol.enabled:
                continue
            artifacts = tuple(artifact.path for artifact in plan_artifacts(site, protocol))
            lines.append(
                PlanLine(
                    site_id=site.id,
                    protocol=protocol.name,
                    renderer=protocol.renderer,
                    output_root=site.outputs.root / protocol.name,
                    daemon=protocol.daemon,
                    daemon_policy=protocol.daemon_policy,
                    action=_action_for_policy(protocol.daemon_policy),
                    artifacts=artifacts,
                )
            )
    return lines


def write_plan_artifacts(config: GatewayConfig) -> list[PlanArtifact]:
    artifacts: list[PlanArtifact] = []
    for site in config.sites:
        for protocol in site.protocols.values():
            if not protocol.enabled:
                continue
            artifacts.extend(plan_artifacts(site, protocol))
    for artifact in artifacts:
        artifact.path.parent.mkdir(parents=True, exist_ok=True)
        artifact.path.write_text(artifact.content, encoding="utf-8")
    return artifacts


def plan_artifacts(site: SiteConfig, protocol: ProtocolConfig) -> list[PlanArtifact]:
    if protocol.name == "clearnet" and protocol.daemon == "nginx":
        return [_nginx_clearnet_artifact(site, protocol)]
    if protocol.name == "tor":
        artifacts = []
        if protocol.daemon == "tor":
            artifacts.append(_tor_artifact(site, protocol))
        artifacts.append(_nginx_loopback_artifact(site, protocol))
        return artifacts
    if protocol.name == "i2p":
        artifacts = []
        if protocol.daemon == "i2pd":
            artifacts.append(_i2pd_artifact(site, protocol))
        artifacts.append(_nginx_loopback_artifact(site, protocol))
        return artifacts
    if protocol.name == "gemini":
        if protocol.daemon == "agate":
            return [_agate_artifact(site, protocol)]
        return []
    return []


def _action_for_policy(policy: str) -> str:
    if policy == "external":
        return "render outputs and config snippets only"
    if policy == "adopt":
        return "adopt existing daemon; fail if missing"
    if policy == "manage":
        return "create AMPG-owned daemon instance"
    if policy == "auto":
        return "adopt healthy daemon, otherwise manage AMPG-owned instance"
    return "unknown policy; no apply support"


def _protocol_plan_dir(site: SiteConfig, protocol: ProtocolConfig) -> Path:
    return site.outputs.plan_root / site.id / protocol.name


def _nginx_clearnet_artifact(site: SiteConfig, protocol: ProtocolConfig) -> PlanArtifact:
    path = _protocol_plan_dir(site, protocol) / "nginx-server.conf"
    content = f"""# Generated by AMPG. Review before installing.
server {{
    listen 80;
    listen [::]:80;
    server_name {site.domain} www.{site.domain};
    root {site.outputs.root / protocol.name};
    index index.html;

    location / {{
        try_files $uri $uri/ /index.html;
    }}
}}
"""
    return PlanArtifact(path=path, content=content)


def _nginx_loopback_artifact(site: SiteConfig, protocol: ProtocolConfig) -> PlanArtifact:
    port = int(protocol.options.get("local_port", 18080 if protocol.name == "tor" else 18081))
    path = _protocol_plan_dir(site, protocol) / "nginx-loopback.conf"
    content = f"""# Generated by AMPG. Review before installing.
server {{
    listen 127.0.0.1:{port};
    server_name _;
    root {site.outputs.root / protocol.name};
    index index.html;

    location / {{
        try_files $uri $uri/ /index.html;
    }}
}}
"""
    return PlanArtifact(path=path, content=content)


def _tor_artifact(site: SiteConfig, protocol: ProtocolConfig) -> PlanArtifact:
    port = int(protocol.options.get("local_port", 18080))
    hidden_service_dir = protocol.options.get(
        "hidden_service_dir", f"/var/lib/ampg/tor/{site.id}"
    )
    path = _protocol_plan_dir(site, protocol) / "torrc-snippet.conf"
    content = f"""# Generated by AMPG. Review before installing.
HiddenServiceDir {hidden_service_dir}/
HiddenServicePort 80 127.0.0.1:{port}
"""
    return PlanArtifact(path=path, content=content)


def _i2pd_artifact(site: SiteConfig, protocol: ProtocolConfig) -> PlanArtifact:
    port = int(protocol.options.get("local_port", 18081))
    key_file = protocol.options.get("keys_file", f"{site.id}-web.dat")
    tunnel_name = protocol.options.get("tunnel_name", f"{site.id}-web")
    path = _protocol_plan_dir(site, protocol) / "i2pd-tunnel.conf"
    content = f"""# Generated by AMPG. Review before installing.
[{tunnel_name}]
type = server
host = 127.0.0.1
port = {port}
keys = {key_file}
"""
    return PlanArtifact(path=path, content=content)


def _agate_artifact(site: SiteConfig, protocol: ProtocolConfig) -> PlanArtifact:
    port = int(protocol.options.get("port", 1965))
    cert_path = protocol.options.get("certificate_path", f"/var/lib/ampg/gemini/{site.id}.crt")
    key_path = protocol.options.get("key_path", f"/var/lib/ampg/gemini/{site.id}.key")
    path = _protocol_plan_dir(site, protocol) / "agate-plan.txt"
    content = f"""# Generated by AMPG. Review before installing.
daemon = agate
hostname = {site.domain}
listen = [::]:{port}
content_root = {site.outputs.root / protocol.name}
certificate = {cert_path}
private_key = {key_path}
"""
    return PlanArtifact(path=path, content=content)