Improve sing-tun cleanup lifecycle

Signed-off-by: Loren Eteval <loren.eteval@proton.me>

Furious/Core/MultiprocessingRuntime.py

@@ -290,6 +290,9 @@ class MultiprocessingRuntime(CoreRuntime):
 
         self.stop()
 
+        if self.cleanupPending:
+            return
+
         if self._monitorConnection is not None:
             try:
                 QtCore.QObject.disconnect(self._monitorConnection)

Furious/Core/SingTUN.py

@@ -173,6 +173,7 @@ class SingTUN(MultiprocessingRuntime):
         self._started = False
         self._cleanExit = False
         self._nativeWasReady = False
+        self._cleanupPending = False
         self.cleanup = None
         self._configuration = copy.deepcopy(config)
 
@@ -300,9 +301,32 @@ class SingTUN(MultiprocessingRuntime):
 
         return self._status.get('error', '')
 
+    @property
+    def cleanupPending(self):
+        return self._cleanupPending
+
     def _pollProcess(self):
         self._readStatus()
 
+        if (
+            self._hostPlan is not None
+            and not self._stopRequested
+            and not self._exitPublished
+        ):
+            error = self._hostPlan.runtimeError()
+
+            if isinstance(error, str) and error:
+                self._status = {'state': 'failed', 'error': error}
+                self._exitPublished = True
+                self._monitor.stop()
+
+                # Keep the live child owned. The connection owner releases it
+                # and its helper through the normal runtime lease.
+                self.publishExit(
+                    RuntimeExit(None, RuntimeExitReason.Unexpected, message=error)
+                )
+                return
+
         super()._pollProcess()
 
     def interpretExit(self, exitcode, *, requested=False):
@@ -319,8 +343,14 @@ class SingTUN(MultiprocessingRuntime):
         return super().interpretExit(exitcode, requested=requested)
 
     def stop(self):
+        self._stopRequested = True
+
         if self._hostPlan is not None:
-            self._hostPlan.drain()
+            if self._hostPlan.drain() is False:
+                self._cleanupPending = True
+                return
+
+        self._cleanupPending = False
 
         process = self._process
 
@@ -352,17 +382,22 @@ class SingTUN(MultiprocessingRuntime):
         if self._hostPlan is not None:
             try:
                 if self._process is None or not self._process.is_alive():
-                    self._hostPlan.finishCleanup(self._cleanExit or not self._started)
+                    self._cleanupPending = (
+                        self._hostPlan.finishCleanup(
+                            self._cleanExit or not self._started
+                        )
+                        is False
+                    )
                 else:
                     # A retained live child prevents native recovery, but host
                     # DNS snapshots can be restored independently.
-                    self._hostPlan.finishDNSCleanup()
+                    self._cleanupPending = self._hostPlan.finishDNSCleanup() is False
             except Exception as ex:
                 # Any non-exit exceptions
 
                 errors.append(str(ex))
 
-        if callable(self.cleanup):
+        if callable(self.cleanup) and not self._cleanupPending:
             try:
                 self.cleanup()
             except Exception as ex:
@@ -376,5 +411,8 @@ class SingTUN(MultiprocessingRuntime):
     def dispose(self):
         super().dispose()
 
+        if self.cleanupPending:
+            return
+
         self._control.close()
         self._childControl.close()

Furious/Frozenlib/SingTUNRouting.py

@@ -31,6 +31,7 @@ import shlex
 import shutil
 import subprocess
 import tempfile
+import time
 
 __all__ = ['SingTUNLinuxRouting']
 
@@ -67,6 +68,7 @@ def _linuxRuleDelete(flag, rule, table, start, device):
         'oif',
         'goto',
         'suppress_prefixlength',
+        'suppress_prefixlen',
         'dport',
         'action',
         'protocol',
@@ -158,8 +160,14 @@ def _linuxRuleDelete(flag, rule, table, start, device):
 
             args.extend([parameter, value])
 
-    if 'suppress_prefixlength' in rule:
-        if rule['suppress_prefixlength'] != 0:
+    suppression = [
+        rule[key]
+        for key in ('suppress_prefixlen', 'suppress_prefixlength')
+        if key in rule
+    ]
+
+    if suppression:
+        if any(type(value) is not int or value != 0 for value in suppression):
             raise RuntimeError('Conflicting Linux suppression rule')
 
         args.extend(['suppress_prefixlength', '0'])
@@ -315,8 +323,6 @@ def _linuxRuleSelectors(command):
 class SingTUNLinuxRouting:
     """Own the helper process and the persistent TUN created for one attempt."""
 
-    AuthorizationTimeout = 60
-
     @staticmethod
     def checkEnvironment():
         """Check Linux tools and the resolver before host preparation."""
@@ -381,6 +387,8 @@ class SingTUNLinuxRouting:
         self.ranges = _linuxRouteRanges(tun)
         self._process = None
         self._directory = None
+        self._holding = False
+        self._ready = False
 
     def cancel(self):
         """Prevent a pending authorization from acquiring more host resources."""
@@ -393,7 +401,24 @@ class SingTUNLinuxRouting:
                 # The worker already reaped the helper and removed its directory.
                 pass
 
-    def drain(self):
+        if self._holding and not self._ready:
+            self._releaseHelper()
+
+    def _releaseHelper(self):
+        """EOF requests the fixed cleanup script, never an arbitrary command."""
+        if self._holding:
+            self._holding = False
+
+            if self._process is not None and self._process.stdin is not None:
+                self._process.stdin.close()
+
+    def runtimeError(self):
+        """Report loss of an authorized helper while forwarding is active."""
+        if self._holding and self._ready and self._process.poll() is not None:
+            return 'sing-tun privileged helper exited unexpectedly'
+        return ''
+
+    def drain(self, *, wait=True):
         """Retain an unfinished helper and its script until all commands exit."""
         process = self._process
 
@@ -402,9 +427,16 @@ class SingTUNLinuxRouting:
                 self._directory.cleanup()
                 self._directory = None
 
-            return
+            return True
+
+        if self._holding and self._ready and process.poll() is None:
+            # The authorized helper stays with the live forwarding runtime.
+            return True
 
         if process.poll() is None:
+            if not wait:
+                return False
+
             try:
                 process.wait(timeout=5)
             except subprocess.TimeoutExpired:
@@ -415,11 +447,17 @@ class SingTUNLinuxRouting:
         self._process = None
         self._directory.cleanup()
         self._directory = None
+        self._holding = False
+        self._ready = False
+        return True
 
-    def execute(self, commands, *, preparing=False, guard=False):
+    def execute(self, commands, *, preparing=False, guard=False, holding=False):
         """Run only argument vectors generated for this attempt, outside Qt."""
         self.drain()
 
+        if self._process is not None:
+            raise RuntimeError('sing-tun privileged helper is already active')
+
         if preparing and self.cancelled.is_set():
             raise RuntimeError('sing-tun host preparation was cancelled')
 
@@ -431,6 +469,9 @@ class SingTUNLinuxRouting:
 
         lines = ['set -eu']
 
+        if holding:
+            lines.extend(self._sessionCleanupLines())
+
         if guard:
             name = self.tun['Name']
             index = shlex.quote('/sys/class/net/' + name + '/ifindex')
@@ -453,13 +494,33 @@ class SingTUNLinuxRouting:
                 index = shlex.quote('/sys/class/net/' + name + '/ifindex')
                 lines.append(f'createdIndex=$(cat {index})')
 
