Added support for temporary disconnects and atomic reconnects

src/vpn_manager_globalprotect/backend.py

@@ -23,6 +23,7 @@ from .auth_flows import AuthFlows
 import base64
 import re
 import psutil
+import signal
 
 
 class LoginTarget(StrEnum):
@@ -49,6 +50,7 @@ class Options(TypedDict, total=False):
     allow_insecure_crypto: bool
     spoof_clientos: str
     use_default_browser: bool
+    auth_cache_timeout: int
 
     # If set, will attempt to open the in-browser request using
     # a Firefox "Open external links in a container" plugin URI,
@@ -349,8 +351,36 @@ class GlobalProtectConnection(
             self._wait_task.cancel()
         self._disconnecting = True
         self._proc.terminate()
+        # If the process takes longer than 10 seconds to exit, kill it sith WIGKILL
+        try:
+            async with timeout(10):
+                await self._proc_wait_task
+        except TimeoutError:
+            self._proc.kill()
+            await self._proc_wait_task
+
+    async def temporary_disconnect(self):
+        f"""{super().temporary_disconnect.__doc__}"""
+        if self._proc is None:
+            self.logger.warn(
+                "Cowardly refusing to disconnect from a VPN that is not connected"
+            )
+            return
+        if self._wait_task is not None:
+            self._wait_task.cancel()
+        self._disconnecting = True
+        self._proc.send_signal(signal.SIGHUP)
         await self._proc_wait_task
 
+    async def reconnect(self):
+        f"""{super().reconnect.__doc__}"""
+        if self._proc is None:
+            self.logger.warn(
+                "Cowardly refusing to atomically reconnect to a VPN that is not connected"
+            )
+            return
+        self._proc.send_signal(signal.SIGUSR2)
+
     @classmethod
     def validate_options(cls, options: dict[str, Variant]):
         # options must contain hostname and login target.
@@ -383,6 +413,11 @@ class GlobalProtectConnection(
             != "b"
         ):
             raise errors.InvalidOptions.invalid_type("use_default_browser", "b", sig)
+        if (
+            sig := options.get("auth_cache_timeout", Variant("u", True)).signature
+            != "u"
+        ):
+            raise errors.InvalidOptions.invalid_type("auth_cache_timeout", "u", sig)
         if (
             sig := options.get("firefox_browser_container", Variant("s", "")).signature
             != "s"
@@ -398,6 +433,7 @@ class GlobalProtectConnection(
         cls.put_value(result, "b", options, "verify_certificate")
         cls.put_value(result, "b", options, "allow_insecure_crypto")
         cls.put_value(result, "b", options, "use_default_browser")
+        cls.put_value(result, "u", options, "auth_cache_timeout")
         cls.put_value(result, "s", options, "firefox_browser_container")
         return result
 
@@ -446,6 +482,11 @@ class GlobalProtectConnection(
                 description="Whether to request use of the default browser when authenticating. The server may not allow this.",
                 default=True,
             ),
+            ConfigSpec(
+                name="auth_cache_timeout",
+                signature="u",
+                description="The timeout on caching of authentication results for reconnecting after unexpected or temporary disconnects. This should never be longer than the session timeout. Leaving empty disables caching entirely.",
+            ),
             ConfigSpec(
                 name="firefox_browser_container",
                 signature="s",