#!/usr/bin/env bash
set -euo pipefail

usage() {
  cat <<'EOF'
Usage: Scripts/burrow-proxy-selftest

Run a controlled Burrow proxy packet-runtime test without changing system proxy
settings or using the user's Burrow database. The script starts local Trojan and
Shadowsocks servers, starts a Burrow daemon against a temporary database/socket,
imports a temporary ProxySubscription pointed at those servers, then runs the
packet-level proxy TCP, UDP, UDP-over-TCP, and plugin probes through daemon
packet streaming.
EOF
}

if [[ "${1:-}" == "-h" || "${1:-}" == "--help" ]]; then
  usage
  exit 0
fi

repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
burrow_bin="${BURROW_BIN:-$repo_root/target/debug/burrow}"

if [[ -z "${BURROW_BIN:-}" ]]; then
  (cd "$repo_root" && cargo build -p burrow)
elif [[ ! -x "$burrow_bin" ]]; then
  echo "BURROW_BIN is not executable: $burrow_bin" >&2
  exit 2
fi

if ! command -v openssl >/dev/null 2>&1; then
  echo "openssl is required for the local TLS server certificate" >&2
  exit 2
fi

if ! command -v uv >/dev/null 2>&1; then
  echo "uv is required for the local Python self-test harness" >&2
  exit 2
fi

tmpdir="$(mktemp -d "${TMPDIR:-/tmp}/burrow-proxy-selftest.XXXXXX")"
daemon_pid=""
server_pid=""
ss_server_pid=""
status=0

cleanup() {
  if [[ -n "$ss_server_pid" ]]; then
    kill "$ss_server_pid" >/dev/null 2>&1 || true
  fi
  if [[ -n "$server_pid" ]]; then
    kill "$server_pid" >/dev/null 2>&1 || true
  fi
  if [[ -n "$daemon_pid" ]]; then
    kill "$daemon_pid" >/dev/null 2>&1 || true
  fi
  if [[ "${BURROW_SELFTEST_KEEP:-}" == "1" || "$status" -ne 0 ]]; then
    echo "preserving self-test artifacts: $tmpdir" >&2
  else
    rm -rf "$tmpdir"
  fi
}
trap 'status=$?; cleanup' EXIT

cert="$tmpdir/cert.pem"
key="$tmpdir/key.pem"
socket="$tmpdir/burrow.sock"
port_file="$tmpdir/server.port"
ss_port_file="$tmpdir/shadowsocks-server.port"
server_result="$tmpdir/server-result.json"
ss_server_result="$tmpdir/shadowsocks-server-result.json"
payload="$tmpdir/proxy-payload.json"
daemon_log="$tmpdir/daemon.log"
server_log="$tmpdir/server.log"
ss_server_log="$tmpdir/shadowsocks-server.log"

openssl req \
  -x509 \
  -newkey rsa:2048 \
  -nodes \
  -subj /CN=localhost \
  -keyout "$key" \
  -out "$cert" \
  -days 1 >/dev/null 2>&1

uv run python - "$port_file" "$server_result" "$cert" "$key" >"$server_log" 2>&1 <<'PY' &
from __future__ import annotations

import hashlib
import json
import socket
import ssl
import struct
import sys

port_file, result_file, cert_file, key_file = sys.argv[1:]
password = "secret"


def read_exact(sock: ssl.SSLSocket, length: int) -> bytes:
    chunks: list[bytes] = []
    remaining = length
    while remaining:
        chunk = sock.recv(remaining)
        if not chunk:
            raise EOFError(f"expected {remaining} more bytes")
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)


def read_socks_addr(sock: ssl.SSLSocket) -> dict[str, object]:
    atyp = read_exact(sock, 1)[0]
    if atyp == 1:
        host = socket.inet_ntop(socket.AF_INET, read_exact(sock, 4))
    elif atyp == 4:
        host = socket.inet_ntop(socket.AF_INET6, read_exact(sock, 16))
    elif atyp == 3:
        length = read_exact(sock, 1)[0]
        host = read_exact(sock, length).decode("idna")
    else:
        raise ValueError(f"unsupported SOCKS address type {atyp}")
    port = struct.unpack("!H", read_exact(sock, 2))[0]
    return {"atyp": atyp, "host": host, "port": port}


def read_http_headers(sock: ssl.SSLSocket) -> bytes:
    data = bytearray()
    while b"\r\n\r\n" not in data:
        chunk = sock.recv(4096)
        if not chunk:
            break
        data.extend(chunk)
        if len(data) > 65535:
            raise ValueError("HTTP request headers exceeded 64 KiB")
    return bytes(data)


ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ctx.load_cert_chain(cert_file, key_file)
ctx.set_alpn_protocols(["h2", "http/1.1"])

listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
listener.bind(("127.0.0.1", 0))
listener.listen(1)
port = listener.getsockname()[1]
with open(port_file, "w", encoding="utf-8") as f:
    f.write(str(port))

raw, peer = listener.accept()
raw.settimeout(10)
result: dict[str, object] = {"peer": str(peer)}
try:
    with ctx.wrap_socket(raw, server_side=True) as tls:
        tls.settimeout(10)
        result["alpn"] = tls.selected_alpn_protocol()
        got_hash = read_exact(tls, 56)
        expected_hash = hashlib.sha224(password.encode()).hexdigest().encode()
        result["password_hash_ok"] = got_hash == expected_hash
        if not result["password_hash_ok"]:
            raise ValueError("unexpected Trojan password hash")
        if read_exact(tls, 2) != b"\r\n":
            raise ValueError("missing Trojan password delimiter")
        command = read_exact(tls, 1)[0]
        result["command"] = command
        target = read_socks_addr(tls)
        result["target"] = target
        if read_exact(tls, 2) != b"\r\n":
            raise ValueError("missing Trojan header delimiter")
        request = read_http_headers(tls)
        result["request_first_line"] = request.splitlines()[0].decode("utf-8", "replace")
        body = json.dumps(
            {
                "ok": True,
                "target": target,
                "request_first_line": result["request_first_line"],
            },
            sort_keys=True,
        ).encode()
        tls.sendall(
            b"HTTP/1.1 200 OK\r\n"
            + f"Content-Length: {len(body)}\r\n".encode()
            + b"Connection: close\r\n"
            + b"Content-Type: application/json\r\n\r\n"
            + body
        )
