brov-macbook/scripts/netcore/gen.py

#!/usr/bin/env python3
"""BroV network core prototype: builds core/config.yaml for mihomo from src/*.

keys/*.vpnkey        one vpn:// key per Amnezia connection (AmneziaWG 2/3); BroV adds them
                     and rewrites keys/amnezia.yaml itself, gen.py does the same on a full build
src/saga.conf        WireGuard client config  -> only 192.168.8.0/24
src/planet9.conf     WireGuard client config  -> only 192.168.68.0/24
src/vless.sub        3x-ui subscription URL (all VLESS nodes)

Secrets stay in this private folder; nothing here goes to git.
"""
import base64, json, os, re, secrets, zlib, configparser

HERE = os.path.dirname(os.path.abspath(__file__))
SRC = os.path.join(HERE, "src")
KEYS = os.path.join(HERE, "keys")
# Installed next to its inputs in /Library/Application Support/BroV/netcore (root, 0700):
# the root core reads its config from here, nothing user-writable is involved.
CORE = HERE

HOME_NET = "192.168.10.0/24"
CLIENTS = {  # name: (conf file, routed subnets, what "off" means)
    # Client networks: off = REJECT (unreachable). Only the LANs are routed — their
    # tunnel subnets overlap (Saga and home both use 10.0.0.x).
    "saga": ("saga.conf", ["192.168.8.0/24"], "REJECT"),
    "planet9": ("planet9.conf", ["192.168.68.0/24", "172.3.3.0/24"], "REJECT"),
    # Home: off = DIRECT (you're at home, the LAN is right there); on = through the
    # home WireGuard when away. Optional: only if src/home.conf exists.
    "home": ("home.conf", ["192.168.10.0/24"], "DIRECT"),
}
AI_DOMAINS = ["anthropic.com", "claude.ai", "claude.com", "openai.com", "chatgpt.com",
              "oaistatic.com", "oaiusercontent.com", "github.com", "githubusercontent.com"]


def mid(v, default):
    """'100-120' -> 110 (mihomo takes single ints for timers)."""
    if v is None or v == "":
        return default
    m = re.match(r"^\s*(\d+)\s*-\s*(\d+)\s*$", str(v))
    return (int(m.group(1)) + int(m.group(2))) // 2 if m else int(v)


def amnezia(path, name):
    key = open(path).read().strip()[len("vpn://"):]
    key += "=" * (-len(key) % 4)
    j = json.loads(zlib.decompress(base64.urlsafe_b64decode(key)[4:]))
    awg = j["containers"][0]["awg"]
    c = awg["last_config"] if isinstance(awg["last_config"], dict) else json.loads(awg["last_config"])
    ver = 3 if c.get("HeaderProtectionKey") else 2
    opt = {"version": ver, "jc": int(c["Jc"]), "jmin": int(c["Jmin"]), "jmax": int(c["Jmax"]),
           "s1": int(c["S1"]), "s2": int(c["S2"])}
    for k in ("S3", "S4"):
        if c.get(k):
            opt[k.lower()] = int(c[k])
    for k in ("H1", "H2", "H3", "H4"):
        v = c[k]
        opt[k.lower()] = int(v) if str(v).isdigit() else v
    for k in ("I1", "I2", "I3", "I4", "I5"):
        if c.get(k):
            opt[k.lower()] = c[k]
    if ver == 3:
        opt.update({
            "header-protection-key": c["HeaderProtectionKey"],
            "content-padding-addition": c.get("ContentPaddingAddition", "0"),
            "rekey-after-time": mid(c.get("RekeyAfterTime"), 120),
            "rekey-timeout": mid(c.get("RekeyTimeout"), 5),
            "reject-after-time": mid(c.get("RejectAfterTime"), 180),
            "keepalive-timeout": mid(c.get("KeepaliveTimeout"), 10),
            "max-handshake-attempts": mid(c.get("MaxHandshakeAttempts"), 18),
            "random-trailers": c.get("RandomTrailers") == "on",
            "disable-cookies": c.get("DisableCookies") == "on",
        })
    return {
        "name": name, "type": "wireguard", "server": c["hostName"], "port": int(c["port"]),
        "ip": c["client_ip"], "private-key": c["client_priv_key"], "public-key": c["server_pub_key"],
        "pre-shared-key": c.get("psk_key") or None, "mtu": int(c.get("mtu", 1376)), "udp": True,
        "persistent-keepalive": mid(c.get("persistent_keep_alive"), 25),
        "amnezia-wg-option": opt,
    }


