116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import subprocess
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
from vpn_egressctl.policy import (
|
|
BandwidthPolicy,
|
|
DnsPolicy,
|
|
HealthcheckPolicy,
|
|
NetworkPolicy,
|
|
Policy,
|
|
RuntimePolicy,
|
|
SingBoxPolicy,
|
|
)
|
|
|
|
VERSION_OUTPUT = """sing-box version 1.14.0
|
|
|
|
Environment: go1.25.9 linux/amd64
|
|
Tags: with_quic,with_gvisor,with_utls
|
|
Revision: testrevision
|
|
CGO: enabled
|
|
"""
|
|
|
|
|
|
def make_policy(root: Path, *, health_url: str | None = "https://health.invalid/") -> Policy:
|
|
etc = root / "etc"
|
|
state = root / "state"
|
|
return Policy(
|
|
schema_version=1,
|
|
sing_box=SingBoxPolicy(
|
|
binary=str(root / "sing-box"),
|
|
config_path=str(etc / "config.json"),
|
|
service="sing-box.service",
|
|
required_version="1.14.0",
|
|
),
|
|
runtime=RuntimePolicy(
|
|
uri_path=str(etc / "hysteria2.uri"),
|
|
state_dir=str(state),
|
|
lock_path=str(root / "run" / "controller.lock"),
|
|
backup_keep=3,
|
|
),
|
|
network=NetworkPolicy(
|
|
upstream_interface="eth0",
|
|
vpn_lan_interface="eth1",
|
|
tun_name="tun-sb0",
|
|
tun_address="172.19.0.1/30",
|
|
mtu=1400,
|
|
route_exclude_address=("10.20.0.0/24", "10.30.0.0/24", "127.0.0.0/8"),
|
|
iproute2_table_index=2022,
|
|
iproute2_rule_index=9000,
|
|
auto_redirect_input_mark="0x2023",
|
|
auto_redirect_output_mark="0x2024",
|
|
auto_redirect_reset_mark="0x2025",
|
|
auto_redirect_nfqueue=100,
|
|
auto_redirect_fallback_rule_index=32768,
|
|
),
|
|
dns=DnsPolicy(
|
|
bootstrap_server="1.1.1.1",
|
|
bootstrap_port=53,
|
|
remote_server="1.1.1.1",
|
|
remote_port=443,
|
|
remote_path="/dns-query",
|
|
remote_tls_server_name="cloudflare-dns.com",
|
|
strategy="ipv4_only",
|
|
),
|
|
bandwidth=BandwidthPolicy(up_mbps=50, down_mbps=200),
|
|
healthcheck=HealthcheckPolicy(
|
|
url=health_url,
|
|
timeout_seconds=0.1,
|
|
settle_seconds=0,
|
|
expected_status=200,
|
|
body_contains="ip=",
|
|
),
|
|
)
|
|
|
|
|
|
class FakeResponse:
|
|
status = 200
|
|
|
|
def __init__(self, body: bytes = b"ip=203.0.113.10\n") -> None:
|
|
self.body = body
|
|
|
|
def __enter__(self) -> "FakeResponse":
|
|
return self
|
|
|
|
def __exit__(self, *args: object) -> None:
|
|
return None
|
|
|
|
def read(self, limit: int = -1) -> bytes:
|
|
return self.body[:limit]
|
|
|
|
def getcode(self) -> int:
|
|
return self.status
|
|
|
|
|
|
class FakeRunner:
|
|
def __init__(self, *, version: str = VERSION_OUTPUT, restart_results: list[int] | None = None, check_result: int = 0) -> None:
|
|
self.version = version
|
|
self.restart_results = list(restart_results or [0])
|
|
self.check_result = check_result
|
|
self.calls: list[list[str]] = []
|
|
|
|
def __call__(self, args: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
|
|
self.calls.append(list(args))
|
|
if len(args) > 1 and args[1] == "version":
|
|
return subprocess.CompletedProcess(args, 0, self.version, "")
|
|
if len(args) > 1 and args[1] == "check":
|
|
return subprocess.CompletedProcess(args, self.check_result, "", "")
|
|
if "restart" in args:
|
|
result = self.restart_results.pop(0) if self.restart_results else 0
|
|
return subprocess.CompletedProcess(args, result, "", "")
|
|
if "is-active" in args:
|
|
return subprocess.CompletedProcess(args, 0, "active\n", "")
|
|
return subprocess.CompletedProcess(args, 0, "", "")
|