finally:
    with open(result_file, "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=2, sort_keys=True)
    listener.close()
PY
server_pid=$!

for _ in {1..100}; do
  if [[ -s "$port_file" ]]; then
    break
  fi
  if ! kill -0 "$server_pid" >/dev/null 2>&1; then
    echo "local Trojan server exited before listening" >&2
    cat "$server_log" >&2 || true
    exit 1
  fi
  sleep 0.05
done

if [[ ! -s "$port_file" ]]; then
  echo "timed out waiting for local Trojan server" >&2
  cat "$server_log" >&2 || true
  exit 1
fi
server_port="$(cat "$port_file")"

uv run python - "$ss_port_file" "$ss_server_result" >"$ss_server_log" 2>&1 <<'PY' &
from __future__ import annotations

import json
import base64
import socket
import struct
import sys

port_file, result_file = sys.argv[1:]


def read_exact(sock: socket.socket, length: int) -> bytes:
    chunks: list[bytes] = []
    remaining = length
    while remaining:
        chunk = sock.recv(remaining)
        if not chunk:
            raise EOFError(f"expected {remaining} more bytes")
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)


def read_socks_addr(sock: socket.socket) -> dict[str, object]:
    atyp = read_exact(sock, 1)[0]
    if atyp == 1:
        host = socket.inet_ntop(socket.AF_INET, read_exact(sock, 4))
    elif atyp == 4:
        host = socket.inet_ntop(socket.AF_INET6, read_exact(sock, 16))
    elif atyp == 3:
        length = read_exact(sock, 1)[0]
        host = read_exact(sock, length).decode("idna")
    else:
        raise ValueError(f"unsupported SOCKS address type {atyp}")
    port = struct.unpack("!H", read_exact(sock, 2))[0]
    return {"atyp": atyp, "host": host, "port": port}


def parse_socks_addr_packet(data: bytes) -> tuple[dict[str, object], int]:
    offset = 0
    atyp = data[offset]
    offset += 1
    if atyp == 1:
        host = socket.inet_ntop(socket.AF_INET, data[offset : offset + 4])
        offset += 4
    elif atyp == 4:
        host = socket.inet_ntop(socket.AF_INET6, data[offset : offset + 16])
        offset += 16
    elif atyp == 3:
        length = data[offset]
        offset += 1
        host = data[offset : offset + length].decode("idna")
        offset += length
    else:
        raise ValueError(f"unsupported SOCKS address type {atyp}")
    port = struct.unpack("!H", data[offset : offset + 2])[0]
    offset += 2
    return {"atyp": atyp, "host": host, "port": port}, offset


def read_uot_addr(sock: socket.socket) -> tuple[dict[str, object], bytes]:
    raw = bytearray()
    atyp = read_exact(sock, 1)[0]
    raw.append(atyp)
    if atyp == 0:
        addr = read_exact(sock, 4)
        raw.extend(addr)
        host = socket.inet_ntop(socket.AF_INET, addr)
    elif atyp == 1:
        addr = read_exact(sock, 16)
        raw.extend(addr)
        host = socket.inet_ntop(socket.AF_INET6, addr)
    elif atyp == 2:
        length = read_exact(sock, 1)[0]
        raw.append(length)
        addr = read_exact(sock, length)
        raw.extend(addr)
        host = addr.decode("idna")
    else:
        raise ValueError(f"unsupported UDP-over-TCP address type {atyp}")
    port_bytes = read_exact(sock, 2)
    raw.extend(port_bytes)
    port = struct.unpack("!H", port_bytes)[0]
    return {"atyp": atyp, "host": host, "port": port}, bytes(raw)


def read_uot_frame(sock: socket.socket) -> tuple[dict[str, object], bytes, bytes]:
    target, target_bytes = read_uot_addr(sock)
    length_bytes = read_exact(sock, 2)
    payload_len = struct.unpack("!H", length_bytes)[0]
    payload = read_exact(sock, payload_len)
    return target, target_bytes + length_bytes + payload, payload


def read_uot_request(sock: socket.socket) -> dict[str, object]:
    version = read_exact(sock, 1)[0]
    target = read_socks_addr(sock)
    return {"version": version, "target": target}


def handle_uot_connection(
    listener: socket.socket,
    result: dict[str, object],
    prefix: str,
    expect_request: bool,
) -> None:
    uot_raw, uot_peer = listener.accept()
    uot_raw.settimeout(10)
    try:
        uot_magic = read_socks_addr(uot_raw)
        if expect_request:
            result[f"{prefix}_request"] = read_uot_request(uot_raw)
        uot_target, uot_frame, uot_payload = read_uot_frame(uot_raw)
        result[f"{prefix}_peer"] = str(uot_peer)
        result[f"{prefix}_magic_target"] = uot_magic
        result[f"{prefix}_target"] = uot_target
        result[f"{prefix}_payload"] = uot_payload.decode("utf-8", "replace")
        uot_raw.sendall(uot_frame)
    finally:
        uot_raw.close()


def read_http_headers(sock: socket.socket) -> bytes:
    return read_http_headers_with_prefix(sock, b"")


def read_http_headers_with_prefix(sock: socket.socket, prefix: bytes) -> bytes:
    data = bytearray(prefix)
    while b"\r\n\r\n" not in data:
        chunk = sock.recv(4096)
        if not chunk:
            break
        data.extend(chunk)
        if len(data) > 65535:
            raise ValueError("HTTP request headers exceeded 64 KiB")
    return bytes(data)


