loop-uh
7acad15dcd
The AmneziaWG sidecar failed at startup with "Physical interface for Amnezia UDP transport not found". Three compounding causes: - physical_network() spawned PowerShell and filtered on HardwareInterface, which wrongly excludes Hyper-V / WSL / Docker vEthernet uplinks, so every AWG key failed identically on such hosts; a transient miss aborted the whole startup with a hard OSError. - A successful tunnel changes the outbound IP (e.g. 10.9.0.49), which the network monitor misread as a real network change and reconnected on, re-running the resolve while the tunnel held the default route — a self-inflicted flap. Replace the PowerShell probe with a pure-WinAPI resolver: extend win_netinfo (GetAdaptersAddresses gateways, DNS, OperStatus, IfType, Ipv4Metric) and pick the up adapter that owns an IPv4 gateway and a routable address, lowest Ipv4Metric. Requiring a gateway naturally excludes the WireGuard/AWG TUN (it has none), so the relay never binds back onto its own tunnel; dropping HardwareInterface fixes vEthernet uplinks. The owned Go core still hard-requires a non-zero interface index (IP_UNICAST_IF, no default-bind fallback), so a transient miss is retried instead of aborting. Ignore network-change events whose address is the active tunnel's own, breaking the reconnect loop. Extract the shared sidecar seam — the loopback SOCKS relay must accept a TCP connection before the sing-box front dials it — into engines/sidecar.wait_for_loopback_relay(). Hysteria delegates to it (no behavior change); Amnezia adopts it as a confirming gate after relay_ready, so both cores share one readiness contract. The divergent domain models (config handoff, handshake detection, failure taxonomy, recovery) intentionally stay per-engine. Verified on Linux: 890 unit tests pass (offscreen). Tunnel-comes-up and Hysteria startup still need on-device confirmation on Windows. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HAjvZYzPW2yTtToKJXGdbS
46 lines
1.7 KiB
Python
46 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import socket
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from PyQt6.QtCore import QCoreApplication
|
|
|
|
from xray_fluent.engines.sidecar import wait_for_loopback_relay
|
|
|
|
_APP = QCoreApplication.instance() or QCoreApplication([])
|
|
|
|
|
|
class WaitForLoopbackRelayTests(unittest.TestCase):
|
|
def test_returns_true_when_listener_accepts(self) -> None:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
|
|
listener.bind(("127.0.0.1", 0))
|
|
listener.listen(1)
|
|
port = listener.getsockname()[1]
|
|
self.assertTrue(wait_for_loopback_relay(port, host="127.0.0.1", timeout=2.0))
|
|
|
|
def test_returns_false_on_timeout_when_nothing_listens(self) -> None:
|
|
# Reserve then close a port so the connection is refused for the whole wait.
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
|
|
probe.bind(("127.0.0.1", 0))
|
|
port = probe.getsockname()[1]
|
|
with patch("xray_fluent.engines.sidecar.readiness.sleep_with_events"):
|
|
self.assertFalse(wait_for_loopback_relay(port, host="127.0.0.1", timeout=0.2))
|
|
|
|
def test_should_continue_false_aborts_immediately(self) -> None:
|
|
calls: list[int] = []
|
|
|
|
def stop_now() -> bool:
|
|
calls.append(1)
|
|
return False
|
|
|
|
with patch("xray_fluent.engines.sidecar.readiness.sleep_with_events") as sleeper:
|
|
self.assertFalse(
|
|
wait_for_loopback_relay(1, host="127.0.0.1", timeout=5.0, should_continue=stop_now)
|
|
)
|
|
self.assertEqual(len(calls), 1)
|
|
sleeper.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|