-                # Roll back device creation if a later setup command fails.
-                rollback = (
-                    'code=$?; if [ "$code" -ne 0 ] '
-                    f'&& [ "$(cat {index} 2>/dev/null)" = "$createdIndex" ]; then '
-                    'ip link delete dev ' + shlex.quote(name) + '; fi'
-                )
-                lines.append('trap ' + shlex.quote(rollback) + ' EXIT')
+                if not holding:
+                    # Roll back device creation if a later setup command fails.
+                    rollback = (
+                        'code=$?; if [ "$code" -ne 0 ] '
+                        f'&& [ "$(cat {index} 2>/dev/null)" = "$createdIndex" ]; then '
+                        'ip link delete dev ' + shlex.quote(name) + '; fi'
+                    )
+                    lines.append('trap ' + shlex.quote(rollback) + ' EXIT')
+
+            if holding and arguments[2:4] == ['rule', 'add']:
+                lines.append('rulesAdded=$((rulesAdded + 1))')
+
+        if holding:
+            ready = pathlib.Path(directory.name, 'ready')
+            staging = pathlib.Path(directory.name, 'ready.tmp')
+            lines.extend(
+                [
+                    'printf "%s\\n" "$createdIndex" > ' + shlex.quote(str(staging)),
+                    'mv -- '
+                    + shlex.quote(str(staging))
+                    + ' '
+                    + shlex.quote(str(ready)),
+                    # The pipe also closes if Furious crashes. No second
+                    # authorization is needed to restore this attempt's state.
+                    'read -r release || :',
+                ]
+            )
 
         with tempfile.NamedTemporaryFile(
             mode='w',
@@ -479,17 +540,33 @@ class SingTUNLinuxRouting:
         try:
             self._process = subprocess.Popen(
                 arguments,
-                stdin=subprocess.DEVNULL,
+                stdin=subprocess.PIPE if holding else subprocess.DEVNULL,
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
             )
 
-            code = self._process.wait(timeout=self.AuthorizationTimeout)
-        except subprocess.TimeoutExpired:
-            # Do not drop a root process that the unprivileged owner cannot kill.
-            # Its cancellation file and script survive until a later drain succeeds.
-            self.cancel()
-            raise RuntimeError('sing-tun privileged helper timed out') from None
+            self._holding = holding
+
+            if holding:
+                ready = pathlib.Path(directory.name, 'ready')
+
+                while self._process.poll() is None:
+                    if self.cancelled.is_set():
+                        self.cancel()
+                        raise RuntimeError('sing-tun host preparation was cancelled')
+
+                    if ready.exists():
+                        self.index = int(ready.read_text(encoding='utf-8').strip())
+                        self._ready = True
+                        return
+
+                    time.sleep(0.05)
+
+                self._releaseHelper()
+                code = self._process.wait()
+            else:
+                # Match tun2socks: a PolicyKit interaction has no deadline.
+                code = self._process.wait()
         except OSError:
             if self._process is None:
                 directory.cleanup()
@@ -506,7 +583,44 @@ class SingTUNLinuxRouting:
                 'sing-tun privileged host operation failed or authorization was cancelled'
             )
 
-    def prepare(self):
+    def _sessionCleanupLines(self):
+        """Pair successful additions with exact deletions in the same helper."""
+        name = shlex.quote(self.tun['Name'])
+        index = shlex.quote('/sys/class/net/' + self.tun['Name'] + '/ifindex')
+        lines = [
+            'createdIndex=; rulesAdded=0',
+            'cleanup() {',
+            'code=$?; trap - EXIT; set +e; failed=0',
+        ]
+
+        for number, rule in reversed(list(enumerate(self.rules, 1))):
+            deletion = rule[:]
+            deletion[3] = 'del'
+            command = ' '.join(shlex.quote(str(value)) for value in deletion)
+            lines.append(
+                f'if [ "$rulesAdded" -ge {number} ]; then {command} || failed=1; fi'
+            )
+
+        lines.extend(
+            [
+                'if [ -n "$createdIndex" ] && [ -e ' + index + ' ]; then',
+                f'if [ "$(cat {index})" = "$createdIndex" ]; then',
+                # resolved discards per-link DNS when the link disappears,
+                # including when DNS application failed partway through.
+                f'resolvectl revert {name} || failed=1',
+                f'ip link delete dev {name} || failed=1',
+                'else failed=1; fi',
+                'fi',
+                '[ "$failed" -eq 0 ] || code=1',
+                'exit "$code"',
+                '}',
+                'trap cleanup EXIT',
+                "trap 'exit 1' HUP INT TERM",
+            ]
+        )
+        return lines
+
+    def prepare(self, dnsAddress='1.1.1.1'):
         name = self.tun['Name']
 
         # Recheck inside the privileged script before asking ip to create a TUN.
@@ -552,8 +666,15 @@ class SingTUNLinuxRouting:
             )
 
         commands.extend(self.rules)
+        commands.extend(
+            [
+                ['resolvectl', 'dns', name, dnsAddress],
+                ['resolvectl', 'domain', name, '~.'],
+            ]
+        )
 
-        self.execute(commands, preparing=True)
+        self.execute(commands, preparing=True, holding=True)
+        self.dnsTouched = True
 
         link = self.link()
 
@@ -584,7 +705,13 @@ class SingTUNLinuxRouting:
         return link
 
     def applyDNS(self, address):
-        self.link()
+        if self.link() is None:
+            raise RuntimeError('sing-tun interface disappeared before DNS confirmation')
+
+        if self._holding and self._ready:
+            # Linux DNS was configured alongside the interface/routes under
+            # the original authorization. Native readiness only confirms it.
+            return
 
         self.dnsTouched = True
 
@@ -601,12 +728,20 @@ class SingTUNLinuxRouting:
         if not self.dnsTouched:
             return
 
+        if self._process is not None:
+            # Do not release host routes while the forwarding child is alive.
+            # cleanup() closes the helper pipe after the child is reaped.
+            raise RuntimeError(
+                'sing-tun DNS restoration requires stopping the forwarding child'
+            )
+
         if self.link() is not None:
             self.execute([['resolvectl', 'revert', self.tun['Name']]], guard=True)
 
         self.dnsTouched = False
 
     def cleanup(self):
+        self._releaseHelper()
         self.drain()
         link = self.link()
 
@@ -618,6 +753,9 @@ class SingTUNLinuxRouting:
         commands = []
         expected = set()
 
+        if self.dnsTouched and link is not None:
+            commands.append(['resolvectl', 'revert', self.tun['Name']])
+
         for rule in self.rules:
             deletion = rule[:]
             deletion[3] = 'del'
@@ -648,6 +786,8 @@ class SingTUNLinuxRouting:
         if self.link() is not None:
             raise RuntimeError('sing-tun interface remains; cleanup ownership retained')
 
+        self.dnsTouched = False
+
         for flag in ('-4', '-6'):
             rules = json.loads(_command(['ip', flag, '-N', '-j', 'rule', 'show']))
 

Furious/Interface/Runtime.py

@@ -200,6 +200,11 @@ class CoreRuntime(ABC):
         """Passively report execution liveness without consuming an exit."""
         return self.state is RuntimeState.Alive
 
+    @property
+    def cleanupPending(self) -> bool:
+        """Report owned asynchronous cleanup; stop/dispose may be polled again."""
+        return False
+
     @staticmethod
     @abstractmethod
     def name() -> str:
