pybridge: Let cockpit-beiboot use package-installed askpass helper

We need and want ensure_ferny_askpass() *only* for the flatpak case. Add
some comments about that, as we already forgot it once.

When running it from distro packages or the cockpit/ws container, we
always want to use the packaged cockpit-askpass helper. Re-use
`patch_libexecdir()` for that, as it already has all the necessary
magic. Turn that into a proper global function.

This fixes running cockpit-beiboot as cockpit-wsinstance system user, as
there is no existing/writable home directory, and trying to mkdir
~/.cache EPERMs.

src/cockpit/beiboot.py

@@ -34,7 +34,7 @@
 from cockpit.channel import ChannelRoutingRule
 from cockpit.channels import PackagesChannel
 from cockpit.jsonutil import JsonObject
-from cockpit.packages import Packages, PackagesLoader
+from cockpit.packages import Packages, PackagesLoader, patch_libexecdir
 from cockpit.peer import Peer
 from cockpit.protocol import CockpitProblem
 from cockpit.router import Router, RoutingRule
@@ -44,9 +44,15 @@
 
 
 def ensure_ferny_askpass() -> Path:
+    """Create askpass executable
+
+    We need this for the flatpak: ssh and thus the askpass program run on the host (via flatpak-spawn),
+    not the flatpak. Thus we cannot use the shipped cockpit-askpass program.
+    """
     src_path = importlib.resources.files(ferny.__name__) / 'interaction_client.py'
     src_data = src_path.read_bytes()
 
+    # Create the file in $XDG_CACHE_HOME, one of the few locations that a flatpak can write to
     xdg_cache_home = os.environ.get('XDG_CACHE_HOME')
     if xdg_cache_home is None:
         xdg_cache_home = os.path.expanduser('~/.cache')
@@ -185,9 +191,21 @@ async def do_connect_transport(self) -> None:
         cmd: Sequence[str] = ('python3', '-ic', '# cockpit-bridge')
         env: Sequence[str] = ()
 
+        in_flatpak = os.path.exists('/.flatpak-info')
+
         # Remote host?  Wrap command with SSH
         if self.destination != 'localhost':
-            ssh_askpass = ensure_ferny_askpass()
+            if in_flatpak:
+                # we run ssh and thus the helper on the host, always use the xdg-cache helper
+                ssh_askpass = ensure_ferny_askpass()
+            else:
+                # outside of the flatpak we expect cockpit-ws and thus an installed helper
+                askpass = patch_libexecdir('${libexecdir}/cockpit-askpass')
+                assert isinstance(askpass, str)
+                ssh_askpass = Path(askpass)
+                if not ssh_askpass.exists():
+                    logger.error("Could not find cockpit-askpass helper at %r", askpass)
+
             env = (
                 f'SSH_ASKPASS={ssh_askpass!s}',
                 'DISPLAY=x',
@@ -203,7 +221,7 @@ async def do_connect_transport(self) -> None:
             cmd = ('ssh', *host_args, shlex.join(cmd))
 
         # Running in flatpak?  Wrap command with flatpak-spawn --host
-        if os.path.exists('/.flatpak-info'):
+        if in_flatpak:
             cmd = ('flatpak-spawn', '--host',
                    *(f'--env={kv}' for kv in env),
                    *cmd)

src/cockpit/packages.py

@@ -92,6 +92,43 @@ def sortify_version(version: str) -> str:
     return '.'.join(part.zfill(8) for part in version.split('.'))
 
 
+@functools.lru_cache()
+def get_libexecdir() -> str:
+    """Detect libexecdir on current machine
+
+    This only works for systems which have cockpit-ws installed.
+    """
+    for candidate in ['/usr/local/libexec', '/usr/libexec', '/usr/local/lib/cockpit', '/usr/lib/cockpit']:
+        if os.path.exists(os.path.join(candidate, 'cockpit-askpass')):
+            return candidate
+    else:
+        logger.warning('Could not detect libexecdir')
+        # give readable error messages
+        return '/nonexistent/libexec'
+
+
+# HACK: Type narrowing over Union types is not supported in the general case,
+# but this works for the case we care about: knowing that when we pass in an
+# JsonObject, we'll get an JsonObject back.
+J = TypeVar('J', JsonObject, JsonDocument)
+
+
+def patch_libexecdir(obj: J) -> J:
+    if isinstance(obj, str):
+        if '${libexecdir}/cockpit-askpass' in obj:
+            # extra-special case: we handle this internally
+            abs_askpass = shutil.which('cockpit-askpass')
+            if abs_askpass is not None:
+                return obj.replace('${libexecdir}/cockpit-askpass', abs_askpass)
+        return obj.replace('${libexecdir}', get_libexecdir())
+    elif isinstance(obj, dict):
+        return {key: patch_libexecdir(value) for key, value in obj.items()}
+    elif isinstance(obj, list):
+        return [patch_libexecdir(item) for item in obj]
+    else:
+        return obj
+
+
 # A document is a binary stream with a Content-Type, optional Content-Encoding,
 # and optional Content-Security-Policy
 class Document(NamedTuple):
@@ -281,42 +318,6 @@ class PackagesLoader:
         'path-not-exists': lambda p: not os.path.exists(p),
     }
 
-    @staticmethod
-    @functools.lru_cache()
-    def get_libexecdir() -> str:
-        """Detect libexecdir on current machine
-
-        This only works for systems which have cockpit-ws installed.
-        """
-        for candidate in ['/usr/local/libexec', '/usr/libexec', '/usr/local/lib/cockpit', '/usr/lib/cockpit']:
-            if os.path.exists(os.path.join(candidate, 'cockpit-askpass')):
-                return candidate
-        else:
-            logger.warning('Could not detect libexecdir')
-            # give readable error messages
-            return '/nonexistent/libexec'
-
-    # HACK: Type narrowing over Union types is not supported in the general case,
-    # but this works for the case we care about: knowing that when we pass in an
-    # JsonObject, we'll get an JsonObject back.
-    J = TypeVar('J', JsonObject, JsonDocument)
-
-    @classmethod
-    def patch_libexecdir(cls, obj: J) -> J:
-        if isinstance(obj, str):
-            if '${libexecdir}/cockpit-askpass' in obj:
-                # extra-special case: we handle this internally
-                abs_askpass = shutil.which('cockpit-askpass')
-                if abs_askpass is not None:
-                    return obj.replace('${libexecdir}/cockpit-askpass', abs_askpass)
-            return obj.replace('${libexecdir}', cls.get_libexecdir())
-        elif isinstance(obj, dict):
-            return {key: cls.patch_libexecdir(value) for key, value in obj.items()}
-        elif isinstance(obj, list):
-            return [cls.patch_libexecdir(item) for item in obj]
-        else:
-            return obj
-
     @classmethod
     def get_xdg_data_dirs(cls) -> Iterable[str]:
         try:
@@ -368,7 +369,7 @@ def patch_manifest(cls, manifest: JsonObject, parent: Path) -> JsonObject:
 
             manifest = cls.merge_patch(manifest, override)
 
-        return cls.patch_libexecdir(manifest)
+        return patch_libexecdir(manifest)
 
     @classmethod
     def load_manifests(cls) -> Iterable[Manifest]: