120 lines
4.8 KiB
Python
120 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import hashlib
|
|
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.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
|
|
|
|
|
|
class DoctorTests(unittest.TestCase):
|
|
def prepare(self, directory: str, uri: str):
|
|
root = Path(directory)
|
|
policy = make_policy(root)
|
|
endpoint = parse_hysteria2_uri(uri)
|
|
Path(policy.runtime.uri_path).parent.mkdir(parents=True)
|
|
Path(policy.runtime.uri_path).write_text(uri + "\n", encoding="utf-8")
|
|
Path(policy.sing_box.config_path).write_bytes(render_bytes(policy, endpoint))
|
|
return policy
|
|
|
|
def test_endpoint_exclusion_is_an_error(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
policy = self.prepare(directory, "hy2://auth@example.com")
|
|
policy = replace(
|
|
policy,
|
|
network=replace(
|
|
policy.network,
|
|
route_exclude_address=policy.network.route_exclude_address + ("8.8.8.8/32",),
|
|
),
|
|
)
|
|
doctor = Doctor(
|
|
policy,
|
|
runner=FakeRunner(),
|
|
resolver=lambda *args: [(None, None, None, None, ("8.8.8.8", 443))],
|
|
)
|
|
checks = doctor.run()
|
|
selected = [check for check in checks if check.name == "endpoint-exclusion"]
|
|
self.assertEqual(selected[0].level, "ERROR")
|
|
|
|
def test_insecure_tls_is_reported_without_secret(self) -> None:
|
|
secret = "NEVER-LOG-ME"
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
policy = self.prepare(directory, f"hy2://{secret}@example.com?insecure=1")
|
|
doctor = Doctor(
|
|
policy,
|
|
runner=FakeRunner(),
|
|
resolver=lambda *args: [(None, None, None, None, ("8.8.4.4", 443))],
|
|
)
|
|
checks = doctor.run()
|
|
output = json.dumps([check.message for check in checks])
|
|
self.assertNotIn(secret, output)
|
|
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)
|
|
|
|
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()
|