@@ -234,6 +239,9 @@ class CoreRuntime(ABC):
         ):
             self.stop()
 
+        if self.cleanupPending:
+            return
+
         self._exitCallback = None
         self._disposed = True
         self.setState(RuntimeState.Disposed)

Furious/Service/ConnectionManager.py

@@ -334,7 +334,7 @@ class _RuntimeReadinessProbe(QtCore.QObject):
 
 
 class _ConditionProbe(QtCore.QObject):
-    """Poll one host condition through a bounded parent-owned Qt timer."""
+    """Poll one host condition through a parent-owned Qt timer."""
 
     finished = QtCore.Signal(bool)
 
@@ -344,7 +344,7 @@ class _ConditionProbe(QtCore.QObject):
 
         self._predicate = predicate
         self._description = description
-        self._timeout = max(int(timeout), 1)
+        self._timeout = None if timeout is None else max(int(timeout), 1)
         self._terminal = False
 
         self._elapsed = QtCore.QElapsedTimer()
@@ -383,7 +383,11 @@ class _ConditionProbe(QtCore.QObject):
             logger.info(f'find {self._description} success')
 
             self._finish(True)
-        elif self._elapsed.isValid() and self._elapsed.elapsed() >= self._timeout:
+        elif (
+            self._timeout is not None
+            and self._elapsed.isValid()
+            and self._elapsed.elapsed() >= self._timeout
+        ):
             logger.error(f'find {self._description} failed')
 
             self._finish(False)
@@ -900,11 +904,7 @@ class ConnectionStartOperation(QtCore.QObject):
             continuation,
             self.stage,
             'sing-tun host preparation timed out',
-            timeout=(
-                SingTUNHostPlan.LinuxHostOperationTimeout
-                if PLATFORM == 'Linux'
-                else 10000
-            ),
+            timeout=None if PLATFORM == 'Linux' else 10000,
         )
 
     def _startSingTUN(self):
@@ -1377,6 +1377,66 @@ class ConnectionStartOperation(QtCore.QObject):
         return True
 
 
+def _cleanupDestroyedStopOperation(reference, *_args):
+    operation = reference()
+    if operation is not None:
+        operation._terminal = True
+        if operation.manager._activeStopOperation is operation:
+            operation.manager._activeStopOperation = None
+
+
+class ConnectionStopOperation(QtCore.QObject):
+    """Observe manager-owned cleanup without timing out authorization."""
+
+    finished = QtCore.Signal(object, str)
+
+    def __init__(self, manager):
+        super().__init__()
+        self.manager = manager
+        self._terminal = False
+        self._timer = QtCore.QTimer(self)
+        self._timer.setInterval(100)
+        connectWeakly(self._timer.timeout, self, '_poll')
+        self.destroyed.connect(
+            functools.partial(_cleanupDestroyedStopOperation, weakref.ref(self))
+        )
+
+    def start(self):
+        if self._terminal or self._timer.isActive():
+            return
+        self._timer.start()
+        self._poll()
+
+    def _poll(self):
+        if self._terminal:
+            return
+        try:
+            complete = self.manager._pollPendingReleases()
+        except Exception as ex:
+            # Any non-exit exceptions
+            if isValid(self):
+                self._finish(str(ex) or type(ex).__name__)
+            return
+
+        if not isValid(self) or self._terminal:
+            return
+        if complete:
+            self._finish('')
+        elif any(lease.releaseError for lease in self.manager._pendingReleases):
+            self._finish('Connection runtime cleanup is incomplete')
+
+    def _finish(self, error):
+        self._terminal = True
+        self._timer.stop()
+        if self.manager._activeStopOperation is self:
+            self.manager._activeStopOperation = None
+        try:
+            self.finished.emit(self, error)
+        finally:
+            if isValid(self):
+                self.deleteLater()
+
+
 class ConnectionManager(Mixins.CleanupOnExit):
     """Coordinate proxy cores, TUN setup, DNS changes, and routing cleanup."""
 
@@ -1392,6 +1452,7 @@ class ConnectionManager(Mixins.CleanupOnExit):
         self._lastStartError = ''
         self._startGeneration = 0
         self._activeStartOperation = None
+        self._activeStopOperation = None
 
     @property
     def runtimes(self):