def wg(name, path):
    p = configparser.ConfigParser()
    p.optionxform = str
    p.read(os.path.join(SRC, path))
    i, peer = p["Interface"], p["Peer"]
    host, port = peer["Endpoint"].rsplit(":", 1)
    out = {"name": name, "type": "wireguard", "server": host, "port": int(port),
           "ip": i["Address"].split("/")[0], "private-key": i["PrivateKey"],
           "public-key": peer["PublicKey"], "mtu": int(i.get("MTU", 1420)), "udp": True}
    if peer.get("PresharedKey"):
        out["pre-shared-key"] = peer["PresharedKey"]
    if peer.get("PersistentKeepalive"):
        out["persistent-keepalive"] = int(peer["PersistentKeepalive"])
    return out


def y(v, ind=0):
    """Tiny YAML emitter (no PyYAML dependency)."""
    pad = "  " * ind
    if isinstance(v, dict):
        lines = []
        for k, x in v.items():
            if x is None:
                continue
            if isinstance(x, (dict, list)) and x:
                lines.append(f"{pad}{k}:\n{y(x, ind + 1)}")
            else:
                lines.append(f"{pad}{k}: {scalar(x)}")
        return "\n".join(lines)
    if isinstance(v, list):
        lines = []
        for x in v:
            if isinstance(x, dict):
                body = y(x, ind + 1).lstrip()
                lines.append(f"{pad}- {body}")
            else:
                lines.append(f"{pad}- {scalar(x)}")
        return "\n".join(lines)
    return pad + scalar(v)


def scalar(x):
    if isinstance(x, bool):
        return "true" if x else "false"
    if isinstance(x, (int, float)):
        return str(x)
    if isinstance(x, list) and not x:
        return "[]"
    return json.dumps(str(x), ensure_ascii=False)


