fix: harden sing-box egress rollout

This commit is contained in:
2026-08-27 03:39:06 +05:00
parent adb3e849df
commit 8ac3b2b1ba
24 changed files with 228 additions and 48 deletions
+2 -2
View File
@@ -12,9 +12,9 @@ class CliTests(unittest.TestCase):
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
_parser().parse_args(["import", "hysteria2://secret@example.com"])
def test_render_requires_output(self) -> None:
def test_render_command_is_not_public(self) -> None:
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
_parser().parse_args(["render"])
_parser().parse_args(["render", "--output", "/etc/sing-box/config.json"])
def test_all_commands_parse(self) -> None:
for command in ("check", "diff", "sync", "status", "doctor", "rollback"):
+32
View File
@@ -1,11 +1,14 @@
from __future__ import annotations
import json
import stat
import subprocess
import tempfile
import unittest
from dataclasses import replace
from pathlib import Path
from types import SimpleNamespace
from unittest import mock
from vpn_egressctl.doctor import Doctor
from vpn_egressctl.renderer_1_13_19 import render_bytes
@@ -58,6 +61,35 @@ class DoctorTests(unittest.TestCase):
selected = [check for check in checks if check.name == "tls-insecure"]
self.assertEqual(selected[0].level, "WARN")
def test_permissions_reject_non_root_owner_on_posix(self) -> None:
with tempfile.TemporaryDirectory() as directory:
doctor = Doctor(make_policy(Path(directory)))
metadata = SimpleNamespace(
st_mode=stat.S_IFREG | 0o600,
st_uid=1000,
st_gid=1000,
)
with (
mock.patch("vpn_egressctl.doctor.Path.stat", return_value=metadata),
mock.patch("vpn_egressctl.doctor.os.name", "posix"),
):
doctor._permissions("/protected/file", 0o600)
self.assertEqual(doctor.checks[-1].level, "ERROR")
self.assertIn("expected 0:0", doctor.checks[-1].message)
def test_permissions_require_expected_object_type(self) -> None:
with tempfile.TemporaryDirectory() as directory:
doctor = Doctor(make_policy(Path(directory)))
metadata = SimpleNamespace(
st_mode=stat.S_IFDIR | 0o700,
st_uid=0,
st_gid=0,
)
with mock.patch("vpn_egressctl.doctor.Path.stat", return_value=metadata):
doctor._permissions("/protected/file", 0o600)
self.assertEqual(doctor.checks[-1].level, "ERROR")
self.assertIn("regular file", doctor.checks[-1].message)
if __name__ == "__main__":
unittest.main()
+27
View File
@@ -0,0 +1,27 @@
from __future__ import annotations
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
class PackagingTests(unittest.TestCase):
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")
self.assertIn("Requires=vpn-egress-guard.service", text)
self.assertIn("After=vpn-egress-guard.service", text)
def test_debian_package_installs_sing_box_drop_in(self) -> None:
install = (ROOT / "debian/install").read_text(encoding="utf-8")
self.assertIn(
"packaging/systemd/sing-box.service.d/10-vpn-egress-guard.conf "
"usr/lib/systemd/system/sing-box.service.d",
install,
)
if __name__ == "__main__":
unittest.main()
+1
View File
@@ -20,6 +20,7 @@ class RealSingBoxIntegrationTests(unittest.TestCase):
version = probe_version(binary)
self.assertEqual(version.version, "1.13.19")
self.assertIn("with_quic", version.tags)
self.assertIn("with_gvisor", version.tags)
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
policy = make_policy(root)
+1
View File
@@ -31,6 +31,7 @@ class RendererTests(unittest.TestCase):
self.assertEqual(outbound["password"], "AUTH")
self.assertEqual(outbound["obfs"]["password"], "OBFS")
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())
+3
View File
@@ -116,6 +116,9 @@ class TransactionTests(unittest.TestCase):
self.assertEqual(Path(policy.sing_box.config_path).read_bytes(), previous)
self.assertEqual(controller.last_good_path.read_bytes(), current)
self.assertEqual(json.loads(controller.state_path.read_text())["status"], "rolled_back")
status = controller.status()
self.assertEqual(status["status"], "rolled_back")
self.assertFalse(status["source_drift"])
def test_diff_never_contains_secrets(self) -> None:
with tempfile.TemporaryDirectory() as directory:
+5
View File
@@ -50,6 +50,10 @@ class UriParserTests(unittest.TestCase):
endpoint = parse_hysteria2_uri("hysteria2://alice%3Acorrect%20horse@example.com")
self.assertEqual(endpoint.password, "alice:correct horse")
def test_percent_encoded_at_is_preserved(self) -> None:
endpoint = parse_hysteria2_uri("hysteria2://alice%40example@example.com")
self.assertEqual(endpoint.password, "alice@example")
def test_unicode_host_is_idna(self) -> None:
endpoint = parse_hysteria2_uri("hysteria2://secret@пример.рф")
self.assertEqual(endpoint.server, "xn--e1afmkfd.xn--p1ai")
@@ -89,6 +93,7 @@ class UriParserTests(unittest.TestCase):
("hysteria2://secret@example.com?pinSHA256=x", "safely"),
("hysteria2://secret@example.com?ech=x", "safely"),
("hysteria2://sec%ZZret@example.com", "percent"),
("hysteria2://alice@example@server.example", "percent-encoded"),
("hysteria2://secret@example.com?sni=%FF", "UTF-8"),
]
for uri, marker in cases:
+5
View File
@@ -32,6 +32,11 @@ class VersionTests(unittest.TestCase):
with self.assertRaises(UnsupportedVersionError):
require_supported(version)
def test_missing_gvisor_rejected(self) -> None:
version = parse_version_output(VERSION_OUTPUT.replace("with_gvisor,", ""))
with self.assertRaises(UnsupportedVersionError):
require_supported(version)
def test_command_failure(self) -> None:
runner = FakeRunner()
runner.version = "garbage"