From f0dd67af5990352bc1dde5a05675cab02d7aeebd Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 04:55:37 +0000 Subject: [PATCH] service: find our own tunnel among any number of foreign ones (#13) Tunnel detection polled a fixed ["gpd0", "tun0", "tun1"] list. With two other tun-based VPNs already up, openconnect's kernel-assigned fallback name lands on tun2 - a name that list never contained - so the loop polled forever: authentication and the ESP tunnel succeeded, but the plugin never handed NetworkManager an IP4 config and vpn.timeout tore the connection down. The candidates are now read from /sys/class/net (gpdN/tunN, numerically ordered, gpd first; tunl0 and other look-alikes excluded), so any number of foreign tunnels may be active. The pre-existing-interface snapshot from #7 grows with them and keeps doing the actual safety work. Widening the candidate set also widens the window in which another VPN activated during the SAML login could be adopted as ours, so the tunnel is additionally attributed to gpclient by the file descriptors it holds (iff: in /proc/PID/fdinfo, across gpclient and its children). That check only narrows the choice when it yields something; when the information is unavailable, the snapshot decides on its own as before. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_011pqLycL9FbCQWYcTKchZjs --- debian/changelog | 14 ++ docs/PYTHON_SERVICE.md | 28 ++- service/nm-gpclient-service.py | 191 +++++++++++++-- tests/README.md | 1 + tests/unit/test_tunnel_detection.py | 346 ++++++++++++++++++++++++++++ 5 files changed, 552 insertions(+), 28 deletions(-) create mode 100644 tests/unit/test_tunnel_detection.py diff --git a/debian/changelog b/debian/changelog index 21f745e..8870700 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,17 @@ +network-manager-gpclient (1.4.1-1) unstable; urgency=medium + + * Fix the connection never leaving "activating" when two other tun-based + VPNs are already up (closes #13): tunnel detection no longer polls a + fixed gpd0/tun0/tun1 list, which could not see our own tunnel once it + landed on tun2 or higher, so NetworkManager's vpn.timeout tore the + connection down. The candidates are read from /sys/class/net instead, + and any number of foreign tunnels may be active + * Attribute the tunnel to gpclient by the file descriptors it holds + (iff: in /proc/PID/fdinfo), so another VPN activated during + authentication is not adopted as ours + + -- WMP Wed, 26 Aug 2026 12:00:00 +0200 + network-manager-gpclient (1.4.0-1) unstable; urgency=medium * Fix portal addresses: --gateway is no longer forced to the server address, diff --git a/docs/PYTHON_SERVICE.md b/docs/PYTHON_SERVICE.md index bdd7cee..5966bf1 100644 --- a/docs/PYTHON_SERVICE.md +++ b/docs/PYTHON_SERVICE.md @@ -12,7 +12,7 @@ The `nm-gpclient-service` is a Python3-based VPN service that implements the Net - Implements D-Bus interface `org.freedesktop.NetworkManager.VPN.Plugin` - Communicates with NetworkManager via System Bus - Manages the `gpclient` process - - Monitors tunnel interfaces (gpd0, tun0, tun1) + - Monitors tunnel interfaces (`gpd0` and any `tunN`, discovered at runtime) 2. **D-Bus Interface** - Service name: `org.freedesktop.NetworkManager.gpclient` @@ -104,12 +104,26 @@ Background thread reads gpclient stdout and detects messages: After detecting message, immediately checks for tunnel interface. ### Tunnel Interface Detection -Every 500ms checks for: -``` -/sys/class/net/gpd0 -/sys/class/net/tun0 -/sys/class/net/tun1 -``` +Every 500ms `/sys/class/net` is scanned for `gpdN`/`tunN` devices (`gpd0` +first, then `tunN` in numeric order). The candidate list is *not* hardcoded: +openconnect falls back to a kernel-assigned `tunN` name whenever `gpd0` is +unavailable, and N is the first free number - with two other tun-based VPNs +already up, our own tunnel lands on `tun2` +([#13](https://github.com/WMP/GlobalProtect-SAML-NetworkManager/issues/13)). + +A candidate is accepted only when all of the following hold: + +1. It has an IPv4 address. +2. It is not one of the interfaces that already existed with the same address + when `Connect()` started (a stale `gpd0`, or another VPN client's tunnel - + [#7](https://github.com/WMP/GlobalProtect-SAML-NetworkManager/issues/7)). + That snapshot is taken from the same scan, so any number of foreign + tunnels may be active. +3. If gpclient's own file descriptors name a tunnel interface + (`iff:` in `/proc/PID/fdinfo/N`, looked up across gpclient and its + children), the candidate is that interface. This keeps another VPN that + comes up mid-connect from being adopted; when the information is + unavailable the snapshot decides on its own. After finding interface: 1. Builds IP configuration (interface name + DNS) diff --git a/service/nm-gpclient-service.py b/service/nm-gpclient-service.py index 0d1fc46..dc19129 100644 --- a/service/nm-gpclient-service.py +++ b/service/nm-gpclient-service.py @@ -61,9 +61,29 @@ NM_VPN_PLUGIN_FAILURE_CONNECT_FAILED = 1 NM_VPN_PLUGIN_FAILURE_BAD_IP_CONFIG = 2 -# Interface names the tunnel detection looks for. gpd0 is created -# exclusively by gpclient; tun0/tun1 may also belong to other VPN clients. -TUNNEL_INTERFACES = ["gpd0", "tun0", "tun1"] +# --- Tunnel interface detection --------------------------------------------- +# +# The tunnel our gpclient brings up is either gpd0 (created exclusively by +# gpclient) or a kernel-assigned tunN device - openconnect falls back to the +# latter, and N is simply the first free number. A fixed candidate list used +# to be enough, but it caps how many *foreign* tunN VPNs may be up at once: +# with two other tun-based VPNs already holding tun0 and tun1, our own tunnel +# lands on tun2, which no fixed list predicted, so detection polled forever +# until NetworkManager's vpn.timeout killed the connection (issue #13). +# +# So the candidates are discovered at runtime instead. The safety work is done +# by the pre-existing-interface snapshot (issue #7) and by matching the tunnel +# against the file descriptors our own gpclient process holds - not by the +# length of a hardcoded list. +NET_SYSFS_PATH = "/sys/class/net" +PROC_PATH = "/proc" + +# gpdN / tunN, and nothing else: "tunl0" (the always-present IPIP tunnel +# device) must not be mistaken for a VPN tunnel. +TUNNEL_INTERFACE_RE = re.compile(r"^(?Pgpd|tun)(?P\d+)$") + +# gpd first (only gpclient creates those), then tunN by number +TUNNEL_INTERFACE_KIND_ORDER = ("gpd", "tun") GPCLIENT_BINARY = "/usr/bin/gpclient" @@ -442,6 +462,104 @@ def resolve_browser(value: str) -> Tuple[str, Optional[str]]: return target, None +def tunnel_candidate_order(iface: str) -> Tuple[int, int]: + """Sort key putting gpd0 first and tunN in numeric (not lexical) order.""" + match = TUNNEL_INTERFACE_RE.match(iface) + if not match: + return len(TUNNEL_INTERFACE_KIND_ORDER), 0 + kind = match.group("kind") + return TUNNEL_INTERFACE_KIND_ORDER.index(kind), int(match.group("index")) + + +def list_tunnel_candidates() -> List[str]: + """Tunnel interfaces present right now, in the order detection tries them. + + Replaces the old fixed ["gpd0", "tun0", "tun1"] list, which could not see + a tunnel that landed on tun2 or higher because other VPNs held the lower + numbers (issue #13). + """ + try: + names = os.listdir(NET_SYSFS_PATH) + except OSError as e: + logger.warning(f"Cannot list {NET_SYSFS_PATH}: {e}") + return [] + + candidates = [name for name in names if TUNNEL_INTERFACE_RE.match(name)] + return sorted(candidates, key=tunnel_candidate_order) + + +def tunnel_ifaces_held_by(pids: List[int]) -> set: + r"""Tunnel interfaces whose file descriptor is held by one of `pids`. + + The kernel names the interface behind a tun file descriptor in + /proc/PID/fdinfo/N ("iff:\ttun2"), which identifies our own tunnel + directly instead of inferring it from what appeared since Connect() + started. Returns an empty set when the information is unavailable + (process already gone, /proc unreadable); the caller then falls back to + the pre-existing-interface snapshot. + """ + ifaces = set() + for pid in pids: + fdinfo_dir = f"{PROC_PATH}/{pid}/fdinfo" + try: + fds = os.listdir(fdinfo_dir) + except OSError: + continue + for fd in fds: + try: + with open(f"{fdinfo_dir}/{fd}", "r") as handle: + content = handle.read() + except OSError: + continue + for line in content.splitlines(): + if line.startswith("iff:"): + name = line.split(":", 1)[1].strip() + if name: + ifaces.add(name) + return ifaces + + +def process_tree(root_pid: int) -> List[int]: + """A PID plus every descendant, so a tunnel opened by a forked helper of + gpclient is still recognised as ours.""" + try: + pids = [int(entry) for entry in os.listdir(PROC_PATH) if entry.isdigit()] + except OSError: + return [root_pid] + + children: Dict[int, List[int]] = {} + for pid in pids: + try: + with open(f"{PROC_PATH}/{pid}/stat", "r") as handle: + stat = handle.read() + except OSError: + continue + # Field 2 is the command name in parentheses and may itself contain + # spaces and parentheses, so the parent PID is read after the last ')' + close = stat.rfind(")") + if close < 0: + continue + fields = stat[close + 1 :].split() + if len(fields) < 2: + continue + try: + children.setdefault(int(fields[1]), []).append(pid) + except ValueError: + continue + + tree: List[int] = [] + seen = set() + queue = [root_pid] + while queue: + pid = queue.pop() + if pid in seen: + continue + seen.add(pid) + tree.append(pid) + queue.extend(children.get(pid, [])) + return tree + + class OutputScanner: """Split a raw PTY output stream into complete lines and a pending tail. @@ -2049,20 +2167,35 @@ async def _snapshot_tunnel_interfaces(self) -> Dict[str, Any]: An interface recorded here (with an unchanged IP) is never accepted by _check_tunnel_loop: it is either a stale gpd0 from a crashed - session or another VPN client's tunnel (issue #7). + session or another VPN client's tunnel (issue #7). Any number of + foreign tunnels may be up - the snapshot grows with them (issue #13). """ snapshot = {} - for iface in TUNNEL_INTERFACES: - if os.path.exists(f"/sys/class/net/{iface}"): - ip_addr, _ = await self._get_iface_ipv4(iface) - snapshot[iface] = ip_addr - logger.info( - f"Interface {iface} (IP: {ip_addr}) already exists before " - "gpclient start - it will be ignored by tunnel detection " - "unless its address changes" - ) + for iface in list_tunnel_candidates(): + ip_addr, _ = await self._get_iface_ipv4(iface) + snapshot[iface] = ip_addr + logger.info( + f"Interface {iface} (IP: {ip_addr}) already exists before " + "gpclient start - it will be ignored by tunnel detection " + "unless its address changes" + ) return snapshot + def _tunnel_ifaces_owned_by_gpclient(self) -> set: + """Tunnel interfaces held open by our own gpclient process tree. + + Empty when the answer is unknown (gpclient already exited, /proc + unreadable) - the caller then relies on the snapshot alone. + """ + process = self.gpclient_process + if process is None or process.returncode is not None: + return set() + try: + return tunnel_ifaces_held_by(process_tree(process.pid)) + except Exception as e: + logger.debug(f"Could not determine gpclient's tunnel interfaces: {e}") + return set() + async def _cleanup_stale_gpd0(self) -> None: """Remove a leftover gpd0 interface from a previous session. @@ -2070,10 +2203,10 @@ async def _cleanup_stale_gpd0(self) -> None: single session via its lock file, so a gpd0 with no running gpclient process is always stale. A stale gpd0 blackholes routing (the portal becomes unreachable) and used to be picked up by tunnel detection as - a live connection (issue #7). tun0/tun1 may belong to other VPN + a live connection (issue #7). tunN devices may belong to other VPN clients and are never touched. """ - if not os.path.exists("/sys/class/net/gpd0"): + if not os.path.exists(f"{NET_SYSFS_PATH}/gpd0"): return try: @@ -2110,7 +2243,7 @@ async def _cleanup_stale_gpd0(self) -> None: except Exception as e: logger.debug(f"'gpclient disconnect' during cleanup failed: {e}") - if os.path.exists("/sys/class/net/gpd0"): + if os.path.exists(f"{NET_SYSFS_PATH}/gpd0"): try: proc = await asyncio.create_subprocess_exec( "ip", "link", "del", "gpd0" @@ -2124,18 +2257,19 @@ async def _cleanup_stale_gpd0(self) -> None: except Exception as e: logger.error(f"Failed to delete stale gpd0: {e}") - if not os.path.exists("/sys/class/net/gpd0"): + if not os.path.exists(f"{NET_SYSFS_PATH}/gpd0"): logger.info("Stale gpd0 interface removed") async def _check_tunnel_loop(self) -> None: """Periodically check for tunnel interface""" try: while True: - for iface in TUNNEL_INTERFACES: - iface_path = f"/sys/class/net/{iface}" - if not os.path.exists(iface_path): - continue + # Which interfaces gpclient itself holds open. Looked up at + # most once per round, and only once a new candidate actually + # turned up, so the /proc walk stays off the idle path. + owned_ifaces = None + for iface in list_tunnel_candidates(): # Check if interface has an IP address (not just exists) ip_addr, prefix = await self._get_iface_ipv4(iface) @@ -2159,6 +2293,21 @@ async def _check_tunnel_loop(self) -> None: ) continue + # A tunnel that appeared after Connect() is normally ours, + # but another VPN may well have been started at the same + # moment. When gpclient's own file descriptors tell us + # which interface is ours, trust that over the timing + # (issue #13). + if owned_ifaces is None: + owned_ifaces = self._tunnel_ifaces_owned_by_gpclient() + if owned_ifaces and iface not in owned_ifaces: + logger.debug( + f"Interface {iface} is new but not held by " + f"gpclient (it holds {sorted(owned_ifaces)}), " + "skipping" + ) + continue + logger.info( f"VPN connected - tunnel interface {iface} detected with IP {ip_addr}!" ) diff --git a/tests/README.md b/tests/README.md index 6cc6eaf..e3f75c7 100644 --- a/tests/README.md +++ b/tests/README.md @@ -173,6 +173,7 @@ tests/ │ ├── test_gateway_selection.py # Gateway list parsing, matching, browser resolution (#7) │ ├── test_select_pty.py # Full output pipeline over a real PTY │ ├── test_openssl_retry.py # Legacy TLS renegotiation retry (#2) +│ ├── test_tunnel_detection.py # Finding our own tunnel among foreign ones (#13) │ └── test_auth_dialog.py # Auth dialog protocol, incl. the SAML case (#8) ├── helpers/ │ ├── __init__.py diff --git a/tests/unit/test_tunnel_detection.py b/tests/unit/test_tunnel_detection.py new file mode 100644 index 0000000..aaa3a91 --- /dev/null +++ b/tests/unit/test_tunnel_detection.py @@ -0,0 +1,346 @@ +""" +Tests for tunnel interface detection (issue #13, building on issue #7). + +Detection used to poll a fixed ["gpd0", "tun0", "tun1"] list. With two other +tun-based VPNs already up, our own tunnel lands on tun2 - a name that list +never contained - so the connection stayed in "activating" until +NetworkManager's vpn.timeout killed it. The candidates are now discovered from +/sys/class/net, and the tunnel is attributed to gpclient by the file +descriptors it holds. + +Run with: make test-unit (or: python3 -m pytest tests/unit -v) +""" + +import asyncio +import os +import socket +import struct + + +def _fake_net(tmp_path, names): + """A stand-in for /sys/class/net containing `names`.""" + net = tmp_path / "sys-class-net" + net.mkdir(exist_ok=True) + for name in names: + (net / name).mkdir(exist_ok=True) + return str(net) + + +def _fake_proc(tmp_path, processes): + """A stand-in for /proc. + + `processes` maps pid -> (comm, ppid, [interface names held via a tun fd]). + """ + proc = tmp_path / "proc" + proc.mkdir(exist_ok=True) + for pid, (comm, ppid, ifaces) in processes.items(): + entry = proc / str(pid) + entry.mkdir(exist_ok=True) + (entry / "stat").write_text(f"{pid} ({comm}) S {ppid} 0 0 0 -1 0\n") + fdinfo = entry / "fdinfo" + fdinfo.mkdir(exist_ok=True) + # fd 0 is always something that is not a tun device + (fdinfo / "0").write_text("pos:\t0\nflags:\t02\nmnt_id:\t24\n") + for number, iface in enumerate(ifaces, start=1): + (fdinfo / str(number)).write_text( + f"pos:\t0\nflags:\t0104002\nmnt_id:\t16\niff:\t{iface}\n" + ) + return str(proc) + + +class _FakeProcess: + """Just enough of asyncio.subprocess.Process for the ownership lookup.""" + + def __init__(self, pid, returncode=None): + self.pid = pid + self.returncode = returncode + + +def _plugin_with_tunnels(service_module, monkeypatch, tmp_path, ips, preexisting): + """A plugin whose interface list and IPs come from `ips` (iface -> IP).""" + monkeypatch.setattr( + service_module, "NET_SYSFS_PATH", _fake_net(tmp_path, ips.keys()) + ) + + async def fake_ipv4(_self, iface): + return ips.get(iface), 24 + + monkeypatch.setattr( + service_module.GpclientVPNPlugin, "_get_iface_ipv4", fake_ipv4 + ) + plugin = service_module.GpclientVPNPlugin() + plugin._preexisting_ifaces = dict(preexisting) + return plugin + + +def _run_loop(plugin, timeout=1.5): + """Run the detection loop; True when it accepted a tunnel and returned.""" + + async def scenario(): + task = asyncio.create_task(plugin._check_tunnel_loop()) + try: + await asyncio.wait_for(task, timeout=timeout) + return True + except asyncio.TimeoutError: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + return False + + return asyncio.run(scenario()) + + +def _ip4_config(signals): + for name, payload in signals: + if name == "Ip4Config": + return payload + return None + + +class TestCandidateDiscovery: + def test_finds_every_tunnel_not_just_the_first_two( + self, service_module, monkeypatch, tmp_path + ): + monkeypatch.setattr( + service_module, + "NET_SYSFS_PATH", + _fake_net(tmp_path, ["lo", "eth0", "tun0", "tun1", "tun2", "gpd0"]), + ) + + assert service_module.list_tunnel_candidates() == [ + "gpd0", + "tun0", + "tun1", + "tun2", + ] + + def test_orders_tunnels_numerically( + self, service_module, monkeypatch, tmp_path + ): + monkeypatch.setattr( + service_module, + "NET_SYSFS_PATH", + _fake_net(tmp_path, ["tun10", "tun2", "tun1"]), + ) + + # Lexical sorting would put tun10 before tun2 + assert service_module.list_tunnel_candidates() == ["tun1", "tun2", "tun10"] + + def test_ignores_look_alike_devices( + self, service_module, monkeypatch, tmp_path + ): + # tunl0 is the always-present IPIP tunnel device, not a VPN tunnel + monkeypatch.setattr( + service_module, + "NET_SYSFS_PATH", + _fake_net(tmp_path, ["tunl0", "tunnelmon", "gpdX", "tun", "tun3"]), + ) + + assert service_module.list_tunnel_candidates() == ["tun3"] + + def test_survives_an_unreadable_sysfs(self, service_module, monkeypatch): + monkeypatch.setattr(service_module, "NET_SYSFS_PATH", "/no/such/path") + + assert service_module.list_tunnel_candidates() == [] + + +class TestSnapshot: + def test_records_all_foreign_tunnels( + self, service_module, monkeypatch, tmp_path + ): + plugin = _plugin_with_tunnels( + service_module, + monkeypatch, + tmp_path, + ips={"tun0": "192.168.1.5", "tun1": "10.8.0.3", "tun2": "10.9.0.3"}, + preexisting={}, + ) + + snapshot = asyncio.run(plugin._snapshot_tunnel_interfaces()) + + # The old 3-name list could only ever record tun0 and tun1 + assert snapshot == { + "tun0": "192.168.1.5", + "tun1": "10.8.0.3", + "tun2": "10.9.0.3", + } + + +class TestDetectionLoop: + def test_detects_a_tunnel_beyond_tun1( + self, service_module, monkeypatch, tmp_path, dbus_signals + ): + """The issue #13 scenario: two foreign VPNs hold tun0 and tun1.""" + plugin = _plugin_with_tunnels( + service_module, + monkeypatch, + tmp_path, + ips={ + "tun0": "192.168.1.5", + "tun1": "10.8.0.3", + "tun2": "10.100.7.42", # ours, brought up by gpclient + }, + preexisting={"tun0": "192.168.1.5", "tun1": "10.8.0.3"}, + ) + + assert _run_loop(plugin) is True + + config = _ip4_config(dbus_signals) + assert config is not None + assert config["tundev"] == ("s", "tun2") + assert config["address"] == ( + "u", + struct.unpack("!I", socket.inet_aton("10.100.7.42"))[0], + ) + assert config["prefix"] == ("u", 24) + assert ( + "StateChanged", + service_module.NM_VPN_SERVICE_STATE_STARTED, + ) in dbus_signals + + def test_still_ignores_pre_existing_tunnels( + self, service_module, monkeypatch, tmp_path, dbus_signals + ): + """Issue #7 must not regress: a foreign tunnel is never adopted.""" + plugin = _plugin_with_tunnels( + service_module, + monkeypatch, + tmp_path, + ips={"tun0": "192.168.1.5", "tun1": "10.8.0.3"}, + preexisting={"tun0": "192.168.1.5", "tun1": "10.8.0.3"}, + ) + + assert _run_loop(plugin) is False + assert dbus_signals == [] + + def test_accepts_a_pre_existing_interface_that_changed_ip( + self, service_module, monkeypatch, tmp_path, dbus_signals + ): + plugin = _plugin_with_tunnels( + service_module, + monkeypatch, + tmp_path, + ips={"gpd0": "10.100.7.42"}, + preexisting={"gpd0": "10.0.0.9"}, + ) + + assert _run_loop(plugin) is True + assert _ip4_config(dbus_signals)["tundev"] == ("s", "gpd0") + + def test_waits_for_an_interface_without_an_ip( + self, service_module, monkeypatch, tmp_path, dbus_signals + ): + plugin = _plugin_with_tunnels( + service_module, + monkeypatch, + tmp_path, + ips={"tun2": None}, + preexisting={}, + ) + + assert _run_loop(plugin) is False + assert dbus_signals == [] + + def test_prefers_the_tunnel_gpclient_holds( + self, service_module, monkeypatch, tmp_path, dbus_signals + ): + """Another VPN coming up mid-connect must not be adopted.""" + plugin = _plugin_with_tunnels( + service_module, + monkeypatch, + tmp_path, + ips={"tun2": "10.8.9.1", "tun3": "10.100.7.42"}, + preexisting={}, + ) + monkeypatch.setattr( + service_module, + "PROC_PATH", + _fake_proc( + tmp_path, + { + 4242: ("gpclient", 1, ["tun3"]), + 4243: ("openvpn", 1, ["tun2"]), + }, + ), + ) + plugin.gpclient_process = _FakeProcess(4242) + + assert _run_loop(plugin) is True + # tun2 sorts first and is equally new, but it is not ours + assert _ip4_config(dbus_signals)["tundev"] == ("s", "tun3") + + def test_falls_back_when_ownership_is_unknown( + self, service_module, monkeypatch, tmp_path, dbus_signals + ): + """No fd information (gpclient already gone) - the snapshot decides.""" + plugin = _plugin_with_tunnels( + service_module, + monkeypatch, + tmp_path, + ips={"tun2": "10.100.7.42"}, + preexisting={}, + ) + monkeypatch.setattr(service_module, "PROC_PATH", _fake_proc(tmp_path, {})) + plugin.gpclient_process = _FakeProcess(4242) + + assert _run_loop(plugin) is True + assert _ip4_config(dbus_signals)["tundev"] == ("s", "tun2") + + +class TestOwnershipLookup: + def test_reads_the_interface_out_of_fdinfo( + self, service_module, monkeypatch, tmp_path + ): + monkeypatch.setattr( + service_module, + "PROC_PATH", + _fake_proc(tmp_path, {77: ("gpclient", 1, ["tun7"])}), + ) + + assert service_module.tunnel_ifaces_held_by([77]) == {"tun7"} + + def test_returns_nothing_for_an_unknown_pid( + self, service_module, monkeypatch, tmp_path + ): + monkeypatch.setattr(service_module, "PROC_PATH", _fake_proc(tmp_path, {})) + + assert service_module.tunnel_ifaces_held_by([77]) == set() + + def test_includes_descendants(self, service_module, monkeypatch, tmp_path): + monkeypatch.setattr( + service_module, + "PROC_PATH", + _fake_proc( + tmp_path, + { + 100: ("gpclient", 1, []), + 101: ("openconnect", 100, ["tun4"]), + 102: ("unrelated", 1, ["tun5"]), + }, + ), + ) + + tree = service_module.process_tree(100) + + assert sorted(tree) == [100, 101] + assert service_module.tunnel_ifaces_held_by(tree) == {"tun4"} + + def test_handles_a_command_name_with_spaces_and_brackets( + self, service_module, monkeypatch, tmp_path + ): + proc = _fake_proc(tmp_path, {200: ("x", 1, []), 201: ("y", 200, ["tun8"])}) + # A comm the naive "split on whitespace" parse would trip over + (tmp_path / "proc" / "201" / "stat").write_text( + "201 (weird ) name) S 200 0 0 0 -1 0\n" + ) + monkeypatch.setattr(service_module, "PROC_PATH", proc) + + assert sorted(service_module.process_tree(200)) == [200, 201] + + def test_reports_nothing_once_gpclient_exited(self, service_module): + plugin = service_module.GpclientVPNPlugin() + plugin.gpclient_process = _FakeProcess(os.getpid(), returncode=0) + + assert plugin._tunnel_ifaces_owned_by_gpclient() == set()