def read_obfs_http_headers(sock: socket.socket) -> tuple[bytes, bytes]:
    data = bytearray()
    while b"\r\n\r\n" not in data:
        chunk = sock.recv(4096)
        if not chunk:
            break
        data.extend(chunk)
        if len(data) > 65535:
            raise ValueError("simple-obfs HTTP headers exceeded 64 KiB")
    header_end = bytes(data).find(b"\r\n\r\n")
    if header_end < 0:
        raise ValueError("simple-obfs HTTP headers were incomplete")
    header_end += 4
    return bytes(data[:header_end]), bytes(data[header_end:])


def read_simple_obfs_tls_payload(sock: socket.socket) -> tuple[bytes, str | None]:
    header = read_exact(sock, 5)
    if header[0] != 22:
        raise ValueError(f"expected TLS handshake record, got type {header[0]}")
    record_len = struct.unpack("!H", header[3:5])[0]
    record = read_exact(sock, record_len)
    if len(record) < 4 or record[0] != 1:
        raise ValueError("expected TLS ClientHello handshake")
    body_len = int.from_bytes(record[1:4], "big")
    body = record[4 : 4 + body_len]
    offset = 0
    offset += 2  # version
    offset += 32  # random
    session_id_len = body[offset]
    offset += 1 + session_id_len
    cipher_len = struct.unpack("!H", body[offset : offset + 2])[0]
    offset += 2 + cipher_len
    compression_len = body[offset]
    offset += 1 + compression_len
    extension_len = struct.unpack("!H", body[offset : offset + 2])[0]
    offset += 2
    end = offset + extension_len
    ticket_payload: bytes | None = None
    server_name: str | None = None
    while offset + 4 <= end:
        extension_type = struct.unpack("!H", body[offset : offset + 2])[0]
        length = struct.unpack("!H", body[offset + 2 : offset + 4])[0]
        offset += 4
        extension = body[offset : offset + length]
        offset += length
        if extension_type == 0x0023:
            ticket_payload = extension
        elif extension_type == 0x0000 and len(extension) >= 5:
            name_len = struct.unpack("!H", extension[3:5])[0]
            if len(extension) >= 5 + name_len:
                server_name = extension[5 : 5 + name_len].decode("idna")
    if ticket_payload is None:
        raise ValueError("simple-obfs TLS ClientHello did not include payload extension")
    return ticket_payload, server_name


def read_simple_obfs_tls_application_payload(sock: socket.socket) -> bytes:
    header = read_exact(sock, 5)
    if header[:3] != b"\x17\x03\x03":
        raise ValueError(f"expected TLS application-data record, got {header.hex()}")
    record_len = struct.unpack("!H", header[3:5])[0]
    return read_exact(sock, record_len)


def read_simple_obfs_tls_plaintext_headers(sock: socket.socket, prefix: bytes) -> bytes:
    data = bytearray(prefix)
    while b"\r\n\r\n" not in data:
        data.extend(read_simple_obfs_tls_application_payload(sock))
        if len(data) > 65535:
            raise ValueError("simple-obfs TLS plaintext headers exceeded 64 KiB")
    return bytes(data)


def read_client_websocket_frame(sock: socket.socket) -> bytes:
    header = read_exact(sock, 2)
    if header[0] & 0x0F != 2:
        raise ValueError(f"expected binary websocket frame, got opcode {header[0] & 0x0F}")
    if header[1] & 0x80 == 0:
        raise ValueError("client websocket frame was not masked")
    length = header[1] & 0x7F
    if length == 126:
        length = struct.unpack("!H", read_exact(sock, 2))[0]
    elif length == 127:
        length = struct.unpack("!Q", read_exact(sock, 8))[0]
    mask = read_exact(sock, 4)
    payload = bytearray(read_exact(sock, length))
    for index, byte in enumerate(payload):
        payload[index] = byte ^ mask[index % 4]
    return bytes(payload)


def server_websocket_frame(payload: bytes) -> bytes:
    if len(payload) < 126:
        return b"\x82" + bytes([len(payload)]) + payload
    if len(payload) <= 65535:
        return b"\x82\x7e" + struct.pack("!H", len(payload)) + payload
    return b"\x82\x7f" + struct.pack("!Q", len(payload)) + payload


def read_websocket_plaintext_headers(sock: socket.socket, prefix: bytes) -> bytes:
    data = bytearray(prefix)
    while b"\r\n\r\n" not in data:
        data.extend(read_client_websocket_frame(sock))
        if len(data) > 65535:
            raise ValueError("websocket plaintext headers exceeded 64 KiB")
    return bytes(data)


def http_header_value(headers: bytes, name: bytes) -> str | None:
    name = name.lower() + b":"
    for line in headers.splitlines():
        if line.lower().startswith(name):
            return line.decode("utf-8", "replace").split(":", 1)[1].strip()
    return None


def decode_websocket_early_data(value: str | None) -> bytes:
    if not value:
        raise ValueError("missing websocket early-data protocol header")
    padding = "=" * (-len(value) % 4)
    return base64.urlsafe_b64decode((value + padding).encode())


def parse_v2ray_mux_client_payload(payload: bytes) -> bytes:
    open_len = struct.unpack("!H", payload[:2])[0]
    open_frame = payload[2 : 2 + open_len]
    if len(open_frame) != open_len or open_frame[:4] != b"\x00\x00\x01\x00":
        raise ValueError("v2ray mux open frame was invalid")
    offset = 2 + open_len
    data_header = payload[offset : offset + 8]
    if len(data_header) != 8 or data_header[:6] != b"\x00\x04\x00\x00\x02\x01":
        raise ValueError("v2ray mux data frame header was invalid")
    data_len = struct.unpack("!H", data_header[6:8])[0]
    data_start = offset + 8
    data = payload[data_start : data_start + data_len]
    if len(data) != data_len:
        raise ValueError("v2ray mux data frame payload was incomplete")
    return data


def v2ray_mux_server_data_frame(payload: bytes) -> bytes:
    if len(payload) > 65535:
        raise ValueError("v2ray mux payload is too large")
    return b"\x00\x04\x00\x00\x02\x01" + struct.pack("!H", len(payload)) + payload