@@ -1539,6 +1600,8 @@ class ConnectionManager(Mixins.CleanupOnExit):
         third-party plugins and non-interactive callers that rely on historical
         synchronous launch semantics.
         """
+        if self._activeStopOperation is not None and isValid(self._activeStopOperation):
+            raise RuntimeError('Connection runtime cleanup is incomplete')
         self._retryPendingReleases()
 
         self._lastStartError = ''
@@ -1578,6 +1641,8 @@ class ConnectionManager(Mixins.CleanupOnExit):
         **kwargs,
     ):
         """Create and schedule one manager-owned startup transaction."""
+        if self._activeStopOperation is not None and isValid(self._activeStopOperation):
+            raise RuntimeError('Connection runtime cleanup is incomplete')
         if self._activeStartOperation is not None:
             self._activeStartOperation.cancel()
 
@@ -2180,12 +2245,43 @@ class ConnectionManager(Mixins.CleanupOnExit):
 
     def _retryPendingReleases(self):
         """Keep exact failed resources and refuse replacement until release succeeds."""
+        if not self._pollPendingReleases():
+            raise RuntimeError('Connection runtime cleanup is incomplete')
+
+    def _pollPendingReleases(self):
         self._pendingReleases[:] = [
             lease for lease in self._pendingReleases if not lease.release()
         ]
+        return not self._pendingReleases
 
-        if self._pendingReleases:
-            raise RuntimeError('Connection runtime cleanup is incomplete')
+    def stopAsync(self):
+        """Keep pending cleanup leases owned and observe their completion in Qt."""
+        if self._activeStopOperation is not None and isValid(self._activeStopOperation):
+            return self._activeStopOperation
+
+        try:
+            operation = ConnectionStopOperation(self)
+        except BaseException:
+            self._retainStopLeases()
+            raise
+        self._activeStopOperation = operation
+        try:
+            self._retainStopLeases()
+            self.cancelStart()
+        except BaseException:
+            self._retainStopLeases()
+            self._activeStopOperation = None
+            operation.deleteLater()
+            raise
+
+        singleShotWeakly(0, operation, 'start')
+        return operation
+
+    def _retainStopLeases(self):
+        self._pendingReleases.extend(reversed(self._leases))
+        self._leases.clear()
+        for lease in self._pendingReleases:
+            lease.router.beginRelease()
 
     def cleanup(self):
         """Release resources owned by the core manager."""

Furious/Service/RuntimeLease.py

@@ -179,6 +179,7 @@ class RuntimeLease:
         self.router = router
         self.applicationTUN = applicationTUN
         self._releaseInProgress = False
+        self.releaseError = ''
 
     @property
     def state(self):
@@ -199,6 +200,7 @@ class RuntimeLease:
 
         self.router.beginRelease()
         self._releaseInProgress = True
+        self.releaseError = ''
 
         stopped, disposed = True, True
 
@@ -209,29 +211,38 @@ class RuntimeLease:
                 # Any non-exit exceptions
 
                 stopped = False
+                self.releaseError = str(ex) or type(ex).__name__
 
                 logger.error(f'error stopping core runtime: {ex}')
 
+            if not stopped or self.runtime.cleanupPending:
+                # Preserve the whole runtime until stop succeeds. Disposal must
+                # not implicitly restart a refused authorization in this call.
+                return False
+
             try:
                 self.runtime.dispose()
             except Exception as ex:
                 # Any non-exit exceptions
 
                 disposed = False
+                self.releaseError = str(ex) or type(ex).__name__
 
                 logger.error(f'error disposing core runtime: {ex}')
 
-            if not stopped or not disposed:
+            if not disposed or self.runtime.cleanupPending:
                 return False
 
             try:
                 if self.runtime.isRunning():
+                    self.releaseError = 'core runtime remains alive after disposal'
                     logger.error('core runtime remains alive after disposal')
 
                     return False
             except Exception as ex:
                 # Any non-exit exceptions
 
+                self.releaseError = str(ex) or type(ex).__name__
                 logger.error(f'could not verify core runtime shutdown: {ex}')
 
                 return False

Furious/Service/SingTUNHost.py

@@ -143,8 +143,6 @@ def _routeRanges(tun):
 class SingTUNHostPlan:
     """Owned by one SingTUN runtime, including any unfinished host worker."""
 
-    LinuxHostOperationTimeout = (SingTUNLinuxRouting.AuthorizationTimeout + 5) * 1000
-
     def __init__(self, configuration, *, platform=PLATFORM):
         copied = copy.deepcopy(configuration)
         proxy = copied.pop('proxy', None)
@@ -168,9 +166,10 @@ class SingTUNHostPlan:
         self._cleanupComplete = False
         self.egressIP = ''
         self._linux = None
+        self._cleanupMethod = None
 
     def begin(self, method, *args):
-        """Schedule bounded host work; the Qt owner polls completion."""
+        """Schedule owned host work; the Qt owner polls completion."""
         if self._worker is not None and self._worker.is_alive():
             raise RuntimeError('sing-tun host worker is already active')
 
@@ -184,12 +183,18 @@ class SingTUNHostPlan:
             except Exception as ex:
                 # Any non-exit exceptions
 
-                self.error = str(ex)
+                self.error = str(ex) or type(ex).__name__
             finally:
                 self._done.set()
 
         self._worker = threading.Thread(target=run, name='sing-tun-host')
-        self._worker.start()
+        try:
+            self._worker.start()
+        except BaseException:
+            self._worker = None
+            self._cleanupMethod = None
+            self._done.set()
+            raise
 
     @property
     def workDone(self):
@@ -198,23 +203,34 @@ class SingTUNHostPlan:
     def workCompleted(self):
         return self.workDone
 
+    def runtimeError(self):
+        return self._linux.runtimeError() if self._linux is not None else ''
+
     def drain(self):
         """Retain a worker on timeout; no late DNS mutation can escape cleanup."""
+        if self._cleanupMethod is not None:
+            # Restoration owns the worker now; never cancel it as preparation.
+            return True
+
         self._cancel.set()
 
         if self._linux is not None:
             self._linux.cancel()
 
         if self._worker is not None:
-            self._worker.join(6)
+            self._worker.join(0 if self.platform == 'Linux' else 6)
 
             if self._worker.is_alive():
+                if self.platform == 'Linux':
+                    return False
                 raise RuntimeError('sing-tun host preparation is still stopping')
 
             self._worker = None
 
         if self._linux is not None:
-            self._linux.drain()
+            return self._linux.drain(wait=self.platform != 'Linux')
+
+        return True
 
     def prepare(self, addresses):
         if SystemRuntime.flatpakID():
@@ -262,7 +278,17 @@ class SingTUNHostPlan:
             self._linux = SingTUNLinuxRouting(tun, table, start, self._cancel)
             self._activated = True
 
-            self._linux.prepare()
+            dns = self.hostSettingsCallers.userTunAdapterInterfaceDNS() or (
+                '1.1.1.1' if tun['Inet4Address'] else '2606:4700:4700::1111'
+            )
+            address = ipaddress.ip_address(dns)
+
+            if not tun['Inet' + str(address.version) + 'Address']:
+                raise ValueError(
+                    'The configured TUN DNS address requires that address family on the interface'
+                )
+
+            self._linux.prepare(str(address))
 
             # Linux host changes belong to the helper; the spawned binding only
             # opens and forwards through the persistent TUN owned by this user.
@@ -517,11 +543,14 @@ class SingTUNHostPlan:
             raise RuntimeError('; '.join(errors))
 
     def cleanup(self, cleanExit):
-        """Restore independent DNS even if exact native recovery refuses cleanup."""
+        """Restore host state, with Linux DNS owned by its authorized helper."""
         errors = []
 
         try:
-            self.restoreDNS()
+            if self._linux is None:
+                self.restoreDNS()
+            # Linux's authorized helper restores DNS with its interface/rules
+            # after native release, through _recoverNative below.
         except Exception as ex:
             # Any non-exit exceptions
 
@@ -542,33 +571,47 @@ class SingTUNHostPlan:
         self._cleanupComplete = True
 
     def finishCleanup(self, cleanExit):
-        """Run host restoration outside Qt; retain the worker on a bounded drain."""
-        if self._cleanupComplete:
-            return
+        """Schedule Linux restoration and report completion without joining Qt."""
+        if self._cleanupComplete and self._worker is None:
+            return True
 
-        self._finishCleanupWork('cleanup', cleanExit)
+        return self._finishCleanupWork('cleanup', cleanExit)
 
     def finishDNSCleanup(self):
         """Restore DNS even when the child cannot be reaped; retain native ownership."""
-        if self._dnsRestore or (self._linux is not None and self._linux.dnsTouched):
-            self._finishCleanupWork('restoreDNS')
+        if (
+            self._cleanupMethod == 'restoreDNS'
+            or self._dnsRestore
+            or (self._linux is not None and self._linux.dnsTouched)
+        ):
+            return self._finishCleanupWork('restoreDNS')
+        return True
 
     def _finishCleanupWork(self, method, *args):
         if self._worker is None:
             # drain() cancelled preparation; restoration must still execute.
             self._cancel.clear()
+            self._cleanupMethod = method
             self.begin(method, *args)
 
-        self._worker.join(6)
+        self._worker.join(0 if self.platform == 'Linux' else 6)
 
         if self._worker.is_alive():
+            if self.platform == 'Linux':
+                return False
             raise RuntimeError('sing-tun host restoration is still running')
 
+        completedMethod = self._cleanupMethod
         self._worker = None
+        self._cleanupMethod = None
 
         if self.error:
             raise RuntimeError(self.error)
 
+        if completedMethod is not None and completedMethod != method:
+            return self._finishCleanupWork(method, *args)
+        return True
+
     def _recoverNative(self):
         if self._linux is not None:
             self._linux.cleanup()

tests/test_connection_startup_async.py

@@ -178,6 +178,41 @@ class ConnectionStartupAsyncTest(TestCase):
             self.assertFalse(isValid(probe))
             self.assertFalse(isValid(probe._timer))
 
+    def testConditionDeadlineCanBeDisabledWithoutChangingOrdinaryTimeouts(self):
+        """Only explicitly unbounded observers survive a long pending condition."""
+        for timeout in (None, 10000):
+            with self.subTest(timeout=timeout):
+                owner = QtCore.QObject()
+                ready, results = [False], []
+                probe = _ConditionProbe(
+                    lambda: ready[0], 'fixture', timeout=timeout, parent=owner
+                )
+                probe.finished.connect(results.append)
+
+                try:
+                    probe.start()
+                    probe._elapsed = mock.Mock()
+                    probe._elapsed.isValid.return_value = True
+                    probe._elapsed.elapsed.return_value = 86400000
+                    probe._poll()
+
+                    if timeout is None:
+                        self.assertEqual(results, [])
+                        self.assertTrue(probe._timer.isActive())
+                        ready[0] = True
+                        probe._poll()
+                        self.assertEqual(results, [True])
+                    else:
+                        self.assertEqual(results, [False])
+
+                    self.assertFalse(probe._timer.isActive())
+                finally:
+                    probe.cancel()
+                    deleteQObject(owner)
+
+                self.assertFalse(isValid(probe))
+                self.assertFalse(isValid(probe._timer))
+
     def testNativeDestructionDuringStagesReleasesUncommittedRuntime(self):
         """Qt destruction ends acquisition and rolls back before child deletion."""
         for stage in (ConnectionStartStage.Preparing, ConnectionStartStage.Committing):

tests/test_external_core.py

@@ -541,6 +541,7 @@ class ExternalCoreProcessTest(unittest.TestCase):
             dnsResolver.resolve.return_value = (True, [])
 
             tunRuntime = mock.Mock(spec=CoreRuntime)
+            tunRuntime.cleanupPending = False
             tunRuntime.isRunning.return_value = False
 
             manager = NoCoreRuntimeConnectionManager(dnsResolver=dnsResolver)

tests/test_runtime_lifecycle.py

@@ -32,6 +32,7 @@ from Furious.Service.RuntimeLease import (
 from Furious.Service.ConnectionManager import ConnectionManager, _ConnectionStartAttempt
 
 from PySide6 import QtCore
+from shiboken6 import isValid, delete as deleteQObject
 
 from tests.support import application, collectAtBoundary, processQtEvents, waitFor
 
@@ -104,6 +105,41 @@ class RuntimeLifecycleTest(TestCase):
     def setUpClass(cls):
         cls.app = application()
 
+    def testCleanupCompletionObserverCanDestroyItsOperation(self):
+        """Parent-owned timers die even if a completion listener deletes the owner."""
+        for _ in range(25):
+            manager = ConnectionManager()
+            runtime = _Runtime()
+            manager._leases.append(RuntimeLease(runtime, RuntimeEventRouter()))
+            operation = manager.stopAsync()
+            operation.finished.connect(lambda current, _error: deleteQObject(current))
+            with mock.patch('sys.excepthook') as qtErrors:
+                operation.start()
+                processQtEvents()
+                qtErrors.assert_not_called()
+            self.assertFalse(isValid(operation))
+            self.assertFalse(isValid(operation._timer))
+            self.assertIsNone(manager._activeStopOperation)
+            self.assertEqual(manager._pendingReleases, [])
+            manager.cleanup()
+
+    def testFailedStopObserverConstructionKeepsRuntimesOwnedAndBlocksReplacement(self):
+        manager = ConnectionManager()
+        runtime = _Runtime()
+        lease = RuntimeLease(runtime, RuntimeEventRouter())
+        manager._leases.append(lease)
+        with mock.patch(
+            'Furious.Service.ConnectionManager.ConnectionStopOperation',
+            side_effect=RuntimeError('observer factory'),
+        ):
+            with self.assertRaisesRegex(RuntimeError, 'observer factory'):
+                manager.stopAsync()
+        self.assertEqual(manager.runtimes, [])
+        self.assertEqual(manager._pendingReleases, [lease])
+        self.assertEqual(lease.state, RuntimeLeaseState.Releasing)
+        self.assertIsNone(manager._activeStopOperation)
+        manager.cleanup()
+
     def testFailedLeaseReleaseRemainsOwnedUntilSuccessfulRetry(self):
         """Rollback retains failed resources and blocks replacement admission."""
         manager = ConnectionManager()

tests/test_sing_tun.py

@@ -40,7 +40,7 @@ from Furious.Service.ConnectionManager import (
     ConnectionStartStage,
     _ConnectionStartAttempt,
 )
-from Furious.Frozenlib.SingTUNRouting import _linuxRuleDelete
+from Furious.Frozenlib.SingTUNRouting import SingTUNLinuxRouting, _linuxRuleDelete
 from Furious.Service.SingTUNHost import (
     SingTUNHostPlan,
     _routeRanges,
@@ -56,7 +56,11 @@ from Furious.Controllers.ConnectionController import (
     ConnectionState,
 )
 from Furious.Qt import AppQDialogButtonBox, gettext as _
-from Furious.Service.RuntimeLease import RuntimeEventRouter, RuntimeLeaseState
+from Furious.Service.RuntimeLease import (
+    RuntimeEventRouter,
+    RuntimeLease,
+    RuntimeLeaseState,
+)
 from Furious.Service.LogManager import LogManager, TUN_LOG_CATEGORY
 
 from PySide6 import QtCore
@@ -154,6 +158,7 @@ class _PreparedSing(SingTUN):
         self._hostPlan = hostPlan
         self._configuration = copy.deepcopy(configuration)
         self.alive = False
+        self._cleanupPending = False
         self.nativeReady = False
         self.released = False
 
@@ -1505,6 +1510,104 @@ class SingTUNChildTest(unittest.TestCase):
         finally:
             runtime.dispose()
 
+    def testUnexpectedHelperExitPublishesOnceAndKeepsTheLiveChildOwned(self):
+        plan = SingTUNHostPlan({}, platform='Linux')
+        tun = plan.configuration['tun_options']
+        tun['Name'] = 'utun101'
+        plan._linux = SingTUNLinuxRouting(tun, 123456, 2000, threading.Event())
+        helper = mock.Mock()
+        helper.poll.return_value = 1
+        plan._linux._process = helper
+        plan._linux._holding = plan._linux._ready = True
+        child = mock.Mock()
+        child.is_alive.return_value = True
+        events = []
+        runtime = SingTUN(
+            {}, hostPlan=plan, exitCallback=lambda owner, event: events.append(event)
+        )
+        runtime._process = child
+
+        try:
+            runtime._pollProcess()
+            runtime._pollProcess()
+            self.assertEqual(len(events), 1)
+            self.assertEqual(events[0].reason, RuntimeExitReason.Unexpected)
+            self.assertIn('helper exited', events[0].message)
+            self.assertIs(runtime.process, child)
+            self.assertTrue(runtime.isRunning())
+            self.assertFalse(runtime.ready)
+        finally:
+            runtime._process = None
+            runtime._hostPlan = None
+            runtime.dispose()
+
+    def testKernelRuleCleanupAllowsSchedulingTheOtherBackend(self):
+        """Retry the failed sing-tun lease before admitting a tun2socks attempt."""
+        module = importlib.import_module('Furious.Service.ConnectionManager')
+        routing = importlib.import_module('Furious.Frozenlib.SingTUNRouting')
+        plan = SingTUNHostPlan({}, platform='Linux')
+        tun = plan.configuration['tun_options']
+        tun['Name'] = 'utun101'
+        plan._linux = SingTUNLinuxRouting(tun, 123456, 2000, threading.Event())
+        plan._linux.index = 77
+        plan._activated = True
+        present, refused = [True], [True]
+
+        def command(arguments):
+            if not present[0]:
+                return '[]'
+            if arguments == ['ip', '-N', '-j', 'link', 'show']:
+                return '[{"ifname":"utun101","ifindex":77}]'
+            if arguments == ['ip', '-4', '-N', '-j', 'rule', 'show']:
+                return '[{"priority":2001,"src":"all","table":123456,"suppress_prefixlen":0,"protocol":0}]'
+            return '[]'
+
+        def release(commands, **kwargs):
+            if refused[0]:
+                raise RuntimeError('fixture refuses cleanup')
+            self.assertIn('suppress_prefixlength', commands[0])
+            present[0] = False
+
+        with isolatedSettings(), mock.patch.object(
+            routing, '_command', side_effect=command
+        ), mock.patch.object(
+            plan._linux, 'execute', side_effect=release
+        ), mock.patch.object(
+            module.ConnectionStartOperation, 'start'
+        ):
+            manager = ConnectionManager()
+            runtime = SingTUN({}, hostPlan=plan)
+            router = RuntimeEventRouter()
+            router.attach(runtime, manager)
+            lease = RuntimeLease(runtime, router, applicationTUN=True)
+            manager._leases.append(lease)
+
+            try:
+                with self.assertRaisesRegex(RuntimeError, 'cleanup is incomplete'):
+                    manager.stopAll()
+                self.assertEqual(manager._pendingReleases, [lease])
+
+                refused[0] = False
+                AppSettings.set('ApplicationTUNBackend', 'tun2socks')
+                cleanup = manager.stopAsync()
+                cleanup.start()
+                self.assertTrue(waitFor(lambda: cleanup._terminal))
+                operation = manager.startAsync(
+                    _Configuration(), 'Global', deepcopy=False
+                )
+                self.assertEqual(manager._pendingReleases, [])
+                self.assertEqual(lease.state, RuntimeLeaseState.Released)
+                self.assertEqual(operation.attempt.tunBackend, 'tun2socks')
+                operation.cancel()
+            finally:
+                refused[0] = False
+                if manager._pendingReleases:
+                    cleanup = manager.stopAsync()
+                    cleanup.start()
+                    self.assertTrue(waitFor(lambda: cleanup._terminal))
+                manager.cleanup()
+                processQtEvents()
+
     def testFailedHandleCloseStillRestoresHostAndRetainsRetryOwnership(self):
         runtime = _ChildRuntime({})
         plan = mock.Mock()
@@ -1558,6 +1661,32 @@ class SingTUNChildTest(unittest.TestCase):
 
 
 class SingTUNHostTest(unittest.TestCase):
+    def testCleanupWorkerStartFailureCanBeRetried(self):
+        plan = SingTUNHostPlan({}, platform='Linux')
+        module = importlib.import_module('Furious.Service.SingTUNHost')
+        worker = mock.Mock()
+        worker.start.side_effect = RuntimeError('worker start')
+        with mock.patch.object(module.threading, 'Thread', return_value=worker):
+            with self.assertRaisesRegex(RuntimeError, 'worker start'):
+                plan.finishCleanup(True)
+        self.assertIsNone(plan._worker)
+        self.assertIsNone(plan._cleanupMethod)
+        self.assertTrue(waitFor(lambda: plan.finishCleanup(True)))
+        self.assertTrue(plan._cleanupComplete)
+        self.assertIsNone(plan._worker)
+
+    def testLinuxRecoveryPollsTheAuthorizationWorkerWithoutBlocking(self):
+        """Polling cleanup adds neither a GUI wait nor an authorization deadline."""
+        for platform in ('Linux', 'Windows'):
+            with self.subTest(platform=platform):
+                plan = SingTUNHostPlan({}, platform=platform)
+                worker = mock.Mock()
+                plan._worker = worker
+                worker.is_alive.return_value = False
+                plan._finishCleanupWork('cleanup', True)
+                worker.join.assert_called_once_with(0 if platform == 'Linux' else 6)
+                self.assertIsNone(plan._worker)
+
     def testLinuxResolverPrerequisiteFailsBeforeNativeActivation(self):
         module = importlib.import_module('Furious.Service.SingTUNHost')
         routingModule = importlib.import_module('Furious.Frozenlib.SingTUNRouting')
@@ -1702,7 +1831,7 @@ class SingTUNHostTest(unittest.TestCase):
         ) as helper:
             plan.prepare(['192.0.2.1'])
 
-        helper.assert_called_once_with()
+        helper.assert_called_once_with('1.1.1.1')
         self.assertTrue(plan._activated)
         self.assertFalse(plan.configuration['tun_options']['AutoRoute'])
         self.assertTrue(plan.configuration['tun_options']['EXP_ExternalConfiguration'])
@@ -1793,6 +1922,42 @@ class SingTUNStartupTest(unittest.TestCase):
     def setUpClass(cls):
         application()
 
+    def testLinuxHostObserverWaitsForAuthorizationWithoutADeadline(self):
+        """An open password request is still pending after a simulated day."""
+        module = importlib.import_module('Furious.Service.ConnectionManager')
+        with isolatedSettings(), mock.patch.object(
+            module, 'PLATFORM', 'Linux'
+        ), mock.patch.object(module.ConnectionStartOperation, 'start'), mock.patch(
+            'sys.excepthook'
+        ) as qtErrors:
+            manager = ConnectionManager()
+            operation = manager.startAsync(_Configuration(), 'Global', deepcopy=False)
+            operation._tun = mock.Mock()
+            plan = operation._tun._hostPlan
+            plan.workCompleted.return_value = False
+
+            try:
+                operation._waitSingHost('_startSingTUN')
+                probe = operation._conditionProbe
+                probe._elapsed = mock.Mock()
+                probe._elapsed.isValid.return_value = True
+                probe._elapsed.elapsed.return_value = 86400000
+                probe._poll()
+                self.assertFalse(operation._terminal)
+                self.assertTrue(probe._timer.isActive())
+
+                plan.workCompleted.return_value = True
+                with mock.patch.object(operation, '_resume') as resume:
+                    probe._poll()
+                    resume.assert_called_once_with('_startSingTUN')
+
+                self.assertIsNone(operation._conditionProbe)
+                self.assertFalse(probe._timer.isActive())
+            finally:
+                manager.cleanup()
+                processQtEvents()
+                qtErrors.assert_not_called()
+
     def testBothStartupPathsKeepCapturedTUNLogSourceAfterPreferenceChanges(self):
         """Changing the next engine cannot relabel a previously captured attempt."""
         module = importlib.import_module('Furious.Service.ConnectionManager')

tests/test_sing_tun_linux_host.py

@@ -76,17 +76,20 @@ class SingTUNLinuxHostTest(unittest.TestCase):
         scripts = []
         process = mock.Mock()
         process.wait.return_value = 0
-        process.poll.return_value = 0
+        process.poll.return_value = None
 
         def launch(arguments, **kwargs):
             self.assertEqual(arguments[:2], ['pkexec', 'bash'])
-            self.assertEqual(kwargs['stdin'], subprocess.DEVNULL)
+            self.assertEqual(kwargs['stdin'], subprocess.PIPE)
             self.assertTrue(
                 Path(arguments[2]).name.startswith(tempfile.gettempprefix())
             )
             self.assertEqual(Path(arguments[2]).suffix, '.sh')
 
             scripts.append(Path(arguments[2]).read_text(encoding='utf-8'))
+            Path(arguments[2]).parent.joinpath('ready').write_text(
+                '77', encoding='utf-8'
+            )
             return process
 
         link = {'ifname': 'utun101', 'ifindex': 77}
@@ -101,6 +104,8 @@ class SingTUNLinuxHostTest(unittest.TestCase):
             self.module, '_command', return_value=json.dumps([link])
         ):
             self.host.prepare()
+            with self.assertRaisesRegex(RuntimeError, 'already active'):
+                self.host.execute([])
             self.host.applyDNS('1.1.1.1')
 
         self.assertEqual(self.host.index, 77)
@@ -113,9 +118,22 @@ class SingTUNLinuxHostTest(unittest.TestCase):
         self.assertIn('trap ', scripts[0])
         self.assertNotIn('ifalias', ''.join(scripts))
         self.assertNotIn('furious-sing-tun-', ''.join(scripts))
-        self.assertIn('/sys/class/net/utun101/ifindex', scripts[1])
-        self.assertIn('resolvectl dns utun101 1.1.1.1', scripts[1])
+        self.assertEqual(len(scripts), 1)
+        self.assertIn('/sys/class/net/utun101/ifindex', scripts[0])
+        self.assertIn('resolvectl dns utun101 1.1.1.1', scripts[0])
         self.assertNotIn('socks5', ''.join(scripts))
+        self.assertIs(self.host._process, process)
+        self.assertIsNotNone(self.host._directory)
+
+        # Forwarding stops first; releasing this same helper needs no pkexec.
+        with mock.patch.object(
+            self.module, '_command', return_value='[]'
+        ), mock.patch.object(self.module.subprocess, 'Popen') as launch:
+            self.host.cleanup()
+            self.host.cleanup()
+
+        launch.assert_not_called()
+        process.stdin.close.assert_called_once()
         self.assertIsNone(self.host._process)
         self.assertIsNone(self.host._directory)
 
@@ -130,37 +148,151 @@ class SingTUNLinuxHostTest(unittest.TestCase):
                     checked.returncode, 0, checked.stderr.decode(errors='replace')
                 )
 
-    def testAuthorizationTimeoutRetainsExactProcessAndCancellationFile(self):
+    def testRecoveryWaitsForAuthorizationWithoutADeadline(self):
         process = mock.Mock()
-        process.wait.side_effect = subprocess.TimeoutExpired('pkexec', 60)
         process.poll.return_value = None
+        entered, authorized, complete = (
+            threading.Event(),
+            threading.Event(),
+            threading.Event(),
+        )
+        errors = []
+
+        def wait(timeout=None):
+            self.assertIsNone(timeout)
+            entered.set()
+            if not authorized.wait(5):
+                raise RuntimeError('fixture authorization was not released')
+            process.poll.return_value = 0
+            return 0
+
+        process.wait.side_effect = wait
+
+        def recover():
+            try:
+                self.host.execute([['ip', 'link', 'delete', 'dev', 'utun101']])
+            except Exception as ex:
+                # Any non-exit exceptions
+                errors.append(ex)
+            finally:
+                complete.set()
 
         with mock.patch.object(
             self.module.os, 'geteuid', return_value=1000, create=True
         ), mock.patch.object(self.module.subprocess, 'Popen', return_value=process):
-            with self.assertRaisesRegex(RuntimeError, 'timed out'):
-                self.host.execute(
-                    [['ip', 'link', 'set', 'dev', 'utun101', 'up']], preparing=True
-                )
+            worker = threading.Thread(target=recover)
+            worker.start()
+            try:
+                self.assertTrue(entered.wait(2))
+                self.assertFalse(complete.is_set())
+                self.assertIs(self.host._process, process)
+                self.assertIsNotNone(self.host._directory)
+                process.terminate.assert_not_called()
+                process.kill.assert_not_called()
+            finally:
+                authorized.set()
+                worker.join(5)
+
+        self.assertFalse(worker.is_alive())
+        self.assertEqual(errors, [])
+        process.wait.assert_called_once_with()
+        self.assertIsNone(self.host._process)
+        self.assertIsNone(self.host._directory)
+
+    def testDelayedAuthorizationIsUsedForSetupDnsAndDisconnect(self):
+        """A slow password entry does not create extra authorization requests."""
+        process = mock.Mock()
+        process.wait.return_value = 0
+        process.poll.return_value = None
+        clock = [0.0]
+        script = []
+
+        def launch(arguments, **kwargs):
+            script.append(Path(arguments[-1]))
+            return process
+
+        def advance(_seconds):
+            clock[0] += 3600
+            if clock[0] >= 86400:
+                script[0].parent.joinpath('ready').write_text('77', encoding='utf-8')
+
+        with mock.patch.object(
+            self.module.os, 'geteuid', return_value=1000, create=True
+        ), mock.patch.object(
+            self.module.os, 'getuid', return_value=1000, create=True
+        ), mock.patch.object(
+            self.module.subprocess, 'Popen', side_effect=launch
+        ) as launchHelper, mock.patch.object(
+            self.module.time, 'monotonic', side_effect=lambda: clock[0]
+        ), mock.patch.object(
+            self.module.time, 'sleep', side_effect=advance
+        ), mock.patch.object(
+            self.module,
+            '_command',
+            return_value=json.dumps([{'ifname': 'utun101', 'ifindex': 77}]),
+        ):
+            self.host.prepare('9.9.9.9')
+            self.host.applyDNS('9.9.9.9')
+            self.host.cancel()
+            self.host.drain()
+            process.stdin.close.assert_not_called()
+
+            with mock.patch.object(self.module, '_command', return_value='[]'):
+                self.host.cleanup()
+
+        self.assertEqual(clock[0], 86400)
+        launchHelper.assert_called_once()
+        process.stdin.close.assert_called_once()
+        self.assertFalse(self.host.dnsTouched)
+
+    def testPendingAuthorizationCancellationRetainsHelperUntilItExits(self):
+        process = mock.Mock()
+        process.poll.return_value = None
+        process.wait.side_effect = subprocess.TimeoutExpired('pkexec', 5)
+
+        def launch(arguments, **kwargs):
+            self.host.cancelled.set()
+            return process
+
+        with mock.patch.object(
+            self.module.os, 'geteuid', return_value=1000, create=True
+        ), mock.patch.object(self.module.subprocess, 'Popen', side_effect=launch):
+            with self.assertRaisesRegex(RuntimeError, 'preparation was cancelled'):
+                self.host.execute([], preparing=True, holding=True)
 
         directory = Path(self.host._directory.name)
-
-        self.assertIs(self.host._process, process)
-        self.assertEqual(len(list(directory.glob('*.sh'))), 1)
         self.assertTrue((directory / 'cancelled').exists())
+        process.stdin.close.assert_called_once()
 
         with self.assertRaisesRegex(RuntimeError, 'still running'):
             self.host.drain()
 
         self.assertIs(self.host._process, process)
-        process.terminate.assert_not_called()
-        process.kill.assert_not_called()
-
-        process.poll.return_value = 0
+        self.assertTrue(directory.exists())
+        process.poll.return_value = 126
         self.host.drain()
-
         self.assertFalse(directory.exists())
-        self.assertIsNone(self.host._process)
+
+    def testLiveForwarderDnsCannotBeReportedAsRestored(self):
+        """A refused child reap retains DNS and the authorized helper together."""
+        plan = SingTUNHostPlan({}, platform='Linux')
+        plan._linux = self.host
+        self.host._process = process = mock.Mock()
+        process.poll.return_value = None
+        self.host._directory = tempfile.TemporaryDirectory()
+        self.host._holding = self.host._ready = self.host.dnsTouched = True
+
+        with self.assertRaisesRegex(
+            RuntimeError, 'requires stopping the forwarding child'
+        ):
+            while not plan.finishDNSCleanup():
+                self.assertTrue(plan._done.wait(2))
+                plan._worker.join(2)
+
+        self.assertTrue(self.host.dnsTouched)
+        self.assertIs(self.host._process, process)
+        process.stdin.close.assert_not_called()
+        self.host._releaseHelper()
 
     def testScriptRetainsSuccessfulSetupAndRollsBackFailure(self):
         bash = _bash()
@@ -171,9 +303,9 @@ class SingTUNLinuxHostTest(unittest.TestCase):
         realPopen = subprocess.Popen
 
         failures = (
-            ('mtu', 'cancel', 'abandon', 'none')
+            ('mtu', 'rule', 'dns', 'cancel', 'abandon', 'none', 'eof')
             if os.name != 'nt'
-            else ('mtu', 'cancel', 'none')
+            else ('mtu', 'rule', 'dns', 'cancel', 'none', 'eof')
         )
 
         for failure in failures:
@@ -209,7 +341,16 @@ class SingTUNLinuxHostTest(unittest.TestCase):
                         '"link set dev")\n'
                         '  if [ "$FAKE_FAILURE" = mtu ] && [ "$5" = mtu ]; then exit 1; fi\n'
                         '  ;;\n'
-                        '"-4 addr add"|"-6 addr add"|"-4 route add"|"-6 route add"|"-4 rule add"|"-6 rule add") ;;\n'
+                        '"-4 rule add"|"-6 rule add")\n'
+                        '  count=$(cat "$FAKE_NET/rules" 2>/dev/null || echo 0)\n'
+                        '  if [ "$FAKE_FAILURE" = rule ] && [ "$count" = 2 ]; then exit 1; fi\n'
+                        '  echo $((count + 1)) > "$FAKE_NET/rules"\n'
+                        '  ;;\n'
+                        '"-4 rule del"|"-6 rule del")\n'
+                        '  count=$(cat "$FAKE_NET/rules")\n'
+                        '  echo $((count - 1)) > "$FAKE_NET/rules"\n'
+                        '  ;;\n'
+                        '"-4 addr add"|"-6 addr add"|"-4 route add"|"-6 route add") ;;\n'
                         '"link delete dev")\n'
                         '  rm "$device/ifindex"\n'
                         '  rmdir "$device"\n'
@@ -219,6 +360,14 @@ class SingTUNLinuxHostTest(unittest.TestCase):
                     )
 
                 ip.chmod(0o755)
+                resolver = binDirectory / 'resolvectl'
+                resolver.write_text(
+                    '#!/usr/bin/env bash\n'
+                    'echo "$1" >> "$FAKE_NET/dns"\n'
+                    'if [ "$FAKE_FAILURE" = dns ] && [ "$1" = domain ]; then exit 1; fi\n',
+                    encoding='utf-8',
+                )
+                resolver.chmod(0o755)
 
                 def launch(arguments, **kwargs):
                     self.assertEqual(arguments[:2], ['pkexec', 'bash'])
@@ -257,21 +406,40 @@ class SingTUNLinuxHostTest(unittest.TestCase):
                 ), mock.patch.object(
                     self.module,
                     '_command',
-                    return_value=json.dumps([{'ifname': 'utun101', 'ifindex': 77}]),
+                    side_effect=lambda args: json.dumps(
+                        [{'ifname': 'utun101', 'ifindex': 77}]
+                        if args == ['ip', '-N', '-j', 'link', 'show']
+                        and (net / 'utun101').exists()
+                        else []
+                    ),
                 ):
-                    if failure == 'none':
+                    if failure in ('none', 'eof'):
                         self.host.prepare()
 
                         self.assertEqual(self.host.index, 77)
+                        self.assertTrue((net / 'utun101').exists())
+                        self.assertIsNone(self.host._process.poll())
+                        if failure == 'eof':
+                            # Closing the parent pipe also covers an application
+                            # exit without calling the normal cleanup method.
+                            self.host._releaseHelper()
+                            self.assertEqual(self.host._process.wait(timeout=5), 0)
+                        self.host.cleanup()
                     else:
                         with self.assertRaisesRegex(
                             RuntimeError, 'host operation failed'
                         ):
                             self.host.prepare()
 
-                self.assertEqual((net / 'utun101').exists(), failure == 'none')
+                self.assertFalse((net / 'utun101').exists())
                 self.assertIsNone(self.host._process)
                 self.assertIsNone(self.host._directory)
+                if (net / 'rules').exists():
+                    self.assertEqual((net / 'rules').read_text().strip(), '0')
+                if (net / 'dns').exists():
+                    self.assertEqual(
+                        (net / 'dns').read_text().splitlines()[-1], 'revert'
+                    )
 
     def testCancellationBeforeAuthorizationCannotAcquireDevice(self):
         scripts = []
@@ -461,6 +629,28 @@ class SingTUNLinuxHostTest(unittest.TestCase):
                 'utun101',
             )
 
+    def testKernelSuppressionFieldMatchesExactRule(self):
+        """Use iproute2's JSON spelling, which differs from its CLI spelling."""
+        expected = {_linuxRuleSelectors(command) for command in self.host.rules}
+        rule = {
+            'priority': 2001,
+            'src': 'all',
+            'table': 123456,
+            'suppress_prefixlen': 0,
+            'protocol': 0,
+        }
+        command = _linuxRuleDelete('-4', rule, 123456, 2000, 'utun101')
+        self.assertIn(_linuxRuleSelectors(command), expected)
+        self.assertIn('suppress_prefixlength', command)
+
+        for value in (1, -1, '0'):
+            with self.subTest(value=value), self.assertRaisesRegex(
+                RuntimeError, 'Conflicting'
+            ):
+                _linuxRuleDelete(
+                    '-4', {**rule, 'suppress_prefixlen': value}, 123456, 2000, 'utun101'
+                )
+
     def testCleanupUsesExactSelectorsAndRetainsRefusedCleanupForRetry(self):
         plan = SingTUNHostPlan({}, platform='Linux')
         plan._linux = self.host
@@ -468,7 +658,13 @@ class SingTUNLinuxHostTest(unittest.TestCase):
         link = {'ifname': 'utun101', 'ifindex': 77}
         self.host.index = 77
         present = True
-        rule = {'priority': 2000, 'src': 'all', 'dst': '198.18.0.0/15', 'table': 123456}
+        rule = {
+            'priority': 2001,
+            'src': 'all',
+            'table': 123456,
+            'suppress_prefixlen': 0,
+            'protocol': 0,
+        }
 
         def command(arguments):
             if arguments == ['ip', '-N', '-j', 'link', 'show']:
@@ -492,8 +688,8 @@ class SingTUNLinuxHostTest(unittest.TestCase):
             nonlocal present
             self.assertTrue(kwargs['guard'])
             self.assertEqual(commands[-1], ['ip', 'link', 'delete', 'dev', 'utun101'])
-            self.assertIn('to', commands[0])
-            self.assertIn('198.18.0.0/15', commands[0])
+            self.assertIn('suppress_prefixlength', commands[0])
+            self.assertIn('123456', commands[0])
             self.assertNotIn('flush', str(commands))
             present = False