mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 07:31:32 +02:00
278 lines
8.9 KiB
Python
278 lines
8.9 KiB
Python
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"
|