def main():
    os.makedirs(CORE, exist_ok=True)
    secret_file = os.path.join(CORE, "api.secret")
    if not os.path.exists(secret_file):
        fd = os.open(secret_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        with os.fdopen(fd, "w") as f:
            f.write(secrets.token_urlsafe(24))
    api_secret = open(secret_file).read().strip()

    # Amnezia connections live in their own provider file so BroV can add keys live.
    keyfiles = sorted(f for f in os.listdir(KEYS) if f.endswith(".vpnkey"))
    awg = [amnezia(os.path.join(KEYS, f), "AWG " + f[:-len(".vpnkey")]) for f in keyfiles]
    prov = os.path.join(KEYS, "amnezia.yaml")
    with open(os.open(prov, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "w") as f:
        f.write(y({"proxies": awg}) + "\n")
    clients = {n: c for n, c in CLIENTS.items() if os.path.exists(os.path.join(SRC, c[0]))}
    proxies = [wg(n, f) for n, (f, _, _) in clients.items()]
    sub = open(os.path.join(SRC, "vless.sub")).read().strip()

    rules = ([] if "home" in clients else [f"IP-CIDR,{HOME_NET},DIRECT,no-resolve"]) + [
             "IP-CIDR,127.0.0.0/8,DIRECT,no-resolve",
             # Home's own public IP (RustDesk, Gitea): never via a foreign exit.
             "IP-CIDR,79.111.14.0/32,DIRECT,no-resolve", "DOMAIN-SUFFIX,sanjeev.ru,DIRECT"]
    for name, (_, nets, _) in clients.items():
        # Through a switch group: BroV turns client networks on/off without a reload.
        rules += [f"IP-CIDR,{n},{name}-sw,no-resolve" for n in nets]
    rules += [f"DOMAIN-SUFFIX,{d},ai-out" for d in AI_DOMAINS]
    rules += ["DOMAIN-SUFFIX,ru,DIRECT", "DOMAIN-SUFFIX,su,DIRECT", "DOMAIN-SUFFIX,xn--p1ai,DIRECT",
              "MATCH,ai-out"]

    cfg = {
        "mixed-port": 7890, "allow-lan": False, "mode": "rule", "log-level": "error", "ipv6": False,
        "external-controller": "127.0.0.1:9097", "secret": api_secret, "unified-delay": True,
        "find-process-mode": "strict",
        "profile": {"store-selected": True},
        "tun": {"enable": True, "stack": "mixed", "auto-route": True, "auto-detect-interface": True,
                "dns-hijack": ["any:53"], "mtu": 1400},
        "dns": {"enable": True, "ipv6": False, "enhanced-mode": "fake-ip", "fake-ip-range": "198.18.0.1/16",
                "fake-ip-filter": ["*.lan", "*.local", "+.duckdns.org"],
                "default-nameserver": ["77.88.8.8", "1.1.1.1"],
                "proxy-server-nameserver": ["77.88.8.8", "1.1.1.1"],
                "nameserver": ["https://1.1.1.1/dns-query#ai-out", "https://8.8.8.8/dns-query#ai-out"],
                "direct-nameserver": ["77.88.8.8", "77.88.8.1"]},
        "proxies": proxies,
        "proxy-providers": {
            "amnezia-keys": {"type": "file", "path": "./keys/amnezia.yaml",
                             "health-check": {"enable": True, "url": "https://www.gstatic.com/generate_204",
                                              "interval": 300}},
            "vless-cluster": {
            # Fetch the list directly: the nodes themselves are dialled directly anyway.
            "type": "http", "url": sub, "interval": 43200, "path": "./providers/vless.yaml", "proxy": "DIRECT",
            "health-check": {"enable": True, "url": "https://www.gstatic.com/generate_204", "interval": 300}}},
        "proxy-groups": [
            # What BroV's globe panel switches: "auto", the Amnezia group, or one VLESS node.
            {"name": "ai-out", "type": "select", "proxies": ["auto", "amnezia"], "use": ["vless-cluster"]},
            # Fastest alive exit among every Amnezia connection and every VLESS node;
            # switches only when another one is 100+ ms faster.
            {"name": "auto", "type": "url-test", "use": ["amnezia-keys", "vless-cluster"],
             "url": "https://www.gstatic.com/generate_204", "interval": 120, "tolerance": 100, "lazy": False},
            # Amnezia: "amnezia-auto" (fastest connection) or one fixed connection.
            {"name": "amnezia", "type": "select", "proxies": ["amnezia-auto"], "use": ["amnezia-keys"]},
            {"name": "amnezia-auto", "type": "url-test", "use": ["amnezia-keys"],
             "url": "https://www.gstatic.com/generate_204", "interval": 120, "tolerance": 100, "lazy": False},
        ] + [
            # Client networks: off (REJECT) until switched on in BroV.
            # Client networks: first option = "off" (REJECT, or DIRECT for home).
            {"name": f"{n}-sw", "type": "select", "proxies": [off, n]} for n, (_, _, off) in clients.items()
        ],
        "rules": rules,
    }
    # The API secret never leaves this root-only folder: BroV goes through netctl.py.
    path = os.path.join(CORE, "config.yaml")
    with open(os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600), "w") as f:
        f.write("# Generated by gen.py — do not edit by hand, do not share.\n" + y(cfg) + "\n")
    print("wrote", path, "|", len(proxies), "proxies + vless subscription |", len(rules), "rules")


def check(path):
    """--check <file>: is this a usable Amnezia key? Prints JSON for BroV."""
    try:
        raw = open(path).read().strip()
        if not raw.startswith("vpn://"):
            raise ValueError("ключ должен начинаться с vpn://")
        key = raw[len("vpn://"):]
        key += "=" * (-len(key) % 4)
        j = json.loads(zlib.decompress(base64.urlsafe_b64decode(key)[4:]))
        cont = j["containers"][0]
        if "awg" not in cont:
            raise ValueError("это не AmneziaWG (контейнер %s) — пока поддерживается только AmneziaWG" % cont.get("container"))
        p = amnezia(path, "check")
        print(json.dumps({"ok": True, "name": j.get("description") or p["server"], "server": p["server"],
                          "port": p["port"], "version": p["amnezia-wg-option"]["version"]}, ensure_ascii=False))
    except Exception as e:
        print(json.dumps({"ok": False, "error": str(e) or e.__class__.__name__}, ensure_ascii=False))


if __name__ == "__main__":
    import sys
    if len(sys.argv) == 3 and sys.argv[1] == "--check":
        check(sys.argv[2])
    else:
        main()