fix: handle KS failures for headless server

config/10-protonvpn-wireguard.conf

@@ -0,0 +1,9 @@
+# Allow NetworkManager to manage WireGuard devices created by Proton VPN.
+#
+# On many Linux distributions (notably Ubuntu), NetworkManager is configured
+# to only manage wifi/gsm/cdma devices by default. Without this override,
+# Proton VPN's WireGuard connections fail with "device is strictly unmanaged".
+#
+# This file is installed by python3-proton-vpn-api-core.
+[keyfile]
+unmanaged-devices=*,except:type:wifi,except:type:gsm,except:type:cdma,except:type:wireguard

debian/python3-proton-vpn-api-core.install

@@ -0,0 +1 @@
+config/10-protonvpn-wireguard.conf /etc/NetworkManager/conf.d/

proton/vpn/backend/networkmanager/core/networkmanager.py

@@ -142,11 +142,6 @@ async def start(self):
     async def stop(self, connection=None):
         """Stops the VPN connection."""
         # We directly remove the connection to avoid leaking NM connections.
-        if not self._is_nm_connection_active():
-            self._notify_subscribers(
-                events.Disconnected(EventContext(connection=self))
-            )
-
         connection = connection or self._get_nm_connection()
         if not connection:
             # It can happen that a connection is stopped while checking if the server
@@ -157,6 +152,13 @@ async def stop(self, connection=None):
         else:
             await self.remove_connection(connection)
 
+        # Always fire Disconnected after cleanup. If the NM callback also fires
+        # one via _on_state_changed, the duplicate is harmless (Disconnected
+        # state ignores redundant Disconnected events).
+        self._notify_subscribers(
+            events.Disconnected(EventContext(connection=self))
+        )
+
     async def remove_connection(self, connection=None):
         """Removes the VPN connection."""
         connection = connection or self._get_nm_connection()

proton/vpn/connection/states.py

@@ -43,6 +43,13 @@
 
 logger = logging.getLogger(__name__)
 
+_KS_HEADLESS_WARNING = (
+    "Kill switch cleanup could not be completed (NetworkManager/polkit "
+    "not available). This has no effect when kill switch is disabled. "
+    "If you need kill switch support, ensure NetworkManager is running "
+    "with polkit privileges, or use a desktop environment."
+)
+
 
 @dataclass
 class StateContext:
@@ -180,21 +187,31 @@ async def run_tasks(self):
         if self.context.reconnection:
             # The Kill switch is enabled to avoid leaks when switching servers, even when
             # the kill switch setting is off.
-            await self.context.kill_switch.enable()
+            try:
+                await self.context.kill_switch.enable()
+            except Exception:  # noqa: BLE001
+                if self.context.kill_switch_setting != KillSwitchSetting.OFF:
+                    raise
+                logger.warning(_KS_HEADLESS_WARNING)
 
             # When a reconnection is expected, an Up event is returned to start a new connection.
             # straight away.
             return events.Up(EventContext(connection=self.context.reconnection))
 
-        if self.context.kill_switch_setting == KillSwitchSetting.PERMANENT:
-            # This is an abstraction leak of the network manager KS.
-            # The only reason for enabling permanent KS here is to switch from the
-            # routed KS to the full KS if the user cancels the connection while in
-            # Connecting state. Otherwise, the full KS should already be there.
-            await self.context.kill_switch.enable(permanent=True)
-        else:
-            await self.context.kill_switch.disable()
-            await self.context.kill_switch.disable_ipv6_leak_protection()
+        try:
+            if self.context.kill_switch_setting == KillSwitchSetting.PERMANENT:
+                # This is an abstraction leak of the network manager KS.
+                # The only reason for enabling permanent KS here is to switch from the
+                # routed KS to the full KS if the user cancels the connection while in
+                # Connecting state. Otherwise, the full KS should already be there.
+                await self.context.kill_switch.enable(permanent=True)
+            else:
+                await self.context.kill_switch.disable()
+                await self.context.kill_switch.disable_ipv6_leak_protection()
+        except Exception:  # noqa: BLE001
+            if self.context.kill_switch_setting != KillSwitchSetting.OFF:
+                raise
+            logger.warning(_KS_HEADLESS_WARNING)
 
         if self.context.split_tunneling:
             # ST config is always cleared independently of if ST is disabled via settings or