def parse_v2ray_mux_data_payload(payload: bytes) -> bytes:
    if len(payload) < 8 or payload[:6] != b"\x00\x04\x00\x00\x02\x01":
        raise ValueError("v2ray mux data frame header was invalid")
    data_len = struct.unpack("!H", payload[6:8])[0]
    data = payload[8 : 8 + data_len]
    if len(data) != data_len:
        raise ValueError("v2ray mux data frame payload was incomplete")
    return data


def read_v2ray_mux_plaintext_headers(sock: socket.socket, prefix: bytes) -> bytes:
    data = bytearray(prefix)
    while b"\r\n\r\n" not in data:
        data.extend(parse_v2ray_mux_data_payload(read_client_websocket_frame(sock)))
        if len(data) > 65535:
            raise ValueError("v2ray mux plaintext headers exceeded 64 KiB")
    return bytes(data)


listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
listener.bind(("127.0.0.1", 0))
listener.listen(16)
port = listener.getsockname()[1]
udp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp.bind(("127.0.0.1", port))
udp.settimeout(10)
with open(port_file, "w", encoding="utf-8") as f:
    f.write(str(port))

raw, peer = listener.accept()
raw.settimeout(10)
result: dict[str, object] = {"tcp_peer": str(peer)}
try:
    target = read_socks_addr(raw)
    result["tcp_target"] = target
    request = read_http_headers(raw)
    result["tcp_request_first_line"] = request.splitlines()[0].decode("utf-8", "replace")
    body = json.dumps(
        {
            "ok": True,
            "target": target,
            "request_first_line": result["tcp_request_first_line"],
        },
        sort_keys=True,
    ).encode()
    raw.sendall(
        b"HTTP/1.1 200 OK\r\n"
        + f"Content-Length: {len(body)}\r\n".encode()
        + b"Connection: close\r\n"
        + b"Content-Type: application/json\r\n\r\n"
        + body
    )
    datagram, udp_peer = udp.recvfrom(65535)
    udp_target, used = parse_socks_addr_packet(datagram)
    udp_payload = datagram[used:]
    result["udp_peer"] = str(udp_peer)
    result["udp_target"] = udp_target
    result["udp_payload"] = udp_payload.decode("utf-8", "replace")
    udp.sendto(datagram[:used] + udp_payload, udp_peer)
    handle_uot_connection(listener, result, "uot", False)
    handle_uot_connection(listener, result, "uot_v2", True)
    obfs_raw, obfs_peer = listener.accept()
    obfs_raw.settimeout(10)
    try:
        obfs_headers, obfs_payload = read_obfs_http_headers(obfs_raw)
        obfs_target, used = parse_socks_addr_packet(obfs_payload)
        result["obfs_http_peer"] = str(obfs_peer)
        result["obfs_http_first_line"] = obfs_headers.splitlines()[0].decode(
            "utf-8",
            "replace",
        )
        result["obfs_http_host"] = next(
            (
                line.decode("utf-8", "replace").split(":", 1)[1].strip()
                for line in obfs_headers.splitlines()
                if line.lower().startswith(b"host:")
            ),
            None,
        )
        result["obfs_http_content_length"] = next(
            (
                line.decode("utf-8", "replace").split(":", 1)[1].strip()
                for line in obfs_headers.splitlines()
                if line.lower().startswith(b"content-length:")
            ),
            None,
        )
        result["obfs_http_target"] = obfs_target
        obfs_request = read_http_headers_with_prefix(obfs_raw, obfs_payload[used:])
        result["obfs_http_request_first_line"] = obfs_request.splitlines()[0].decode(
            "utf-8",
            "replace",
        )
        body = json.dumps(
            {
                "ok": True,
                "target": obfs_target,
                "request_first_line": result["obfs_http_request_first_line"],
            },
            sort_keys=True,
        ).encode()
        obfs_raw.sendall(
            b"HTTP/1.1 101 Switching Protocols\r\n"
            b"Upgrade: websocket\r\n"
            b"Connection: Upgrade\r\n\r\n"
            b"HTTP/1.1 200 OK\r\n"
            + f"Content-Length: {len(body)}\r\n".encode()
            + b"Connection: close\r\n"
            + b"Content-Type: application/json\r\n\r\n"
            + body
        )
    finally:
        obfs_raw.close()
    obfs_tls_raw, obfs_tls_peer = listener.accept()
    obfs_tls_raw.settimeout(10)
    try:
        obfs_tls_payload, obfs_tls_server_name = read_simple_obfs_tls_payload(obfs_tls_raw)
        obfs_tls_target, used = parse_socks_addr_packet(obfs_tls_payload)
        obfs_tls_request = read_simple_obfs_tls_plaintext_headers(
            obfs_tls_raw,
            obfs_tls_payload[used:],
        )
        result["obfs_tls_peer"] = str(obfs_tls_peer)
        result["obfs_tls_server_name"] = obfs_tls_server_name
        result["obfs_tls_target"] = obfs_tls_target
        result["obfs_tls_request_first_line"] = obfs_tls_request.splitlines()[0].decode(
            "utf-8",
            "replace",
        )
        body = json.dumps(
            {
                "ok": True,
                "target": obfs_tls_target,
                "request_first_line": result["obfs_tls_request_first_line"],
            },
            sort_keys=True,
        ).encode()
        response = (
            b"HTTP/1.1 200 OK\r\n"
            + f"Content-Length: {len(body)}\r\n".encode()
            + b"Connection: close\r\n"
            + b"Content-Type: application/json\r\n\r\n"
            + body
        )
        obfs_tls_raw.sendall(bytes(105) + len(response).to_bytes(2, "big") + response)
    finally:
        obfs_tls_raw.close()
    v2ray_raw, v2ray_peer = listener.accept()
    v2ray_raw.settimeout(10)
    try:
        v2ray_headers = read_http_headers(v2ray_raw)
        result["v2ray_http_upgrade_peer"] = str(v2ray_peer)
        result["v2ray_http_upgrade_first_line"] = v2ray_headers.splitlines()[0].decode(
            "utf-8",
            "replace",
        )
        result["v2ray_http_upgrade_host"] = next(
            (
                line.decode("utf-8", "replace").split(":", 1)[1].strip()
                for line in v2ray_headers.splitlines()
                if line.lower().startswith(b"host:")
            ),
            None,
        )
        result["v2ray_http_upgrade_has_websocket_key"] = any(
            line.lower().startswith(b"sec-websocket-key:")
            for line in v2ray_headers.splitlines()
        )
        v2ray_raw.sendall(
            b"HTTP/1.1 101 Switching Protocols\r\n"
            b"Upgrade: websocket\r\n"
            b"Connection: Upgrade\r\n\r\n"
        )
        v2ray_target = read_socks_addr(v2ray_raw)
        v2ray_request = read_http_headers(v2ray_raw)
        result["v2ray_http_upgrade_target"] = v2ray_target
        result["v2ray_http_upgrade_request_first_line"] = (
            v2ray_request.splitlines()[0].decode(
                "utf-8",
                "replace",
            )
        )
        body = json.dumps(
            {
                "ok": True,
                "target": v2ray_target,
                "request_first_line": result["v2ray_http_upgrade_request_first_line"],
            },
            sort_keys=True,
        ).encode()
        v2ray_raw.sendall(
            b"HTTP/1.1 200 OK\r\n"
            + f"Content-Length: {len(body)}\r\n".encode()
            + b"Connection: close\r\n"
            + b"Content-Type: application/json\r\n\r\n"
            + body
        )
    finally:
        v2ray_raw.close()
    v2ray_ws_raw, v2ray_ws_peer = listener.accept()
    v2ray_ws_raw.settimeout(10)
    try:
        v2ray_ws_headers = read_http_headers(v2ray_ws_raw)
        result["v2ray_websocket_peer"] = str(v2ray_ws_peer)
        result["v2ray_websocket_first_line"] = (
            v2ray_ws_headers.splitlines()[0].decode(
                "utf-8",
                "replace",
            )
        )
        result["v2ray_websocket_host"] = next(
            (
                line.decode("utf-8", "replace").split(":", 1)[1].strip()
                for line in v2ray_ws_headers.splitlines()
                if line.lower().startswith(b"host:")
            ),
            None,
        )
        result["v2ray_websocket_has_websocket_key"] = any(
            line.lower().startswith(b"sec-websocket-key:")
            for line in v2ray_ws_headers.splitlines()
        )
        v2ray_ws_raw.sendall(
            b"HTTP/1.1 101 Switching Protocols\r\n"
            b"Upgrade: websocket\r\n"
            b"Connection: Upgrade\r\n\r\n"
        )
        v2ray_ws_payload = read_client_websocket_frame(v2ray_ws_raw)
        v2ray_ws_target, used = parse_socks_addr_packet(v2ray_ws_payload)
        v2ray_ws_request = read_websocket_plaintext_headers(
            v2ray_ws_raw,
            v2ray_ws_payload[used:],
        )
        result["v2ray_websocket_target"] = v2ray_ws_target
        result["v2ray_websocket_request_first_line"] = (
            v2ray_ws_request.splitlines()[0].decode(
                "utf-8",
                "replace",
            )
        )
        body = json.dumps(
            {
                "ok": True,
                "target": v2ray_ws_target,
                "request_first_line": result["v2ray_websocket_request_first_line"],
            },
            sort_keys=True,
        ).encode()
        response = (
            b"HTTP/1.1 200 OK\r\n"
            + f"Content-Length: {len(body)}\r\n".encode()
            + b"Connection: close\r\n"
            + b"Content-Type: application/json\r\n\r\n"
            + body
        )
        v2ray_ws_raw.sendall(server_websocket_frame(response))
    finally:
        v2ray_ws_raw.close()
    v2ray_mux_raw, v2ray_mux_peer = listener.accept()
    v2ray_mux_raw.settimeout(10)
    try:
        v2ray_mux_headers = read_http_headers(v2ray_mux_raw)
        result["v2ray_mux_peer"] = str(v2ray_mux_peer)
        result["v2ray_mux_first_line"] = (
            v2ray_mux_headers.splitlines()[0].decode(
                "utf-8",
                "replace",
            )
        )
        result["v2ray_mux_host"] = next(
            (
                line.decode("utf-8", "replace").split(":", 1)[1].strip()
                for line in v2ray_mux_headers.splitlines()
                if line.lower().startswith(b"host:")
            ),
            None,
        )
        result["v2ray_mux_has_websocket_key"] = any(
            line.lower().startswith(b"sec-websocket-key:")
            for line in v2ray_mux_headers.splitlines()
        )
        v2ray_mux_raw.sendall(
            b"HTTP/1.1 101 Switching Protocols\r\n"
            b"Upgrade: websocket\r\n"
            b"Connection: Upgrade\r\n\r\n"
        )
        v2ray_mux_payload = parse_v2ray_mux_client_payload(
            read_client_websocket_frame(v2ray_mux_raw),
        )
        v2ray_mux_target, used = parse_socks_addr_packet(v2ray_mux_payload)
        v2ray_mux_request = read_v2ray_mux_plaintext_headers(
            v2ray_mux_raw,
            v2ray_mux_payload[used:],
        )
        result["v2ray_mux_target"] = v2ray_mux_target
        result["v2ray_mux_request_first_line"] = (
            v2ray_mux_request.splitlines()[0].decode(
                "utf-8",
                "replace",
            )
        )
        body = json.dumps(
            {
                "ok": True,
                "target": v2ray_mux_target,
                "request_first_line": result["v2ray_mux_request_first_line"],
            },
            sort_keys=True,
        ).encode()
        response = (
            b"HTTP/1.1 200 OK\r\n"
            + f"Content-Length: {len(body)}\r\n".encode()
            + b"Connection: close\r\n"
            + b"Content-Type: application/json\r\n\r\n"
            + body
        )
        v2ray_mux_raw.sendall(
            server_websocket_frame(v2ray_mux_server_data_frame(response)),
        )
    finally:
        v2ray_mux_raw.close()
    gost_ws_raw, gost_ws_peer = listener.accept()
    gost_ws_raw.settimeout(10)
    try:
        gost_ws_headers = read_http_headers(gost_ws_raw)
        result["gost_websocket_peer"] = str(gost_ws_peer)
        result["gost_websocket_first_line"] = (
            gost_ws_headers.splitlines()[0].decode(
                "utf-8",
                "replace",
            )
        )
        result["gost_websocket_host"] = next(
            (
                line.decode("utf-8", "replace").split(":", 1)[1].strip()
                for line in gost_ws_headers.splitlines()
                if line.lower().startswith(b"host:")
            ),
            None,
        )
        result["gost_websocket_has_websocket_key"] = any(
            line.lower().startswith(b"sec-websocket-key:")
            for line in gost_ws_headers.splitlines()
        )
        gost_ws_raw.sendall(
            b"HTTP/1.1 101 Switching Protocols\r\n"
            b"Upgrade: websocket\r\n"
            b"Connection: Upgrade\r\n\r\n"
        )
        gost_ws_payload = read_client_websocket_frame(gost_ws_raw)
        gost_ws_target, used = parse_socks_addr_packet(gost_ws_payload)
        gost_ws_request = read_websocket_plaintext_headers(
            gost_ws_raw,
            gost_ws_payload[used:],
        )
        result["gost_websocket_target"] = gost_ws_target
        result["gost_websocket_request_first_line"] = (
            gost_ws_request.splitlines()[0].decode(
                "utf-8",
                "replace",
            )
        )
        body = json.dumps(
            {
                "ok": True,
                "target": gost_ws_target,
                "request_first_line": result["gost_websocket_request_first_line"],
            },
            sort_keys=True,
        ).encode()
        response = (
            b"HTTP/1.1 200 OK\r\n"
            + f"Content-Length: {len(body)}\r\n".encode()
            + b"Connection: close\r\n"
            + b"Content-Type: application/json\r\n\r\n"
            + body
        )
        gost_ws_raw.sendall(server_websocket_frame(response))
    finally:
        gost_ws_raw.close()
    v2ray_ed_raw, v2ray_ed_peer = listener.accept()
    v2ray_ed_raw.settimeout(10)
    try:
        v2ray_ed_headers = read_http_headers(v2ray_ed_raw)
        result["v2ray_early_data_peer"] = str(v2ray_ed_peer)
        result["v2ray_early_data_first_line"] = (
            v2ray_ed_headers.splitlines()[0].decode(
                "utf-8",
                "replace",
            )
        )
        result["v2ray_early_data_host"] = http_header_value(v2ray_ed_headers, b"host")
        result["v2ray_early_data_has_websocket_key"] = (
            http_header_value(v2ray_ed_headers, b"sec-websocket-key") is not None
        )
        early_data = decode_websocket_early_data(
            http_header_value(v2ray_ed_headers, b"sec-websocket-protocol"),
        )
        result["v2ray_early_data_len"] = len(early_data)
        v2ray_ed_raw.sendall(
            b"HTTP/1.1 101 Switching Protocols\r\n"
            b"Upgrade: websocket\r\n"
            b"Connection: Upgrade\r\n\r\n"
        )
        v2ray_ed_payload = early_data + read_client_websocket_frame(v2ray_ed_raw)
        v2ray_ed_target, used = parse_socks_addr_packet(v2ray_ed_payload)
        v2ray_ed_request = read_websocket_plaintext_headers(
            v2ray_ed_raw,
            v2ray_ed_payload[used:],
        )
        result["v2ray_early_data_target"] = v2ray_ed_target
        result["v2ray_early_data_request_first_line"] = (
            v2ray_ed_request.splitlines()[0].decode(
                "utf-8",
                "replace",
            )
        )
        body = json.dumps(
            {
                "ok": True,
                "target": v2ray_ed_target,
                "request_first_line": result["v2ray_early_data_request_first_line"],
            },
            sort_keys=True,
        ).encode()
        response = (
            b"HTTP/1.1 200 OK\r\n"
            + f"Content-Length: {len(body)}\r\n".encode()
            + b"Connection: close\r\n"
            + b"Content-Type: application/json\r\n\r\n"
            + body
        )
        v2ray_ed_raw.sendall(server_websocket_frame(response))
    finally:
        v2ray_ed_raw.close()
