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]: