利用免费 Python 容器部署 SOCK5 和 HTTP
步骤
注册一个 KataBump 账号，账号还是比较好注册的，随便一个邮箱就可以，注册链接：https://dashboard.katabump.com/auth/login
登入 KataBump 平台，点击 Order 按钮领取免费的 Python 容器，Name 随便填，Type 勾选
Python，Resources 默认就行(第一个才是免费的，其余的也创建不了）然后过人机验证，点确认。机器创建完，首页有个 Your servers，点 Action 下面的红色 see 就能找到你的机器信息，然后获取你的翼龙面板，点击有🚀图标 Access server 就能找到翼龙面板的登入网址和账号密码。
登入翼龙面板。点击 Files 进入文件，点击➕的 NEW Files 添加启动文件（没有启动文件无法启动容器）启动文件类型选 Python，文件名
app.py（再添加一个requirements.txt的空文件）就可以按绿色按钮启动容器了。正式使用更换文件的账号密码，默认 http 和 sock5 账号都是 admin，密码：123456
代码
HTTP
import socket
import threading
import select
import base64
import os
PORT = int(os.environ.get('SERVER_PORT', os.environ.get('PORT', 20006)))
# ================= 账号密码配置 =================
AUTH_USER = "admin"
AUTH_PASS = "123456"
# ===============================================
def handle_client(client_sock, client_addr):
print(f"[+] 收到客户端连接: {client_addr}")
remote_sock = None
try:
# 读取 HTTP 请求头
request_data = b""
while b"\r\n\r\n" not in request_data:
chunk = client_sock.recv(4096)
if not chunk:
break
request_data += chunk
if len(request_data) > 8192:
break
if not request_data:
client_sock.close()
return
lines = request_data.split(b"\r\n")
first_line = lines[0].decode('utf-8', errors='ignore')
parts = first_line.split(" ")
if len(parts) < 2:
client_sock.close()
return
method, url = parts[0], parts[1]
# 检查 Proxy-Authorization 认证
authed = False
expected_auth = base64.b64encode(f"{AUTH_USER}:{AUTH_PASS}".encode()).decode()
for line in lines:
if line.lower().startswith(b"proxy-authorization: basic "):
token = line.split(b" ")[2].decode()
if token == expected_auth:
authed = True
break
if not authed:
# 认证失败，返回 407 要求输入账号密码
challenge = (
"HTTP/1.1 407 Proxy Authentication Required\r\n"
"Proxy-Authenticate: Basic realm=\"Proxy Required\"\r\n"
"Content-Length: 0\r\n\r\n"
)
client_sock.sendall(challenge.encode())
client_sock.close()
return
# 解析目标地址
if method == "CONNECT":
# HTTPS / 隧道代理
host_port = url
if ":" in host_port:
target_host, target_port = host_port.split(":")
target_port = int(target_port)
else:
target_host = host_port
target_port = 443
# 回复客户端 CONNECT 建立成功
client_sock.sendall(b"HTTP/1.1 200 Connection Established\r\n\r\n")
else:
# 普通 HTTP 代理
if url.startswith("http://"):
url = url[7:]
if "/" in url:
host_port, _ = url.split("/", 1)
else:
host_port = url
if ":" in host_port:
target_host, target_port = host_port.split(":")
target_port = int(target_port)
else:
target_host = host_port
target_port = 80
# 把剩余的请求体转发给目标
# 简化处理：对于普通 HTTP，把首行 URL 里的协议域名去掉
# 实际生产中可以更完整，这里重点保证主干通畅
print(f"[->] 正在转发到目标: {target_host}:{target_port}")
remote_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
remote_sock.connect((target_host, target_port))
# 如果是普通 HTTP 且带了剩余数据，可以转发，如果是 CONNECT 则直接开始双向流转发
if method != "CONNECT":
# 重新构造请求发给远端
new_first_line = f"{method} {url} HTTP/1.1\r\n"
# 过滤掉 Proxy 相关的头
forward_data = new_first_line.encode() + b"\r\n".join([l for l in lines[1:] if not l.lower().startswith(b"proxy-")])
remote_sock.sendall(forward_data)
# 双向流量转发
sockets = [client_sock, remote_sock]
while True:
r, w, e = select.select(sockets, [], [], 60)
if not r:
break
if client_sock in r:
data = client_sock.recv(4096)
if not data:
break
remote_sock.sendall(data)
elif remote_sock in r:
data = remote_sock.recv(4096)
if not data:
break
client_sock.sendall(data)
except Exception as e:
print(f"[!] 异常: {e}")
finally:
try:
client_sock.close()
except:
pass
if remote_sock:
try:
remote_sock.close()
except:
pass
def main():
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("0.0.0.0", PORT))
server.listen(128)
print(f"====== HTTP 代理服务已启动，监听端口 {PORT} ======")
while True:
client_sock, addr = server.accept()
threading.Thread(target=handle_client, args=(client_sock, addr), daemon=True).start()
if __name__ == '__main__':
main()SOCKS5
import socket
import threading
import select
import os
PORT = int(os.environ.get('SERVER_PORT', os.environ.get('PORT', 20006)))
# ================= 账号密码配置 =================
AUTH_USER = "admin"
AUTH_PASS = "123456"
# ===============================================
def handle_client(client_sock, client_addr):
print(f"[+] 收到客户端连接: {client_addr}")
remote_sock = None
try:
# 1. SOCKS5 握手阶段 1: 协商认证方式
header = client_sock.recv(2)
if len(header) < 2 or header[0] != 0x05:
client_sock.close()
return
nmethods = header[1]
methods = client_sock.recv(nmethods)
# 要求使用用户名密码认证 (0x02)，如果客户端不支持则退化为无认证 (0x00)
# 这里为了安全，强制要求 0x02 认证，或者允许 0x00
if 0x02 in methods:
# 告诉客户端：选择用户名密码认证
client_sock.sendall(b"\x05\x02")
# 2. SOCKS5 用户名密码认证子协议 (RFC 1929)
sub_ver = client_sock.recv(1)
if not sub_ver or sub_ver[0] != 0x01:
client_sock.close()
return
ulen = client_sock.recv(1)[0]
user = client_sock.recv(ulen).decode('utf-8', errors='ignore')
plen = client_sock.recv(1)[0]
passwd = client_sock.recv(plen).decode('utf-8', errors='ignore')
if user == AUTH_USER and passwd == AUTH_PASS:
# 认证成功
client_sock.sendall(b"\x01\x00")
else:
# 认证失败
client_sock.sendall(b"\x01\x01")
client_sock.close()
print(f"[-] SOCKS5 认证失败: user={user}")
return
else:
# 不要求认证直接过 (根据需要开启)
client_sock.sendall(b"\x05\x00")
# 3. SOCKS5 请求阶段：解析目标地址
req = client_sock.recv(4)
if len(req) < 4 or req[0] != 0x05 or req[1] != 0x01: # 0x01 代表 CONNECT 请求
client_sock.close()
return
atyp = req[3]
if atyp == 0x01: # IPv4
target_host = socket.inet_ntoa(client_sock.recv(4))
elif atyp == 0x03: # 域名
addr_len = client_sock.recv(1)[0]
target_host = client_sock.recv(addr_len).decode('utf-8', errors='ignore')
elif atyp == 0x04: # IPv6
target_host = socket.inet_ntop(socket.AF_INET6, client_sock.recv(16))
else:
client_sock.close()
return
port_bytes = client_sock.recv(2)
target_port = int.from_bytes(port_bytes, 'big')
print(f"[->] 正在通过 SOCKS5 转发到目标: {target_host}:{target_port}")
# 连接远端目标服务器
remote_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
remote_sock.connect((target_host, target_port))
# 4. 回复客户端 SOCKS5 连接建立成功
# 0x00 表示成功，后面跟绑定地址和端口（这里简单返回 0 即可）
reply = b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00"
client_sock.sendall(reply)
# 5. 双向流量转发 (TCP 管道)
sockets = [client_sock, remote_sock]
while True:
r, w, e = select.select(sockets, [], [], 60)
if not r:
break
if client_sock in r:
data = client_sock.recv(4096)
if not data:
break
remote_sock.sendall(data)
elif remote_sock in r:
data = remote_sock.recv(4096)
if not data:
break
client_sock.sendall(data)
except Exception as e:
print(f"[!] 异常: {e}")
finally:
try:
client_sock.close()
except:
pass
if remote_sock:
try:
remote_sock.close()
except:
pass
def main():
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("0.0.0.0", PORT))
server.listen(128)
print(f"====== SOCKS5 代理服务已启动，监听端口 {PORT} ======")
while True:
client_sock, addr = server.accept()
threading.Thread(target=handle_client, args=(client_sock, addr), daemon=True).start()
if __name__ == '__main__':
main()Vless + Reality
"""
High-performance KataBump bootstrap for rust-reality v1.6.1.
Design:
- ONLY the official static x86_64 MUSL release is supported.
- NO gcompat, glibc fallback, package manager, source build, cargo or rustc.
- VLESS + REALITY + xtls-rprx-vision, standalone/direct.
- runtime.profile = dedicated
- runtime.tuning.mode = startup
- runtime.tuning.objective = throughput
- IPv4-only for the diagnosed KataBump environment.
- Logging is disabled by default for minimum steady-state overhead.
- Python calls execve(), so Python is not resident after startup.
Required KataBump environment:
SERVER_IP
SERVER_PORT
Optional environment:
RR_SERVER_ADDRESS=example.com # client link address instead of SERVER_IP
RR_PORT=20202 # override SERVER_PORT
RR_SNI=www.microsoft.com # force one REALITY cover
RR_NODE_NAME=KataBump-rust-reality # v2rayN display name
RR_LOG_OUTPUT=stderr # default: none
RR_REGENERATE=1 # explicitly discard old node identity
"""
from __future__ import annotations
import contextlib
import fcntl
import hashlib
import ipaddress
import json
import os
import re
import shutil
import struct
import subprocess
import sys
import tarfile
import tempfile
import time
import urllib.parse
import urllib.request
from pathlib import Path
# ---------------------------------------------------------------------------
# Pinned release
# ---------------------------------------------------------------------------
VERSION = "1.6.1"
TAG = "v1.6.1"
REPOSITORY = "jacek4yang/rust-reality"
ASSET = "rust-reality-v1.6.1-linux-x86_64-musl.tar.gz"
RELEASE_BASE = (
f"https://github.com/{REPOSITORY}/releases/download/{TAG}"
)
ASSET_URL = f"{RELEASE_BASE}/{ASSET}"
SHA256SUMS_URL = f"{RELEASE_BASE}/SHA256SUMS"
# Candidates are intentionally ordinary large TLS 1.3 sites. rust-reality's
# own probe-dest is authoritative; incompatible candidates are ignored.
DEFAULT_SNI_CANDIDATES = (
"www.microsoft.com",
"www.apple.com",
"www.cloudflare.com",
"www.amazon.com",
"www.ibm.com",
"www.nvidia.com",
)
HOME = Path(os.environ.get("HOME", "/home/container"))
STATE_DIR = HOME / ".rust-reality-node" / TAG
BINARY = STATE_DIR / "rust-reality"
CONFIG = STATE_DIR / "config.json"
CLIENT_META = STATE_DIR / "client.json"
CLIENT_LINK = STATE_DIR / "client-link.txt"
LOCK_FILE = STATE_DIR / ".bootstrap.lock"
MAX_DOWNLOAD_BYTES = 128 * 1024 * 1024
HTTP_TIMEOUT = 30
PUBLIC_KEY_RE = re.compile(
r"REALITY public key for the client:\s*([A-Za-z0-9_-]{40,64})"
)
class BootstrapError(RuntimeError):
pass
def log(message: str) -> None:
print(
f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {message}",
flush=True,
)
def fatal(message: str) -> "NoReturn":
raise BootstrapError(message)
# ---------------------------------------------------------------------------
# Small robust primitives
# ---------------------------------------------------------------------------
def atomic_write(path: Path, data: bytes, mode: int = 0o600) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_name = tempfile.mkstemp(
prefix=f".{path.name}.",
suffix=".tmp",
dir=path.parent,
)
tmp = Path(tmp_name)
try:
with os.fdopen(fd, "wb") as stream:
stream.write(data)
stream.flush()
os.fsync(stream.fileno())
os.chmod(tmp, mode)
os.replace(tmp, path)
# Best-effort durability of the rename itself.
try:
directory_fd = os.open(path.parent, os.O_RDONLY)
except OSError:
directory_fd = None
if directory_fd is not None:
try:
os.fsync(directory_fd)
finally:
os.close(directory_fd)
finally:
with contextlib.suppress(FileNotFoundError):
tmp.unlink()
def run(
args: list[str],
*,
timeout: float = 15.0,
) -> subprocess.CompletedProcess[str]:
try:
result = subprocess.run(
args,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
timeout=timeout,
)
except subprocess.TimeoutExpired as exc:
raise BootstrapError(
f"command timed out after {timeout}s: {' '.join(args)}"
) from exc
except OSError as exc:
raise BootstrapError(
f"cannot execute {' '.join(args)}: {exc}"
) from exc
if result.returncode != 0:
detail = (result.stderr or result.stdout).strip()
if len(detail) > 4000:
detail = detail[-4000:]
raise BootstrapError(
f"command failed ({result.returncode}): {' '.join(args)}"
+ (f"\n{detail}" if detail else "")
)
return result
def request(url: str):
return urllib.request.Request(
url,
headers={
"User-Agent": f"katabump-rust-reality/{VERSION}",
"Accept": "*/*",
},
)
def fetch_text(url: str, max_bytes: int = 1024 * 1024) -> str:
last_error: Exception | None = None
for attempt in range(3):
try:
with urllib.request.urlopen(
request(url),
timeout=HTTP_TIMEOUT,
) as response:
data = response.read(max_bytes + 1)
if len(data) > max_bytes:
raise BootstrapError(
f"response exceeds {max_bytes} bytes: {url}"
)
return data.decode("utf-8")
except Exception as exc:
last_error = exc
if attempt != 2:
time.sleep(1 << attempt)
raise BootstrapError(
f"failed to fetch {url}: {last_error}"
)
def download(url: str, destination: Path) -> None:
last_error: Exception | None = None
for attempt in range(3):
partial = destination.with_name(destination.name + ".part")
with contextlib.suppress(FileNotFoundError):
partial.unlink()
try:
total = 0
with urllib.request.urlopen(
request(url),
timeout=HTTP_TIMEOUT,
) as source, open(partial, "wb") as target:
while True:
chunk = source.read(1024 * 1024)
if not chunk:
break
total += len(chunk)
if total > MAX_DOWNLOAD_BYTES:
raise BootstrapError(
f"download exceeds {MAX_DOWNLOAD_BYTES} bytes"
)
target.write(chunk)
target.flush()
os.fsync(target.fileno())
if total == 0:
raise BootstrapError("download returned an empty file")
os.replace(partial, destination)
return
except Exception as exc:
last_error = exc
with contextlib.suppress(FileNotFoundError):
partial.unlink()
if attempt != 2:
time.sleep(1 << attempt)
raise BootstrapError(
f"failed to download {url}: {last_error}"
)
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with open(path, "rb") as stream:
for chunk in iter(
lambda: stream.read(1024 * 1024),
b"",
):
digest.update(chunk)
return digest.hexdigest()
# ---------------------------------------------------------------------------
# MUSL release installation
# ---------------------------------------------------------------------------
def release_archive_sha256() -> str:
sums = fetch_text(SHA256SUMS_URL)
for raw_line in sums.splitlines():
fields = raw_line.strip().split()
if len(fields) < 2:
continue
filename = fields[-1].lstrip("*")
if filename != ASSET:
continue
digest = fields[0].lower()
if not re.fullmatch(r"[0-9a-f]{64}", digest):
raise BootstrapError(
f"invalid checksum for {ASSET} in SHA256SUMS"
)
return digest
raise BootstrapError(
f"{ASSET} is not listed in the pinned release SHA256SUMS"
)
def verify_static_x86_64_elf(path: Path) -> None:
"""
Minimal ELF64 parser:
- ELF64
- little-endian
- x86_64 (EM_X86_64 = 62)
- no PT_INTERP
No external readelf/file dependency.
"""
with open(path, "rb") as stream:
header = stream.read(64)
if len(header) < 64 or header[:4] != b"\x7fELF":
raise BootstrapError("release binary is not ELF")
# EI_CLASS=2: ELF64, EI_DATA=1: little endian.
if header[4] != 2 or header[5] != 1:
raise BootstrapError(
"release binary is not little-endian ELF64"
)
(
_e_type,
e_machine,
_e_version,
_e_entry,
e_phoff,
_e_shoff,
_e_flags,
_e_ehsize,
e_phentsize,
e_phnum,
_e_shentsize,
_e_shnum,
_e_shstrndx,
) = struct.unpack_from("<HHIQQQIHHHHHH", header, 16)
if e_machine != 62:
raise BootstrapError(
f"release binary architecture is not x86_64 "
f"(e_machine={e_machine})"
)
if e_phentsize < 56 or e_phnum > 4096:
raise BootstrapError("invalid ELF program-header table")
file_size = path.stat().st_size
ph_end = e_phoff + e_phentsize * e_phnum
if e_phoff > file_size or ph_end > file_size:
raise BootstrapError(
"ELF program-header table is outside the file"
)
stream.seek(e_phoff)
# PT_INTERP = 3. A fully static MUSL release must not need one.
for _ in range(e_phnum):
ph = stream.read(e_phentsize)
if len(ph) != e_phentsize:
raise BootstrapError(
"truncated ELF program-header table"
)
p_type = struct.unpack_from("<I", ph, 0)[0]
if p_type == 3:
raise BootstrapError(
"release binary contains PT_INTERP; "
"expected the fully static MUSL asset"
)
def binary_is_usable() -> bool:
if not BINARY.is_file():
return False
try:
verify_static_x86_64_elf(BINARY)
os.chmod(BINARY, 0o755)
version = run(
[str(BINARY), "--version"],
timeout=5,
).stdout.strip()
return version == f"rust-reality {VERSION}"
except Exception:
return False
def install_binary() -> None:
if binary_is_usable():
return
archive = STATE_DIR / ASSET
log(
f"installing rust-reality {VERSION} "
f"(static x86_64 MUSL)"
)
expected_sha256 = release_archive_sha256()
download(ASSET_URL, archive)
actual_sha256 = sha256_file(archive)
if actual_sha256 != expected_sha256:
with contextlib.suppress(FileNotFoundError):
archive.unlink()
raise BootstrapError(
"release archive SHA256 mismatch: "
f"expected={expected_sha256} actual={actual_sha256}"
)
with tarfile.open(archive, "r:gz") as tar:
candidates = [
member
for member in tar.getmembers()
if member.isfile()
and Path(member.name).name == "rust-reality"
]
if len(candidates) != 1:
raise BootstrapError(
"release archive must contain exactly one "
"rust-reality executable"
)
member = candidates[0]
if not 0 < member.size <= MAX_DOWNLOAD_BYTES:
raise BootstrapError(
f"invalid rust-reality size: {member.size}"
)
source = tar.extractfile(member)
if source is None:
raise BootstrapError(
"cannot read rust-reality from release archive"
)
fd, tmp_name = tempfile.mkstemp(
prefix=".rust-reality.",
suffix=".tmp",
dir=STATE_DIR,
)
tmp = Path(tmp_name)
try:
with os.fdopen(fd, "wb") as target:
shutil.copyfileobj(
source,
target,
length=1024 * 1024,
)
target.flush()
os.fsync(target.fileno())
os.chmod(tmp, 0o755)
verify_static_x86_64_elf(tmp)
os.replace(tmp, BINARY)
finally:
with contextlib.suppress(FileNotFoundError):
tmp.unlink()
with contextlib.suppress(FileNotFoundError):
archive.unlink()
if not binary_is_usable():
raise BootstrapError(
"the pinned static MUSL rust-reality binary "
"cannot execute in this container"
)
# ---------------------------------------------------------------------------
# KataBump environment
# ---------------------------------------------------------------------------
def public_port() -> int:
raw = (
os.environ.get("RR_PORT")
or os.environ.get("SERVER_PORT")
)
if not raw:
raise BootstrapError(
"SERVER_PORT is missing; set RR_PORT explicitly"
)
try:
value = int(raw)
except ValueError as exc:
raise BootstrapError(
f"invalid port: {raw!r}"
) from exc
if not 1 <= value <= 65535:
raise BootstrapError(
f"port outside 1..65535: {value}"
)
return value
def public_host() -> str:
value = (
os.environ.get("RR_SERVER_ADDRESS")
or os.environ.get("SERVER_IP")
or ""
).strip()
value = value.strip("[]")
if not value:
raise BootstrapError(
"SERVER_IP is missing; set RR_SERVER_ADDRESS"
)
return value
# ---------------------------------------------------------------------------
# REALITY SNI selection
# ---------------------------------------------------------------------------
def probe_sni_once(sni: str) -> int | None:
try:
result = run(
[
str(BINARY),
"probe-dest",
"--target",
f"{sni}:443",
"--server-name",
sni,
"--timeout-ms",
"4000",
],
timeout=6,
)
report = json.loads(result.stdout)
if report.get("compatible") is not True:
return None
return int(report["totalMillis"])
except Exception:
return None
def select_sni() -> str:
forced = os.environ.get("RR_SNI", "").strip()
if forced:
latency = probe_sni_once(forced)
if latency is None:
raise BootstrapError(
f"RR_SNI={forced!r} failed rust-reality probe-dest"
)
log(f"using forced REALITY SNI: {forced} ({latency} ms)")
return forced
log("probing REALITY cover candidates")
results: list[tuple[int, str]] = []
for sni in DEFAULT_SNI_CANDIDATES:
# Two attempts reduce one-off DNS/TCP noise. We use the better
# successful result so one cgroup-throttled sample does not poison
# selection.
samples = [
value
for value in (
probe_sni_once(sni),
probe_sni_once(sni),
)
if value is not None
]
if not samples:
log(f" {sni}: incompatible/unreachable")
continue
latency = min(samples)
results.append((latency, sni))
log(f" {sni}: {latency} ms")
if not results:
raise BootstrapError(
"no compatible REALITY cover found; "
"set RR_SNI to a reachable TLS 1.3 hostname"
)
latency, sni = min(results)
log(f"selected REALITY SNI: {sni} ({latency} ms)")
return sni
# ---------------------------------------------------------------------------
# rust-reality configuration
# ---------------------------------------------------------------------------
def apply_performance_profile(
config: dict,
port: int,
) -> dict:
"""
Keep the generated standalone/direct topology and let rust-reality derive
all numeric resource limits from the real cgroup.
We specify intent, not hand-tuned magic numbers.
"""
log_output = (
os.environ.get("RR_LOG_OUTPUT", "none")
.strip()
.lower()
)
if log_output not in {"none", "stderr"}:
raise BootstrapError(
"RR_LOG_OUTPUT must be 'none' or 'stderr'"
)
config["log"] = {
"level": "error",
"output": log_output,
}
network = config.setdefault("network", {})
dial = network.setdefault("dial", {})
dial["mode"] = "ipv4Only"
runtime = config.setdefault("runtime", {})
runtime["profile"] = "dedicated"
runtime["tuning"] = {
"mode": "startup",
"objective": "throughput",
}
try:
inbound = config["inbounds"][0]
except (KeyError, IndexError, TypeError) as exc:
raise BootstrapError(
"generated configuration contains no public inbound"
) from exc
inbound["port"] = port
inbound["listen"] = {
"mode": "ipv4Only",
}
return config
def validate_config_file(
path: Path,
*,
self_test: bool,
) -> None:
run(
[
str(BINARY),
"check",
"--config",
str(path),
],
timeout=10,
)
if self_test:
run(
[
str(BINARY),
"self-test",
"--config",
str(path),
],
timeout=25,
)
def generate_node(port: int) -> dict:
sni = select_sni()
generated = run(
[
str(BINARY),
"config",
"generate",
"standalone",
"--listen",
"0.0.0.0",
"--port",
str(port),
"--target",
f"{sni}:443",
"--server-name",
sni,
],
timeout=10,
)
try:
config = json.loads(generated.stdout)
config = apply_performance_profile(config, port)
user = config["inbounds"][0]["settings"]["clients"][0]
uuid = str(user["id"])
short_ids = user["shortIds"]
if not isinstance(short_ids, list) or not short_ids:
raise ValueError("empty shortIds")
short_id = str(short_ids[0])
except Exception as exc:
raise BootstrapError(
"unexpected rust-reality generated configuration"
) from exc
public_key_match = PUBLIC_KEY_RE.search(generated.stderr)
if public_key_match is None:
raise BootstrapError(
"rust-reality did not return the REALITY public key"
)
public_key = public_key_match.group(1)
config_data = (
json.dumps(
config,
ensure_ascii=False,
indent=2,
)
+ "\n"
).encode()
fd, tmp_name = tempfile.mkstemp(
prefix=".config.",
suffix=".json",
dir=STATE_DIR,
)
tmp = Path(tmp_name)
try:
with os.fdopen(fd, "wb") as stream:
stream.write(config_data)
stream.flush()
os.fsync(stream.fileno())
# Expensive validation is first-generation only.
validate_config_file(
tmp,
self_test=True,
)
finally:
with contextlib.suppress(FileNotFoundError):
tmp.unlink()
meta = {
"version": VERSION,
"uuid": uuid,
"shortId": short_id,
"publicKey": public_key,
"sni": sni,
}
# Client metadata first; CONFIG is the completion marker.
atomic_write(
CLIENT_META,
(
json.dumps(meta, ensure_ascii=False, indent=2)
+ "\n"
).encode(),
)
atomic_write(CONFIG, config_data)
return meta
def explicitly_regenerate_if_requested() -> None:
if os.environ.get("RR_REGENERATE") != "1":
return
log(
"RR_REGENERATE=1: deleting persisted node identity "
"(old client links will stop working)"
)
for path in (
CONFIG,
CLIENT_META,
CLIENT_LINK,
):
with contextlib.suppress(FileNotFoundError):
path.unlink()
def load_existing_node(port: int) -> dict | None:
if not CONFIG.exists():
return None
if not CLIENT_META.exists():
raise BootstrapError(
f"{CONFIG} exists but {CLIENT_META} is missing. "
"Set RR_REGENERATE=1 once to create a new identity."
)
try:
config = json.loads(CONFIG.read_text("utf-8"))
meta = json.loads(CLIENT_META.read_text("utf-8"))
except Exception as exc:
raise BootstrapError(
"persisted node state is unreadable. "
"Set RR_REGENERATE=1 once to rebuild it."
) from exc
# Reapply the current high-performance profile so an older bootstrap's
# persisted config cannot silently retain conservative settings.
config = apply_performance_profile(config, port)
config_data = (
json.dumps(
config,
ensure_ascii=False,
indent=2,
)
+ "\n"
).encode()
atomic_write(CONFIG, config_data)
# Startup validation is cheap and catches stale/corrupt state.
validate_config_file(
CONFIG,
self_test=False,
)
required_meta = {
"uuid",
"shortId",
"publicKey",
"sni",
}
if not required_meta <= set(meta):
raise BootstrapError(
"persisted client metadata is incomplete. "
"Set RR_REGENERATE=1 once to rebuild it."
)
return meta
def load_or_create_node(port: int) -> dict:
explicitly_regenerate_if_requested()
existing = load_existing_node(port)
if existing is not None:
return existing
return generate_node(port)
# ---------------------------------------------------------------------------
# v2rayN link
# ---------------------------------------------------------------------------
def vless_link(
meta: dict,
host: str,
port: int,
) -> str:
try:
parsed_ip = ipaddress.ip_address(host)
authority_host = (
f"[{host}]"
if parsed_ip.version == 6
else host
)
except ValueError:
authority_host = host
query = urllib.parse.urlencode(
{
"encryption": "none",
"flow": "xtls-rprx-vision",
"security": "reality",
"sni": meta["sni"],
"fp": "chrome",
"pbk": meta["publicKey"],
"sid": meta["shortId"],
"type": "tcp",
"headerType": "none",
},
safe="",
)
node_name = urllib.parse.quote(
os.environ.get(
"RR_NODE_NAME",
"KataBump-rust-reality",
),
safe="",
)
return (
f"vless://{meta['uuid']}@"
f"{authority_host}:{port}"
f"?{query}#{node_name}"
)
# ---------------------------------------------------------------------------
# Entrypoint
# ---------------------------------------------------------------------------
def main() -> None:
if sys.platform != "linux":
raise BootstrapError("Linux is required")
if os.uname().machine != "x86_64":
raise BootstrapError(
"this bootstrap requires x86_64"
)
STATE_DIR.mkdir(
parents=True,
exist_ok=True,
)
os.chmod(STATE_DIR, 0o700)
# Prevent duplicate panel starts racing persistent state.
with open(LOCK_FILE, "a+b") as lock:
fcntl.flock(
lock.fileno(),
fcntl.LOCK_EX,
)
port = public_port()
host = public_host()
install_binary()
meta = load_or_create_node(port)
link = vless_link(meta, host, port)
atomic_write(
CLIENT_LINK,
(link + "\n").encode(),
)
print()
print("=" * 78)
print(
f"rust-reality {VERSION} | "
"standalone/direct | static MUSL"
)
print("runtime : dedicated / startup / throughput")
print("network : IPv4-only")
print(
f"logging : "
f"{os.environ.get('RR_LOG_OUTPUT', 'none')}"
)
print(f"server : {host}:{port}")
print(f"SNI : {meta['sni']}")
print()
print("COPY THIS LINK INTO v2rayN:")
print()
print(link)
print("=" * 78)
print()
sys.stdout.flush()
sys.stderr.flush()
log(
"execve rust-reality; Python bootstrap is leaving memory"
)
environment = os.environ.copy()
environment.setdefault(
"RUST_BACKTRACE",
"0",
)
os.execve(
BINARY,
[
str(BINARY),
"serve",
"--config",
str(CONFIG),
],
environment,
)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
raise SystemExit(130)
except Exception as exc:
print(
f"[FATAL] {exc}",
file=sys.stderr,
flush=True,
)
raise SystemExit(1)续期
目前 KataBump 领取的免费容器需要 4 天续期一次
评论
隐私政策
你无需删除空行，直接评论以获取最佳展示效果
