prod/singbox_glue
5
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

release: prepare 0.2.0 for sing-box 1.14

Browse Source
This commit is contained in:
Crimson
2026-09-08 19:26:00 +05:00
parent 81ea89f7fa
commit 76dce5b47a
42 changed files with 1634 additions and 549 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/fixtures/sing-box-1.14.0-gecko.json
Vendored
+105
View File
@@ -0,0 +1,105 @@
{
"log": {
"level": "info",
"timestamp": true
},
"dns": {
"servers": [
{
"type": "udp",
"tag": "bootstrap-dns",
"server": "1.1.1.1",
"server_port": 53,
"bind_interface": "eth0"
},
{
"type": "https",
"tag": "remote-dns",
"server": "1.1.1.1",
"server_port": 443,
"path": "/dns-query",
"tls": {
"enabled": true,
"server_name": "cloudflare-dns.com"
},
"detour": "hy2-out"
}
],
"final": "remote-dns",
"strategy": "ipv4_only"
},
"inbounds": [
{
"type": "tun",
"tag": "tun-in",
"interface_name": "tun-sb0",
"address": [
"172.19.0.1/30"
],
"mtu": 1400,
"dns_mode": "disabled",
"auto_route": true,
"iproute2_table_index": 2022,
"iproute2_rule_index": 9000,
"auto_redirect": true,
"auto_redirect_input_mark": "0x2023",
"auto_redirect_output_mark": "0x2024",
"auto_redirect_reset_mark": "0x2025",
"auto_redirect_nfqueue": 100,
"auto_redirect_iproute2_fallback_rule_index": 32768,
"strict_route": true,
"stack": "mixed",
"route_exclude_address": [
"10.20.0.0/24",
"10.30.0.0/24",
"127.0.0.0/8"
]
}
],
"outbounds": [
{
"type": "hysteria2",
"tag": "hy2-out",
"server": "fi.api.withen.pro",
"up_mbps": 50,
"down_mbps": 200,
"password": "AUTH",
"tls": {
"enabled": true,
"server_name": "fi.api.withen.pro",
"insecure": false
},
"disable_chrome_parrot": false,
"bind_interface": "eth0",
"domain_resolver": {
"server": "bootstrap-dns",
"strategy": "ipv4_only"
},
"server_port": 443,
"obfs": {
"type": "gecko",
"password": "OBFS",
"min_packet_size": 512,
"max_packet_size": 1200
}
}
],
"route": {
"auto_detect_interface": true,
"default_domain_resolver": {
"server": "bootstrap-dns",
"strategy": "ipv4_only"
},
"final": "hy2-out",
"rules": [
{
"network": [
"tcp",
"udp"
],
"port": 53,
"action": "hijack-dns"
}
]
}
}
tests/helpers.py
+2 -2
View File
@@ -14,7 +14,7 @@ from vpn_egressctl.policy import (
SingBoxPolicy,
)
VERSION_OUTPUT = """sing-box version 1.13.19
VERSION_OUTPUT = """sing-box version 1.14.0
Environment: go1.25.9 linux/amd64
Tags: with_quic,with_gvisor,with_utls
@@ -32,7 +32,7 @@ def make_policy(root: Path, *, health_url: str | None = "https://health.invalid/
binary=str(root / "sing-box"),
config_path=str(etc / "config.json"),
service="sing-box.service",
required_version="1.13.19",
required_version="1.14.0",
),
runtime=RuntimePolicy(
uri_path=str(etc / "hysteria2.uri"),
tests/test_doctor.py
+25 -1
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import json
import hashlib
import stat
import subprocess
import tempfile
@@ -11,7 +12,9 @@ from types import SimpleNamespace
from unittest import mock
from vpn_egressctl.doctor import Doctor
from vpn_egressctl.renderer_1_13_19 import render_bytes
from vpn_egressctl.fsutil import atomic_write_json
from vpn_egressctl.metadata import build_config_metadata
from vpn_egressctl.renderer_1_14_0 import render_bytes
from vpn_egressctl.uri import parse_hysteria2_uri
from tests.helpers import FakeRunner, make_policy
@@ -90,6 +93,27 @@ class DoctorTests(unittest.TestCase):
self.assertEqual(doctor.checks[-1].level, "ERROR")
self.assertIn("regular file", doctor.checks[-1].message)
def test_invalid_last_good_metadata_is_reported(self) -> None:
with tempfile.TemporaryDirectory() as directory:
policy = self.prepare(directory, "hy2://auth@example.com")
state_dir = Path(policy.runtime.state_dir)
state_dir.mkdir(parents=True)
last_good = state_dir / "last-good.json"
last_good.write_bytes(b"{}\n")
atomic_write_json(
state_dir / "last-good.meta.json",
build_config_metadata(hashlib.sha256(b"different").hexdigest(), "1.14.0"),
)
doctor = Doctor(
policy,
runner=FakeRunner(),
resolver=lambda *args: [(None, None, None, None, ("8.8.4.4", 443))],
)
checks = doctor.run()
selected = [check for check in checks if check.name == "last-good-metadata"]
self.assertEqual(selected[0].level, "ERROR")
self.assertIn("checksum", selected[0].message)
if __name__ == "__main__":
unittest.main()
tests/test_installcheck.py
+78
View File
@@ -0,0 +1,78 @@
from __future__ import annotations
import hashlib
import json
import tempfile
import unittest
from pathlib import Path
from vpn_egressctl.errors import ValidationError
from vpn_egressctl.installcheck import verify_install_state
class InstallCheckTests(unittest.TestCase):
def test_empty_install_root_is_allowed(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
verify_install_state(root / "policy.json", root / "state", root / "config.json")
def test_legacy_policy_is_rejected_instead_of_migrated(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
raw = json.loads(Path("config/policy.json").read_text(encoding="utf-8"))
raw["sing_box"]["required_version"] = "1.13.19"
policy = root / "policy.json"
policy.write_text(json.dumps(raw), encoding="utf-8")
with self.assertRaisesRegex(ValidationError, "exactly 1.14.0"):
verify_install_state(policy, root / "state", root / "config.json")
def test_legacy_state_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
state_dir = root / "state"
state_dir.mkdir()
(state_dir / "state.json").write_text(
json.dumps({"controller_version": "0.1.0"}), encoding="utf-8"
)
with self.assertRaisesRegex(ValidationError, "another vpn-egressctl release"):
verify_install_state(root / "policy.json", state_dir, root / "config.json")
def test_unmanaged_existing_config_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
config = root / "config.json"
config.write_text("{}\n", encoding="utf-8")
with self.assertRaisesRegex(ValidationError, "unmanaged"):
verify_install_state(root / "policy.json", root / "state", config)
def test_orphaned_legacy_backup_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
state_dir = root / "state"
state_dir.mkdir()
(state_dir / "last-good.json").write_text("{}\n", encoding="utf-8")
with self.assertRaisesRegex(ValidationError, "Orphaned"):
verify_install_state(root / "policy.json", state_dir, root / "config.json")
def test_same_release_state_and_config_are_allowed(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
config = root / "config.json"
config.write_bytes(b"{}\n")
state_dir = root / "state"
state_dir.mkdir()
(state_dir / "state.json").write_text(
json.dumps(
{
"controller_version": "0.2.0",
"sing_box_version": "1.14.0",
"config_sha256": hashlib.sha256(b"{}\n").hexdigest(),
}
),
encoding="utf-8",
)
verify_install_state(root / "policy.json", state_dir, config)
if __name__ == "__main__":
unittest.main()
tests/test_metadata.py
+55
View File
@@ -0,0 +1,55 @@
from __future__ import annotations
import hashlib
import json
import tempfile
import unittest
from pathlib import Path
from vpn_egressctl.errors import ValidationError
from vpn_egressctl.metadata import build_config_metadata, load_config_metadata
class MetadataTests(unittest.TestCase):
def write(self, root: Path, value: dict) -> Path:
path = root / "config.meta.json"
path.write_text(json.dumps(value), encoding="utf-8")
return path
def test_current_metadata_is_accepted(self) -> None:
with tempfile.TemporaryDirectory() as directory:
data = b"{}\n"
digest = hashlib.sha256(data).hexdigest()
path = self.write(
Path(directory), build_config_metadata(digest, "1.14.0")
)
value = load_config_metadata(
path, config_sha256=digest, sing_box_version="1.14.0"
)
self.assertEqual(value["controller_version"], "0.2.0")
def test_wrong_engine_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
digest = hashlib.sha256(b"{}\n").hexdigest()
path = self.write(
Path(directory), build_config_metadata(digest, "1.13.19")
)
with self.assertRaisesRegex(ValidationError, "another sing-box"):
load_config_metadata(
path, config_sha256=digest, sing_box_version="1.14.0"
)
def test_unknown_keys_are_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
digest = hashlib.sha256(b"{}\n").hexdigest()
metadata = build_config_metadata(digest, "1.14.0")
metadata["future"] = True
path = self.write(Path(directory), metadata)
with self.assertRaisesRegex(ValidationError, "unsupported structure"):
load_config_metadata(
path, config_sha256=digest, sing_box_version="1.14.0"
)
if __name__ == "__main__":
unittest.main()
tests/test_packaging.py
+26
View File
@@ -8,6 +8,32 @@ ROOT = Path(__file__).resolve().parents[1]
class PackagingTests(unittest.TestCase):
def test_makefile_supports_src_layout(self) -> None:
makefile = (ROOT / "Makefile").read_text(encoding="utf-8")
self.assertIn("PYTHONPATH=src:.", makefile)
def test_release_metadata_is_mpl_2_and_version_0_2(self) -> None:
pyproject = (ROOT / "pyproject.toml").read_text(encoding="utf-8")
self.assertIn('version = "0.2.0"', pyproject)
self.assertIn('license = "MPL-2.0"', pyproject)
license_text = (ROOT / "LICENSE").read_text(encoding="utf-8")
self.assertIn("Mozilla Public License Version 2.0", license_text)
debian_copyright = (ROOT / "debian/copyright").read_text(encoding="utf-8")
self.assertIn("License: MPL-2.0", debian_copyright)
def test_debian_dependency_is_exactly_1_14_0(self) -> None:
control = (ROOT / "debian/control").read_text(encoding="utf-8")
self.assertIn("sing-box (= 1.14.0)", control)
def test_in_place_upgrade_is_rejected(self) -> None:
preinst = (ROOT / "debian/preinst").read_text(encoding="utf-8")
self.assertIn('[ "$1" = upgrade ]', preinst)
self.assertIn("exit 1", preinst)
def test_postinst_checks_for_legacy_state(self) -> None:
postinst = (ROOT / "debian/postinst").read_text(encoding="utf-8")
self.assertIn("vpn_egressctl.installcheck", postinst)
def test_sing_box_drop_in_requires_guard(self) -> None:
drop_in = ROOT / "packaging/systemd/sing-box.service.d/10-vpn-egress-guard.conf"
text = drop_in.read_text(encoding="utf-8")
tests/test_policy.py
+2 -2
View File
@@ -21,7 +21,7 @@ class PolicyTests(unittest.TestCase):
def test_production_policy(self) -> None:
policy = self.load(self.raw)
self.assertEqual(policy.sing_box.required_version, "1.13.19")
self.assertEqual(policy.sing_box.required_version, "1.14.0")
self.assertEqual(policy.network.iproute2_table_index, 2022)
def test_unknown_root_key(self) -> None:
@@ -40,7 +40,7 @@ class PolicyTests(unittest.TestCase):
self.load(self.raw)
def test_version_is_exactly_pinned(self) -> None:
for value in ("1.13", ">=1.13,<1.14", "1.14.0"):
for value in ("1.13.19", "1.14", ">=1.14,<1.15", "1.14.1", "1.15.0"):
raw = json.loads(json.dumps(self.raw))
raw["sing_box"]["required_version"] = value
with self.subTest(value=value), self.assertRaises(ValidationError):
tests/test_real_sing_box.py
+18 -12
View File
@@ -6,7 +6,7 @@ import tempfile
import unittest
from pathlib import Path
from vpn_egressctl.renderer_1_13_19 import render_bytes
from vpn_egressctl.renderer_1_14_0 import render_bytes
from vpn_egressctl.uri import parse_hysteria2_uri
from vpn_egressctl.version import probe_version
@@ -14,24 +14,30 @@ from tests.helpers import make_policy
class RealSingBoxIntegrationTests(unittest.TestCase):
@unittest.skipUnless(os.environ.get("SING_BOX_1_13_19"), "real sing-box 1.13.19 binary is not configured")
def test_real_binary_accepts_golden_config(self) -> None:
binary = os.environ["SING_BOX_1_13_19"]
@unittest.skipUnless(os.environ.get("SING_BOX_1_14_0"), "real sing-box 1.14.0 binary is not configured")
def test_real_binary_accepts_golden_configs(self) -> None:
binary = os.environ["SING_BOX_1_14_0"]
version = probe_version(binary)
self.assertEqual(version.version, "1.13.19")
self.assertEqual(version.version, "1.14.0")
self.assertIn("with_quic", version.tags)
self.assertIn("with_gvisor", version.tags)
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
policy = make_policy(root)
endpoint = parse_hysteria2_uri(
"hysteria2://auth@example.com:443?obfs=salamander&obfs-password=obfs"
)
config = root / "config.json"
config.write_bytes(render_bytes(policy, endpoint))
command = "check" if "linux/" in version.environment else "format"
result = subprocess.run([binary, command, "-c", str(config)], capture_output=True, text=True)
self.assertEqual(result.returncode, 0, result.stderr)
for obfs in ("gecko", "salamander"):
with self.subTest(obfs=obfs):
endpoint = parse_hysteria2_uri(
f"hysteria2://auth@example.com:443?obfs={obfs}&obfs-password=obfs"
)
config = root / f"config-{obfs}.json"
config.write_bytes(render_bytes(policy, endpoint))
result = subprocess.run(
[binary, command, "-c", str(config)],
capture_output=True,
text=True,
)
self.assertEqual(result.returncode, 0, result.stderr)
if __name__ == "__main__":
tests/test_renderer.py
+31 -2
View File
@@ -5,18 +5,30 @@ import tempfile
import unittest
from pathlib import Path
from vpn_egressctl.renderer_1_13_19 import render_bytes, render_config
from vpn_egressctl.renderer_1_14_0 import render_bytes, render_config
from vpn_egressctl.uri import parse_hysteria2_uri
from tests.helpers import make_policy
class RendererTests(unittest.TestCase):
def test_gecko_golden_fixture(self) -> None:
with tempfile.TemporaryDirectory() as directory:
actual = render_bytes(
make_policy(Path(directory)),
parse_hysteria2_uri(
"hysteria2://AUTH@fi.api.withen.pro:443/"
"?insecure=0&obfs=gecko&obfs-password=OBFS"
),
)
expected = Path("tests/fixtures/sing-box-1.14.0-gecko.json").read_bytes()
self.assertEqual(actual, expected)
def test_complete_production_shape(self) -> None:
with tempfile.TemporaryDirectory() as directory:
policy = make_policy(Path(directory))
endpoint = parse_hysteria2_uri(
"hysteria2://AUTH@fi.api.withen.pro:443/?insecure=0&obfs=salamander&obfs-password=OBFS"
"hysteria2://AUTH@fi.api.withen.pro:443/?insecure=0&obfs=gecko&obfs-password=OBFS"
)
config = render_config(policy, endpoint)
self.assertEqual(list(config), ["log", "dns", "inbounds", "outbounds", "route"])
@@ -26,16 +38,33 @@ class RendererTests(unittest.TestCase):
self.assertEqual(tun["auto_redirect_input_mark"], "0x2023")
self.assertEqual(tun["auto_redirect_output_mark"], "0x2024")
self.assertEqual(tun["auto_redirect_reset_mark"], "0x2025")
self.assertEqual(tun["dns_mode"], "disabled")
outbound = config["outbounds"][0]
self.assertEqual(outbound["server"], "fi.api.withen.pro")
self.assertEqual(outbound["password"], "AUTH")
self.assertEqual(outbound["obfs"]["password"], "OBFS")
self.assertEqual(outbound["obfs"]["type"], "gecko")
self.assertEqual(outbound["obfs"]["min_packet_size"], 512)
self.assertEqual(outbound["obfs"]["max_packet_size"], 1200)
self.assertFalse(outbound["disable_chrome_parrot"])
self.assertNotIn("bbr_profile", outbound)
self.assertEqual(outbound["bind_interface"], "eth0")
self.assertEqual([item["tag"] for item in config["outbounds"]], ["hy2-out"])
self.assertEqual(config["dns"]["servers"][1]["detour"], "hy2-out")
self.assertEqual(config["route"]["final"], "hy2-out")
self.assertNotIn("185.156.108.141", render_bytes(policy, endpoint).decode())
def test_salamander_fallback_does_not_render_gecko_fields(self) -> None:
with tempfile.TemporaryDirectory() as directory:
config = render_config(
make_policy(Path(directory)),
parse_hysteria2_uri(
"hy2://auth@example.com?obfs=salamander&obfs-password=legacy"
),
)
obfs = config["outbounds"][0]["obfs"]
self.assertEqual(obfs, {"type": "salamander", "password": "legacy"})
def test_multi_port_mapping(self) -> None:
with tempfile.TemporaryDirectory() as directory:
config = render_config(
tests/test_transaction.py
+120 -6
View File
@@ -1,21 +1,26 @@
from __future__ import annotations
import json
import hashlib
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from vpn_egressctl.errors import ApplyError, UnsupportedVersionError, ValidationError
from vpn_egressctl.renderer_1_13_19 import render_bytes
from vpn_egressctl.fsutil import atomic_write_json
from vpn_egressctl.metadata import build_config_metadata
from vpn_egressctl.renderer_1_14_0 import render_bytes
from vpn_egressctl.transaction import Controller
from vpn_egressctl.uri import parse_hysteria2_uri
from vpn_egressctl.version import parse_version_output
from tests.helpers import FakeResponse, FakeRunner, VERSION_OUTPUT, make_policy
OLD_CONFIG = b'{"log":{"level":"error"}}\n'
URI_OLD = "hysteria2://OLD@example.com:443/?obfs=salamander&obfs-password=OLDOBFS"
URI_NEW = "hysteria2://NEW@example.com:443/?obfs=salamander&obfs-password=NEWOBFS"
URI_OLD = "hysteria2://OLD@example.com:443/?obfs=gecko&obfs-password=OLDOBFS"
URI_NEW = "hysteria2://NEW@example.com:443/?obfs=gecko&obfs-password=NEWOBFS"
class TransactionTests(unittest.TestCase):
@@ -33,15 +38,77 @@ class TransactionTests(unittest.TestCase):
)
return controller, selected, policy
def mark_managed(self, controller: Controller, data: bytes, source: str = URI_OLD) -> None:
controller._write_state(
status="ok",
version=parse_version_output(VERSION_OUTPUT),
config_data=data,
source_data=source,
endpoint=parse_hysteria2_uri(source),
changed=False,
)
def write_last_good(self, controller: Controller, data: bytes) -> None:
controller.state_dir.mkdir(parents=True, exist_ok=True)
controller.last_good_path.write_bytes(data)
atomic_write_json(
controller.last_good_meta_path,
build_config_metadata(hashlib.sha256(data).hexdigest(), "1.14.0"),
)
def test_first_clean_sync_has_no_cross_release_backup(self) -> None:
with tempfile.TemporaryDirectory() as directory:
controller, _, policy = self.make(directory)
self.assertTrue(controller.sync())
self.assertTrue(Path(policy.sing_box.config_path).exists())
self.assertFalse(controller.last_good_path.exists())
def test_first_clean_sync_failure_removes_candidate_and_stops_service(self) -> None:
with tempfile.TemporaryDirectory() as directory:
runner = FakeRunner(restart_results=[1])
controller, _, policy = self.make(directory, runner=runner)
with self.assertRaises(ApplyError) as caught:
controller.sync()
self.assertTrue(caught.exception.rollback_ok)
self.assertFalse(Path(policy.sing_box.config_path).exists())
self.assertTrue(any("stop" in call for call in runner.calls))
state = json.loads(controller.state_path.read_text(encoding="utf-8"))
self.assertEqual(state["status"], "initial_apply_failed_service_stopped")
def test_state_write_failure_rolls_back_first_install(self) -> None:
with tempfile.TemporaryDirectory() as directory:
controller, runner, policy = self.make(directory)
with (
mock.patch.object(controller, "_write_state", side_effect=OSError("disk")),
self.assertRaises(ApplyError) as caught,
):
controller.sync()
self.assertTrue(caught.exception.rollback_ok)
self.assertFalse(Path(policy.sing_box.config_path).exists())
self.assertTrue(any("stop" in call for call in runner.calls))
def test_unmanaged_existing_config_requires_clean_install(self) -> None:
with tempfile.TemporaryDirectory() as directory:
controller, _, policy = self.make(directory)
Path(policy.sing_box.config_path).write_bytes(OLD_CONFIG)
with self.assertRaisesRegex(ValidationError, "clean installation"):
controller.sync()
self.assertEqual(Path(policy.sing_box.config_path).read_bytes(), OLD_CONFIG)
def test_sync_applies_and_records_last_good(self) -> None:
with tempfile.TemporaryDirectory() as directory:
controller, runner, policy = self.make(directory)
Path(policy.sing_box.config_path).write_bytes(OLD_CONFIG)
self.mark_managed(controller, OLD_CONFIG)
changed = controller.sync()
self.assertTrue(changed)
installed = Path(policy.sing_box.config_path).read_bytes()
self.assertIn(b'"password": "NEW"', installed)
self.assertEqual(controller.last_good_path.read_bytes(), OLD_CONFIG)
self.assertTrue(controller.last_good_meta_path.exists())
backups = list(controller.backup_dir.glob("config.*.json"))
self.assertEqual(len(backups), 1)
self.assertTrue(Path(str(backups[0]) + ".meta").exists())
state_text = controller.state_path.read_text(encoding="utf-8")
self.assertNotIn("NEWOBFS", state_text)
self.assertNotIn('"NEW"', state_text)
@@ -67,7 +134,7 @@ class TransactionTests(unittest.TestCase):
def test_version_gate_is_non_mutating(self) -> None:
with tempfile.TemporaryDirectory() as directory:
output = VERSION_OUTPUT.replace("1.13.19", "1.13.12", 1)
output = VERSION_OUTPUT.replace("1.14.0", "1.13.19", 1)
controller, _, policy = self.make(directory, runner=FakeRunner(version=output))
Path(policy.sing_box.config_path).write_bytes(OLD_CONFIG)
with self.assertRaises(UnsupportedVersionError):
@@ -79,6 +146,7 @@ class TransactionTests(unittest.TestCase):
runner = FakeRunner(restart_results=[1, 0])
controller, _, policy = self.make(directory, runner=runner)
Path(policy.sing_box.config_path).write_bytes(OLD_CONFIG)
self.mark_managed(controller, OLD_CONFIG)
with self.assertRaises(ApplyError) as caught:
controller.sync()
self.assertTrue(caught.exception.rollback_ok)
@@ -91,6 +159,7 @@ class TransactionTests(unittest.TestCase):
controller, _, policy = self.make(directory, runner=runner)
Path(policy.runtime.uri_path).write_text(URI_OLD + "\n", encoding="utf-8")
Path(policy.sing_box.config_path).write_bytes(OLD_CONFIG)
self.mark_managed(controller, OLD_CONFIG)
with self.assertRaises(ApplyError):
controller.import_uri(URI_NEW)
self.assertEqual(Path(policy.runtime.uri_path).read_text().strip(), URI_OLD)
@@ -110,8 +179,8 @@ class TransactionTests(unittest.TestCase):
current = render_bytes(policy, parse_hysteria2_uri(URI_NEW))
previous = render_bytes(policy, parse_hysteria2_uri(URI_OLD))
Path(policy.sing_box.config_path).write_bytes(current)
controller.state_dir.mkdir(parents=True)
controller.last_good_path.write_bytes(previous)
self.mark_managed(controller, current, URI_NEW)
self.write_last_good(controller, previous)
controller.rollback()
self.assertEqual(Path(policy.sing_box.config_path).read_bytes(), previous)
self.assertEqual(controller.last_good_path.read_bytes(), current)
@@ -119,6 +188,51 @@ class TransactionTests(unittest.TestCase):
status = controller.status()
self.assertEqual(status["status"], "rolled_back")
self.assertFalse(status["source_drift"])
self.assertTrue(status["last_good_valid"])
def test_legacy_last_good_without_metadata_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
controller, _, policy = self.make(directory)
current = render_bytes(policy, parse_hysteria2_uri(URI_NEW))
Path(policy.sing_box.config_path).write_bytes(current)
self.mark_managed(controller, current, URI_NEW)
controller.state_dir.mkdir(parents=True, exist_ok=True)
controller.last_good_path.write_bytes(OLD_CONFIG)
with self.assertRaisesRegex(ValidationError, "metadata"):
controller.rollback()
def test_manual_rollback_metadata_failure_restores_both_sides(self) -> None:
with tempfile.TemporaryDirectory() as directory:
controller, _, policy = self.make(directory)
current = render_bytes(policy, parse_hysteria2_uri(URI_NEW))
previous = render_bytes(policy, parse_hysteria2_uri(URI_OLD))
Path(policy.sing_box.config_path).write_bytes(current)
self.mark_managed(controller, current, URI_NEW)
self.write_last_good(controller, previous)
original_meta = controller.last_good_meta_path.read_bytes()
with (
mock.patch(
"vpn_egressctl.transaction.atomic_write_json",
side_effect=OSError("disk"),
),
self.assertRaises(ApplyError) as caught,
):
controller.rollback()
self.assertTrue(caught.exception.rollback_ok)
self.assertEqual(Path(policy.sing_box.config_path).read_bytes(), current)
self.assertEqual(controller.last_good_path.read_bytes(), previous)
self.assertEqual(controller.last_good_meta_path.read_bytes(), original_meta)
def test_last_good_checksum_mismatch_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
controller, _, policy = self.make(directory)
current = render_bytes(policy, parse_hysteria2_uri(URI_NEW))
Path(policy.sing_box.config_path).write_bytes(current)
self.mark_managed(controller, current, URI_NEW)
self.write_last_good(controller, OLD_CONFIG)
controller.last_good_path.write_bytes(b"{}\n")
with self.assertRaisesRegex(ValidationError, "checksum"):
controller.rollback()
def test_diff_never_contains_secrets(self) -> None:
with tempfile.TemporaryDirectory() as directory:
tests/test_uri.py
+23 -3
View File
@@ -30,6 +30,24 @@ class UriParserTests(unittest.TestCase):
self.assertEqual(endpoint.obfs_type, "salamander")
self.assertEqual(endpoint.obfs_password, "o@p")
def test_production_gecko_uri(self) -> None:
endpoint = parse_hysteria2_uri(
"hysteria2://peer%2Bsecret@example.com:443/"
"?insecure=0&obfs=gecko&obfs-password=a%26b%3Dc%2Bd&sni=vpn.example.com"
)
self.assertEqual(endpoint.password, "peer+secret")
self.assertEqual(endpoint.obfs_type, "gecko")
self.assertEqual(endpoint.obfs_password, "a&b=c+d")
def test_raw_plus_in_query_is_not_decoded_as_space(self) -> None:
endpoint = parse_hysteria2_uri(
"hy2://secret@example.com?obfs=gecko&obfs-password=a+b"
)
self.assertEqual(endpoint.obfs_password, "a+b")
def test_encoded_duplicate_name_is_rejected(self) -> None:
self.assert_invalid("hy2://secret@example.com?sni=a&%73ni=b", "Duplicate")
def test_ipv6(self) -> None:
endpoint = parse_hysteria2_uri("hysteria2://secret@[2001:db8::1]:444/")
self.assertEqual(endpoint.server, "2001:db8::1")
@@ -87,11 +105,13 @@ class UriParserTests(unittest.TestCase):
("hysteria2://secret@example.com?unknown=x", "Unsupported"),
("hysteria2://secret@example.com?sni=a&sni=b", "Duplicate"),
("hysteria2://secret@example.com?insecure=true", "exactly"),
("hysteria2://secret@example.com?obfs=gecko", "1.13.19"),
("hysteria2://secret@example.com?obfs=gecko", "obfs-password"),
("hysteria2://secret@example.com?obfs=salamander", "obfs-password"),
("hysteria2://secret@example.com?obfs-password=x", "requires"),
("hysteria2://secret@example.com?pinSHA256=x", "safely"),
("hysteria2://secret@example.com?ech=x", "safely"),
("hysteria2://secret@example.com?pinSHA256=x", "not implemented"),
("hysteria2://secret@example.com?ech=x", "not implemented"),
("hysteria2://secret@example.com?sni", "query parameter"),
("hysteria2://secret@example.com?sni=a&&insecure=0", "Empty"),
("hysteria2://sec%ZZret@example.com", "percent"),
("hysteria2://alice@example@server.example", "percent-encoded"),
("hysteria2://secret@example.com?sni=%FF", "UTF-8"),
tests/test_version.py
+3 -3
View File
@@ -12,13 +12,13 @@ from tests.helpers import FakeRunner, VERSION_OUTPUT
class VersionTests(unittest.TestCase):
def test_parse_and_require(self) -> None:
version = parse_version_output(VERSION_OUTPUT)
self.assertEqual(version.version, "1.13.19")
self.assertEqual(version.version, "1.14.0")
self.assertIn("with_quic", version.tags)
require_supported(version)
def test_other_versions_are_rejected(self) -> None:
for value in ("1.13.12", "1.13.20", "1.14.0", "1.13.19-rc.1"):
text = VERSION_OUTPUT.replace("1.13.19", value, 1)
for value in ("1.13.19", "1.13.21", "1.14.1", "1.14.0-rc.5", "1.15.0-alpha.1"):
text = VERSION_OUTPUT.replace("1.14.0", value, 1)
with self.subTest(value=value), self.assertRaises(UnsupportedVersionError):
require_supported(parse_version_output(text))