finally:
    raw.close()
    udp.close()
    with open(result_file, "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=2, sort_keys=True)
    listener.close()
PY
ss_server_pid=$!

for _ in {1..100}; do
  if [[ -s "$ss_port_file" ]]; then
    break
  fi
  if ! kill -0 "$ss_server_pid" >/dev/null 2>&1; then
    echo "local Shadowsocks server exited before listening" >&2
    cat "$ss_server_log" >&2 || true
    exit 1
  fi
  sleep 0.05
done

if [[ ! -s "$ss_port_file" ]]; then
  echo "timed out waiting for local Shadowsocks server" >&2
  cat "$ss_server_log" >&2 || true
  exit 1
fi
ss_server_port="$(cat "$ss_port_file")"

cat >"$payload" <<EOF
{
  "name": "burrow proxy selftest",
  "source": { "url": "selftest://local-proxies" },
  "detected_format": "uri-list",
  "selected_ordinal": 1,
  "selected_name": "local trojan",
  "nodes": [
    {
      "ordinal": 1,
      "name": "local trojan",
      "protocol": "trojan",
      "server": "127.0.0.1",
      "port": $server_port,
      "warnings": [],
      "config": {
        "type": "trojan",
        "password": "secret",
        "sni": "localhost",
        "alpn": [],
        "alpn_present": true,
        "skip_cert_verify": true,
        "network": "tcp",
        "udp": true,
        "client_fingerprint": null,
        "fingerprint": null
      }
    },
    {
      "ordinal": 2,
      "name": "local shadowsocks none",
      "protocol": "shadowsocks",
      "server": "127.0.0.1",
      "port": $ss_server_port,
      "warnings": [],
      "config": {
        "type": "shadowsocks",
        "cipher": "none",
        "password": "unused",
        "udp": true,
        "udp_over_tcp": false,
        "udp_over_tcp_version": 1,
        "client_fingerprint": null,
        "plugin": null
      }
    },
    {
      "ordinal": 3,
      "name": "local shadowsocks none uot",
      "protocol": "shadowsocks",
      "server": "127.0.0.1",
      "port": $ss_server_port,
      "warnings": [],
      "config": {
        "type": "shadowsocks",
        "cipher": "none",
        "password": "unused",
        "udp": true,
        "udp_over_tcp": true,
        "udp_over_tcp_version": 1,
        "client_fingerprint": null,
        "plugin": null
      }
    },
    {
      "ordinal": 4,
      "name": "local shadowsocks none uot v2",
      "protocol": "shadowsocks",
      "server": "127.0.0.1",
      "port": $ss_server_port,
      "warnings": [],
      "config": {
        "type": "shadowsocks",
        "cipher": "none",
        "password": "unused",
        "udp": true,
        "udp_over_tcp": true,
        "udp_over_tcp_version": 2,
        "client_fingerprint": null,
        "plugin": null
      }
    },
    {
      "ordinal": 5,
      "name": "local shadowsocks none obfs http",
      "protocol": "shadowsocks",
      "server": "127.0.0.1",
      "port": $ss_server_port,
      "warnings": [],
      "config": {
        "type": "shadowsocks",
        "cipher": "none",
        "password": "unused",
        "udp": true,
        "udp_over_tcp": false,
        "udp_over_tcp_version": 1,
        "client_fingerprint": null,
        "plugin": {
          "name": "obfs",
          "opts": {
            "mode": "http",
            "host": "front.example"
          }
        }
      }
    },
    {
      "ordinal": 6,
      "name": "local shadowsocks none obfs tls",
      "protocol": "shadowsocks",
      "server": "127.0.0.1",
      "port": $ss_server_port,
      "warnings": [],
      "config": {
        "type": "shadowsocks",
        "cipher": "none",
        "password": "unused",
        "udp": true,
        "udp_over_tcp": false,
        "udp_over_tcp_version": 1,
        "client_fingerprint": null,
        "plugin": {
          "name": "obfs",
          "opts": {
            "mode": "tls",
            "host": "front.example"
          }
        }
      }
    },
    {
      "ordinal": 7,
      "name": "local shadowsocks none v2ray http upgrade",
      "protocol": "shadowsocks",
      "server": "127.0.0.1",
      "port": $ss_server_port,
      "warnings": [],
      "config": {
        "type": "shadowsocks",
        "cipher": "none",
        "password": "unused",
        "udp": true,
        "udp_over_tcp": false,
        "udp_over_tcp_version": 1,
        "client_fingerprint": null,
        "plugin": {
          "name": "v2ray-plugin",
          "opts": {
            "mode": "websocket",
            "host": "front.example",
            "path": "/upgrade",
            "mux": "false",
            "v2ray-http-upgrade": "true"
          }
        }
      }
    },
    {
      "ordinal": 8,
      "name": "local shadowsocks none v2ray websocket",
      "protocol": "shadowsocks",
      "server": "127.0.0.1",
      "port": $ss_server_port,
      "warnings": [],
      "config": {
        "type": "shadowsocks",
        "cipher": "none",
        "password": "unused",
        "udp": true,
        "udp_over_tcp": false,
        "udp_over_tcp_version": 1,
        "client_fingerprint": null,
        "plugin": {
          "name": "v2ray-plugin",
          "opts": {
            "mode": "websocket",
            "host": "front.example",
            "path": "/ws",
            "mux": "false"
          }
        }
      }
    },
    {
      "ordinal": 9,
      "name": "local shadowsocks none v2ray mux",
      "protocol": "shadowsocks",
      "server": "127.0.0.1",
      "port": $ss_server_port,
      "warnings": [],
      "config": {
        "type": "shadowsocks",
        "cipher": "none",
        "password": "unused",
        "udp": true,
        "udp_over_tcp": false,
        "udp_over_tcp_version": 1,
        "client_fingerprint": null,
        "plugin": {
          "name": "v2ray-plugin",
          "opts": {
            "mode": "websocket",
            "host": "front.example",
            "path": "/mux"
          }
        }
      }
    },
    {
      "ordinal": 10,
      "name": "local shadowsocks none gost websocket",
      "protocol": "shadowsocks",
      "server": "127.0.0.1",
      "port": $ss_server_port,
      "warnings": [],
      "config": {
        "type": "shadowsocks",
        "cipher": "none",
        "password": "unused",
        "udp": true,
        "udp_over_tcp": false,
        "udp_over_tcp_version": 1,
        "client_fingerprint": null,
        "plugin": {
          "name": "gost-plugin",
          "opts": {
            "mode": "websocket",
            "host": "front.example",
            "path": "/gost",
            "mux": "false"
          }
        }
      }
    },
    {
      "ordinal": 11,
      "name": "local shadowsocks none v2ray early data",
      "protocol": "shadowsocks",
      "server": "127.0.0.1",
      "port": $ss_server_port,
      "warnings": [],
      "config": {
        "type": "shadowsocks",
        "cipher": "none",
        "password": "unused",
        "udp": true,
        "udp_over_tcp": false,
        "udp_over_tcp_version": 1,
        "client_fingerprint": null,
        "plugin": {
          "name": "v2ray-plugin",
          "opts": {
            "mode": "websocket",
            "host": "front.example",
            "path": "/ed?ed=8&trace=1",
            "mux": "false"
          }
        }
      }
    }
  ],
  "warnings": []
}
EOF

(
  cd "$tmpdir"
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" daemon >"$daemon_log" 2>&1
) &
daemon_pid=$!

for _ in {1..100}; do
  if [[ -S "$socket" ]]; then
    break
  fi
  if ! kill -0 "$daemon_pid" >/dev/null 2>&1; then
    echo "Burrow daemon exited before creating socket" >&2
    cat "$daemon_log" >&2 || true
    exit 1
  fi
  sleep 0.05
done

if [[ ! -S "$socket" ]]; then
  echo "timed out waiting for Burrow daemon socket" >&2
  cat "$daemon_log" >&2 || true
  exit 1
fi

(
  cd "$tmpdir"
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" network-add 1 3 "$payload"
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" tunnel-config
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-tcp-probe \
    --dns-name selftest.invalid \
    --remote 1.1.1.1:80 \
    --timeout-ms 8000
)

wait "$server_pid"
server_pid=""

echo
echo "Local Trojan server observed:"
cat "$server_result"
echo

(
  cd "$tmpdir"
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-subscription-select 1 2
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" tunnel-config
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-tcp-probe \
    --dns-name selftest.invalid \
    --remote 1.1.1.1:80 \
    --timeout-ms 8000
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-udp-echo \
    --message burrow-shadowsocks-udp-selftest \
    --timeout-ms 8000 \
    9.9.9.9:5353
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-subscription-select 1 3
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" tunnel-config
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-udp-echo \
    --message burrow-shadowsocks-uot-selftest \
    --timeout-ms 8000 \
    9.9.9.9:5354
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-subscription-select 1 4
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" tunnel-config
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-udp-echo \
    --message burrow-shadowsocks-uot-v2-selftest \
    --timeout-ms 8000 \
    9.9.9.9:5355
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-subscription-select 1 5
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" tunnel-config
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-tcp-probe \
    --dns-name selftest.invalid \
    --remote 1.1.1.1:80 \
    --timeout-ms 8000
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-subscription-select 1 6
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" tunnel-config
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-tcp-probe \
    --dns-name selftest.invalid \
    --remote 1.1.1.1:80 \
    --timeout-ms 8000
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-subscription-select 1 7
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" tunnel-config
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-tcp-probe \
    --dns-name selftest.invalid \
    --remote 1.1.1.1:80 \
    --timeout-ms 8000
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-subscription-select 1 8
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" tunnel-config
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-tcp-probe \
    --dns-name selftest.invalid \
    --remote 1.1.1.1:80 \
    --timeout-ms 8000
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-subscription-select 1 9
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" tunnel-config
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-tcp-probe \
    --dns-name selftest.invalid \
    --remote 1.1.1.1:80 \
    --timeout-ms 8000
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-subscription-select 1 10
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" tunnel-config
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-tcp-probe \
    --dns-name selftest.invalid \
    --remote 1.1.1.1:80 \
    --timeout-ms 8000
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-subscription-select 1 11
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" tunnel-config
  BURROW_SOCKET_PATH="$socket" "$burrow_bin" proxy-tcp-probe \
    --dns-name selftest.invalid \
    --remote 1.1.1.1:80 \
    --timeout-ms 8000
)

wait "$ss_server_pid"
ss_server_pid=""

echo
echo "Local Shadowsocks server observed:"
cat "$ss_server_result"
echo

uv run python - "$server_result" "$ss_server_result" <<'PY'
from __future__ import annotations

import json
import sys

trojan_path, shadowsocks_path = sys.argv[1:]


def load(path: str) -> dict[str, object]:
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


def expect(value: object, expected: object, label: str) -> None:
    if value != expected:
        raise AssertionError(f"{label}: expected {expected!r}, got {value!r}")


def expect_startswith(value: object, prefix: str, label: str) -> None:
    if not isinstance(value, str) or not value.startswith(prefix):
        raise AssertionError(f"{label}: expected prefix {prefix!r}, got {value!r}")


def expect_target(result: dict[str, object], key: str, host: str, port: int) -> None:
    target = result.get(key)
    if not isinstance(target, dict):
        raise AssertionError(f"{key}: missing target object")
    expect(target.get("host"), host, f"{key}.host")
    expect(target.get("port"), port, f"{key}.port")


trojan = load(trojan_path)
expect(trojan.get("password_hash_ok"), True, "trojan password hash")
expect(trojan.get("command"), 1, "trojan tcp command")
expect(trojan.get("request_first_line"), "GET / HTTP/1.1", "trojan request")
expect_target(trojan, "target", "selftest.invalid", 80)

ss = load(shadowsocks_path)
for key in [
    "tcp_target",
    "obfs_http_target",
    "obfs_tls_target",
    "v2ray_http_upgrade_target",
    "v2ray_websocket_target",
    "v2ray_mux_target",
    "gost_websocket_target",
    "v2ray_early_data_target",
]:
    expect_target(ss, key, "selftest.invalid", 80)

expect(ss.get("udp_payload"), "burrow-shadowsocks-udp-selftest", "native UDP payload")
expect_target(ss, "udp_target", "9.9.9.9", 5353)
expect(ss.get("uot_payload"), "burrow-shadowsocks-uot-selftest", "legacy UOT payload")
expect_target(ss, "uot_magic_target", "sp.udp-over-tcp.arpa", 0)
expect_target(ss, "uot_target", "9.9.9.9", 5354)
expect(ss.get("uot_v2_payload"), "burrow-shadowsocks-uot-v2-selftest", "v2 UOT payload")
expect_target(ss, "uot_v2_magic_target", "sp.v2.udp-over-tcp.arpa", 0)
expect_target(ss, "uot_v2_target", "9.9.9.9", 5355)
expect_startswith(ss.get("obfs_http_host"), "front.example:", "simple-obfs HTTP host")
expect(ss.get("obfs_tls_server_name"), "front.example", "simple-obfs TLS SNI")
expect(ss.get("v2ray_http_upgrade_has_websocket_key"), False, "v2ray raw upgrade key")
expect(ss.get("v2ray_websocket_has_websocket_key"), True, "v2ray websocket key")
expect(ss.get("v2ray_mux_has_websocket_key"), True, "v2ray mux websocket key")
expect(ss.get("gost_websocket_has_websocket_key"), True, "gost websocket key")
expect(ss.get("v2ray_early_data_first_line"), "GET /ed?trace=1 HTTP/1.1", "early-data path")
expect(ss.get("v2ray_early_data_len"), 8, "early-data length")

print("Proxy self-test assertions passed")
PY