cockpit.superuser: prevent 'any' from being set

ae5ce0a introduced a bug in the code that handled the case where
the `any` superuser bridge was requested: previously, the code would set
`self.current` to the name of the selected peer, but after the change,
it would get set to `any` instead.  Fix that up by ignoring the value of
`name` after we use it for finding the peer.

This allows us to simplify the condition for checking if the running
bridge ought to be shut down when reloading packages.  The current logic
reads that we should shutdown the current bridge "if there's any bridge
which has our name" (which is very likely, in case nothing changed).  We
need to change that to `not any`, but we can also simplify things by
storing the original config object on the peer and checking if it
appears in the new list (in its entirety).

Thanks to Jelle for the testcase.

Fixes #19327

Co-authored-by: Jelle van der Waa <[email protected]>

src/cockpit/peer.py

@@ -219,10 +219,12 @@ def do_kill(self, host: Optional[str], group: Optional[str]) -> None:
 
 
 class ConfiguredPeer(Peer):
+    config: BridgeConfig
     args: Sequence[str]
     env: Sequence[str]
 
     def __init__(self, router: Router, config: BridgeConfig):
+        self.config = config
         self.args = config.spawn
         self.env = config.environ
         super().__init__(router)

src/cockpit/superuser.py

@@ -40,12 +40,10 @@
 
 
 class SuperuserPeer(ConfiguredPeer):
-    name: str
     responder: ferny.InteractionResponder
 
-    def __init__(self, router: Router, name: str, config: BridgeConfig, responder: ferny.InteractionResponder):
+    def __init__(self, router: Router, config: BridgeConfig, responder: ferny.InteractionResponder):
         super().__init__(router, config)
-        self.name = name
         self.responder = responder
 
     async def do_connect_transport(self) -> None:
@@ -169,7 +167,7 @@ async def go(self, name: str, responder: ferny.InteractionResponder) -> None:
             raise bus.BusError('cockpit.Superuser.Error', f'Unknown superuser bridge type "{name}"')
 
         self.current = 'init'
-        self.peer = SuperuserPeer(self.router, name, config, responder)
+        self.peer = SuperuserPeer(self.router, config, responder)
         self.peer.add_done_callback(self.peer_done)
 
         try:
@@ -179,7 +177,7 @@ async def go(self, name: str, responder: ferny.InteractionResponder) -> None:
         except (OSError, PeerError) as exc:
             raise bus.BusError('cockpit.Superuser.Error', str(exc)) from exc
 
-        self.current = name
+        self.current = self.peer.config.name
 
     def set_configs(self, configs: Sequence[BridgeConfig]):
         logger.debug("set_configs() with %d items", len(configs))
@@ -190,18 +188,10 @@ def set_configs(self, configs: Sequence[BridgeConfig]):
 
         logger.debug("  bridges are now %s", self.bridges)
 
-        # If the currently active bridge got removed, stop it
+        # If the currently active bridge config is not in the new set of configs, stop it
         if self.peer is not None:
-            if self.current == 'any':
-                removed = len(self.superuser_configs) == 0
-            else:
-                removed = any(c.name == self.peer.name for c in self.superuser_configs)
-
-            if removed:
-                logger.debug(
-                    "  stopping current superuser bridge '%s' (peer name %s), as it disappeared from configs",
-                    self.current, self.peer.name)
-
+            if self.peer.config not in self.superuser_configs:
+                logger.debug("  stopping superuser bridge '%s': it disappeared from configs", self.peer.config.name)
                 self.stop()
 
     def cancel_prompt(self):

test/verify/check-superuser

@@ -56,6 +56,14 @@ class TestSuperuser(testlib.MachineCase):
         b.leave_page()
         b.check_superuser_indicator("Limited access")
 
+        # A reload should not lose privileges
+        b.become_superuser()
+        b.reload()
+        b.check_superuser_indicator("Administrative access")
+
+        # Drop privileges
+        b.drop_superuser()
+
         # We want to be lectured again
         self.restore_file("/var/db/sudo/lectured/admin")
         self.restore_file("/var/lib/sudo/lectured/admin")