@@ -254,10 +271,11 @@ async def run_tasks(self):
         # is to avoid leaks when switching servers, even with the kill switch turned off.
         # However, when the kill switch setting is off, the kill switch has to be removed when
         # reaching the connected state.
-        await self.context.kill_switch.enable(
-            self.context.connection.server,
-            permanent=permanent_ks
-        )
+        if self.context.kill_switch_setting != KillSwitchSetting.OFF:
+            await self.context.kill_switch.enable(
+                self.context.connection.server,
+                permanent=permanent_ks
+            )
 
         await self.context.connection.start()
 
@@ -299,9 +317,22 @@ def _on_event(self, event: events.Event):
         return self
 
     async def run_tasks(self):
+        try:
+            if self.context.kill_switch_setting == KillSwitchSetting.OFF:
+                await self.context.kill_switch.enable_ipv6_leak_protection()
+                await self.context.kill_switch.disable()
+            else:
+                # This is specific to the routing table KS implementation and should be removed.
+                # At this point we switch from the routed KS to the full-on KS.
+                await self.context.kill_switch.enable(
+                    permanent=(self.context.kill_switch_setting == KillSwitchSetting.PERMANENT)
+                )
+        except Exception:  # noqa: BLE001
+            if self.context.kill_switch_setting != KillSwitchSetting.OFF:
+                raise
+            logger.warning(_KS_HEADLESS_WARNING)
+
         if self.context.kill_switch_setting == KillSwitchSetting.OFF:
-            await self.context.kill_switch.enable_ipv6_leak_protection()
-            await self.context.kill_switch.disable()
             if self.context.split_tunneling_setting.enabled:
                 try:
                     await self.context.split_tunneling.set_config(
@@ -313,13 +344,6 @@ async def run_tasks(self):
                     # impact the core VPN functionality.
                     logger.exception("Error setting split tunnel configuration")
 
-        else:
-            # This is specific to the routing table KS implementation and should be removed.
-            # At this point we switch from the routed KS to the full-on KS.
-            await self.context.kill_switch.enable(
-                permanent=(self.context.kill_switch_setting == KillSwitchSetting.PERMANENT)
-            )
-
         await self.context.connection.add_persistence()
 
 

proton/vpn/core/connection.py

@@ -189,30 +189,43 @@ async def _apply_kill_switch_setting(self, kill_switch_setting: KillSwitchSettin
         """Enables/disables the kill switch depending on the setting value."""
         kill_switch = self._current_state.context.kill_switch
 
-        if kill_switch_setting == KillSwitchSetting.PERMANENT:
-            await kill_switch.enable(permanent=True)
-            # Since full KS already prevents IPv6 leaks:
-            await kill_switch.disable_ipv6_leak_protection()
-
-        elif kill_switch_setting == KillSwitchSetting.ON:
-            if isinstance(self._current_state, states.Disconnected):
-                await kill_switch.disable()
-                await kill_switch.disable_ipv6_leak_protection()
-            else:
-                await kill_switch.enable(permanent=False)
+        try:
+            if kill_switch_setting == KillSwitchSetting.PERMANENT:
+                await kill_switch.enable(permanent=True)
                 # Since full KS already prevents IPv6 leaks:
                 await kill_switch.disable_ipv6_leak_protection()
 
-        elif kill_switch_setting == KillSwitchSetting.OFF:
-            if isinstance(self._current_state, states.Disconnected):
-                await kill_switch.disable()
-                await kill_switch.disable_ipv6_leak_protection()
-            else:
-                await kill_switch.enable_ipv6_leak_protection()
-                await kill_switch.disable()
+            elif kill_switch_setting == KillSwitchSetting.ON:
+                if isinstance(self._current_state, states.Disconnected):
+                    await kill_switch.disable()
+                    await kill_switch.disable_ipv6_leak_protection()
+                else:
+                    await kill_switch.enable(permanent=False)
+                    # Since full KS already prevents IPv6 leaks:
+                    await kill_switch.disable_ipv6_leak_protection()
+
+            elif kill_switch_setting == KillSwitchSetting.OFF:
+                if isinstance(self._current_state, states.Disconnected):
+                    await kill_switch.disable()
+                    await kill_switch.disable_ipv6_leak_protection()
+                else:
+                    await kill_switch.enable_ipv6_leak_protection()
+                    await kill_switch.disable()
 
-        else:
-            raise RuntimeError(f"Unexpected kill switch setting: {kill_switch_setting}")
+            else:
+                raise RuntimeError(f"Unexpected kill switch setting: {kill_switch_setting}")
+        except RuntimeError:
+            raise
+        except Exception:  # noqa: BLE001
+            if kill_switch_setting != KillSwitchSetting.OFF:
+                raise
+            logger.warning(
+                "Kill switch cleanup could not be completed "
+                "(NetworkManager/polkit not available). This has no effect "
+                "when kill switch is disabled. If you need kill switch "
+                "support, ensure NetworkManager is running with polkit "
+                "privileges, or use a desktop environment."
+            )
 
     async def _apply_split_tunneling_settings(
             self, st_settings: SplitTunnelingSetting, ks_setting: KillSwitchSetting

setup.py

@@ -1,11 +1,51 @@
 #!/usr/bin/env python
 
 from setuptools import setup, find_namespace_packages
+from setuptools.command.install import install
+from setuptools.command.develop import develop
 import re
+import os
+import shutil
+import subprocess
+
 
 VERSIONS = 'versions.yml'
 VERSION = re.search(r'version: (\S+)', open(VERSIONS, encoding='utf-8').readline()).group(1)
 
+NM_CONF_SRC = os.path.join(os.path.dirname(__file__), "config", "10-protonvpn-wireguard.conf")
+NM_CONF_DST = "/etc/NetworkManager/conf.d/10-protonvpn-wireguard.conf"
+
+
+def _install_nm_config():
+    """Install NetworkManager config to manage WireGuard devices."""
+    if os.path.exists(NM_CONF_SRC) and os.path.isdir("/etc/NetworkManager/conf.d"):
+        try:
+            shutil.copy2(NM_CONF_SRC, NM_CONF_DST)
+            subprocess.run(
+                ["systemctl", "reload", "NetworkManager"],
+                check=False, capture_output=True
+            )
+        except PermissionError:
+            print(
+                "\n[proton-vpn-api-core] Could not install NetworkManager config.\n"
+                "Run manually if WireGuard connections fail:\n"
+                f"  sudo cp {NM_CONF_SRC} {NM_CONF_DST}\n"
+                "  sudo systemctl reload NetworkManager\n"
+            )
+
+
+class PostInstall(install):
+    def run(self):
+        super().run()
+        _install_nm_config()
+
+
+class PostDevelop(develop):
+    def run(self):
+        super().run()
+        _install_nm_config()
+
+
 setup(
     name="proton-vpn-api-core",
     version=VERSION,
@@ -36,6 +76,10 @@
         "proton.vpn.backend.networkmanager.killswitch.default*",
         "proton.vpn.backend.networkmanager.killswitch.wireguard*",
     ]),
+    cmdclass={
+        "install": PostInstall,
+        "develop": PostDevelop,
+    },
     entry_points={
         "proton_loader_backend": [
             "linuxnetworkmanager = proton.vpn.backend.networkmanager.core:LinuxNetworkManager",