import argparse
import asyncio
import base64
import importlib.resources
import logging
import os
import re
import shlex
import tempfile
import time
from pathlib import Path
from typing import Dict, Literal, Optional, Sequence
from cockpit import polyfills
from cockpit._vendor import ferny
from cockpit._vendor.bei import bootloader
from cockpit.beipack import BridgeBeibootHelper
from cockpit.bridge import parse_os_release, setup_logging
from cockpit.channel import ChannelRoutingRule
from cockpit.channels.packages import PackagesChannel
from cockpit.jsonutil import JsonObject, get_str
from cockpit.osinfo import supported_oses
from cockpit.packages import Packages, PackagesLoader, patch_libexecdir
from cockpit.peer import Peer
from cockpit.protocol import CockpitProblem, CockpitProtocolError
from cockpit.router import Router, RoutingRule
from cockpit.transports import StdioTransport
logger = logging.getLogger('cockpit.beiboot')
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()
xdg_cache_home = os.environ.get('XDG_CACHE_HOME')
if xdg_cache_home is None:
xdg_cache_home = os.path.expanduser('~/.cache')
os.makedirs(xdg_cache_home, exist_ok=True)
dest_path = Path(xdg_cache_home, 'cockpit-client-askpass')
logger.debug("Checking if %s exists...", dest_path)
try:
if dest_path.read_bytes() != src_data:
logger.debug(" ... it exists but is not the same version...")
raise ValueError
if not dest_path.stat().st_mode & 0o100:
logger.debug(" ... it has the correct contents, but is not executable...")
raise ValueError
except (FileNotFoundError, ValueError):
logger.debug(" ... writing contents.")
dest_path.write_bytes(src_data)
dest_path.chmod(0o700)
return dest_path
class ProxyPackagesLoader(PackagesLoader):
file_status: Dict[str, bool]
def __init__(self, file_status: Dict[str, bool]):
self.file_status = file_status
def path_exists(self, path: str) -> bool:
return self.file_status.get(path, False)
BEIBOOT_GADGETS = {
"report_exists": r"""
import os
def report_exists(files):
command('cockpit.report-exists', {name: os.path.exists(name) for name in files})
""",
"check_os_release": r"""
import os
def check_os_release(_argv):
try:
with open('/etc/os-release') as f:
command('cockpit.check-os-release', f.read())
except OSError:
command('cockpit.check-os-release', "")
""",
"force_exec": r"""
import os
def force_exec(argv):
try:
os.execvp(argv[0], argv)
except OSError as e:
command('cockpit.fail-no-cockpit', str(e))
""",
**ferny.BEIBOOT_GADGETS
}
class DefaultRoutingRule(RoutingRule):
peer: 'Peer | None'
def __init__(self, router: Router):
super().__init__(router)
def apply_rule(self, options: JsonObject) -> 'Peer | None':
return self.peer
def shutdown(self) -> None:
if self.peer is not None:
self.peer.close()
class AuthorizeResponder(ferny.AskpassHandler):
commands = ('ferny.askpass', 'cockpit.report-exists', 'cockpit.fail-no-cockpit', 'cockpit.check-os-release')
router: Router
def __init__(self, router: Router, basic_password: Optional[str]):
self.router = router
self.basic_password = basic_password
self.have_basic_password = basic_password is not None
async def do_askpass(self, messages: str, prompt: str, hint: str) -> Optional[str]:
logger.debug("AuthorizeResponder: prompt %r, messages %r, hint %r", prompt, messages, hint)
if self.have_basic_password and 'password:' in prompt.lower():
if self.basic_password is not None:
logger.debug("AuthorizeResponder: sending Basic auth password for prompt %r", prompt)
reply = self.basic_password
self.basic_password = None
return reply
if hint == 'none':
return None
fp_match = re.search(r'\n(\w+) key fingerprint is:? ([^.\n]+)', prompt)
host_match = re.search(r"authenticity of host '([^ ]+) ", prompt)
args = {}
if fp_match and host_match:
hostname = host_match.group(1)
if hostname == '127.0.0.1':
logger.debug('auto-accepting fingerprint for 127.0.0.1: %s', host_match)
return 'yes'
args['host-key'] = f'{hostname} {fp_match.group(1)} login-data'
args['default'] = fp_match.group(2)
challenge_id = f'{os.getpid()}-{time.time()}'
challenge_prefix = f'X-Conversation {challenge_id}'
challenge = challenge_prefix + ' ' + base64.b64encode(prompt.encode()).decode()
response = await self.router.request_authorization(challenge,
timeout=None,
messages=messages,
prompt=prompt,
hint=hint,
echo=False,
**args)
if not response.startswith(challenge_prefix):
raise CockpitProtocolError(
f"AuthorizeResponder: response {response} does not match challenge {challenge_prefix}")
b64 = response.removeprefix(challenge_prefix).strip()
response = base64.b64decode(b64.encode()).decode()
logger.debug('Returning a %d chars response', len(response))
return response
async def do_custom_command(self, command: str, args: tuple, fds: list[int], stderr: str) -> None:
logger.debug('Got ferny command %s %s %s', command, args, stderr)
if command == 'cockpit.report-exists':
file_status, = args
self.router.packages = Packages(loader=ProxyPackagesLoader(file_status))
self.router.routing_rules.insert(0, ChannelRoutingRule(self.router, [PackagesChannel]))
if command == 'cockpit.fail-no-cockpit':
raise CockpitProblem('no-cockpit', message=args[0])
if command == 'cockpit.check-os-release':
remote_os = parse_os_release(args[0])
logger.debug("cockpit.check-os-release: remote OS: %r", remote_os)
logger.debug("cockpit.check-os-release: supported OSes: %r", supported_oses)
for osinfo in supported_oses:
if all(remote_os.get(k) == v for k, v in osinfo.items()):
logger.debug("cockpit.check-os-release: remote matches supported OS %r", osinfo)
return
logger.debug("cockpit.check-os-release: remote: %r", remote_os)
try:
with open("/etc/os-release") as f:
this_os = parse_os_release(f.read())
except OSError as e:
logger.warning("failed to read local /etc/os-release, skipping OS compatibility check: %s", e)
return
if remote_os.get('ID') == this_os.get('ID') and remote_os.get('VERSION_ID') == this_os.get('VERSION_ID'):
logger.debug("cockpit.check-os-release: remote OS matches local OS %r", this_os)
return
unsupported = f'{remote_os.get("NAME", remote_os.get("ID", "?"))} {remote_os.get("VERSION_ID", "")}'
raise CockpitProblem('no-cockpit', unsupported=unsupported)
def python_interpreter(comment: str) -> tuple[Sequence[str], Sequence[str]]:
return ('python3', '-ic', f'# {comment}'), ()
def via_ssh(cmd: Sequence[str], dest: str, ssh_askpass: Path, *ssh_opts: str) -> tuple[Sequence[str], Sequence[str]]:
host, _, port = dest.rpartition(':')
if port.isdigit() and not host.endswith(':'):
if host.startswith('[') and host.endswith(']'):
host = host[1:-1]
destination = ['-p', port, '--', host]
else:
destination = ['--', dest]
return (
'ssh', *ssh_opts, *destination, shlex.join(cmd)
), (
f'SSH_ASKPASS={ssh_askpass!s}',
'DISPLAY=x',
'SSH_ASKPASS_REQUIRE=force',
)
def flatpak_spawn(cmd: Sequence[str], env: Sequence[str]) -> tuple[Sequence[str], Sequence[str]]:
return (
'flatpak-spawn', '--host',
*(f'--env={kv}' for kv in env),
*cmd
), (
)
class SshPeer(Peer):
mode: 'Literal["always"] | Literal["never"] | Literal["supported"] | Literal["auto"]'
def __init__(self, router: Router, destination: str, args: argparse.Namespace):
self.destination = destination
self.remote_bridge = args.remote_bridge
self.tmpdir = tempfile.TemporaryDirectory()
self.known_hosts_file = Path(self.tmpdir.name) / 'user-known-hosts'
self.basic_password: 'str | None' = None
super().__init__(router)
async def do_connect_transport(self) -> None:
if os.path.exists('/.flatpak-info'):
await self.connect_from_flatpak()
else:
await self.connect_from_bastion_host()
async def connect_from_flatpak(self) -> None:
cmd, env = python_interpreter('cockpit-bridge')
if self.destination != 'localhost':
cmd, env = via_ssh(cmd, self.destination, ensure_ferny_askpass())
cmd, env = flatpak_spawn(cmd, env)
await self.boot(cmd, env)
async def connect_from_bastion_host(self) -> None:
known_hosts = None
args = ['-o', 'NumberOfPasswordPrompts=1']
auth = await self.router.request_authorization_object("*")
response = get_str(auth, 'response')
if response.startswith('Basic '):
decoded = base64.b64decode(response[6:]).decode()
user_password, _, known_hosts = decoded.partition('\0')
user, _, self.basic_password = user_password.partition(':')
if user:
logger.debug("got username %s and password from Basic auth", user)
args += ['-l', user]
if self.basic_password is None:
args += ['-o', 'PasswordAuthentication=no']
cmd, env = python_interpreter('cockpit-bridge')
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_known_hosts = os.getenv('COCKPIT_SSH_KNOWN_HOSTS_FILE')
if env_known_hosts is not None:
args += ['-o', f'GlobalKnownHostsFile={env_known_hosts}']
if known_hosts is not None:
self.known_hosts_file.write_text(known_hosts)
args += ['-o', f'UserKnownHostsfile={self.known_hosts_file!s}']
cmd, env = via_ssh(cmd, self.destination, ssh_askpass, *args)
await self.boot(cmd, env)
async def boot(self, cmd: Sequence[str], env: Sequence[str]) -> None:
beiboot_helper = BridgeBeibootHelper(self)
agent = ferny.InteractionAgent([AuthorizeResponder(self.router, self.basic_password), beiboot_helper])
logger.debug("Launching command: cmd=%s env=%s", cmd, env)
transport = await self.spawn(cmd, env, stderr=agent, start_new_session=True)
if self.remote_bridge == 'auto':
exec_cockpit_bridge_steps = [('try_exec', (['cockpit-bridge'],))]
elif self.remote_bridge == 'always':
exec_cockpit_bridge_steps = [('force_exec', (['cockpit-bridge'],))]
elif self.remote_bridge == 'supported':
exec_cockpit_bridge_steps = [('try_exec', (['cockpit-bridge'],)), ('check_os_release', ([],))]
else:
assert self.remote_bridge == 'never'
exec_cockpit_bridge_steps = []
stage1 = bootloader.make_bootloader([
*exec_cockpit_bridge_steps,
('report_exists', [list(PackagesLoader.get_condition_files())]),
*beiboot_helper.steps,
], gadgets=BEIBOOT_GADGETS)
transport.write(stage1.encode())
await agent.communicate()
def do_superuser_init_done(self) -> None:
self.basic_password = None
def do_authorize(self, message: JsonObject) -> None:
logger.debug("SshPeer.do_authorize: %r; have password %s", message, self.basic_password is not None)
if get_str(message, 'challenge').startswith('plain1:'):
cookie = get_str(message, 'cookie')
if self.basic_password is not None:
logger.debug("SshPeer.do_authorize: responded with password")
self.write_control(command='authorize', cookie=cookie, response=self.basic_password)
self.basic_password = None
return
logger.debug("SshPeer.do_authorize: authentication-unavailable")
self.write_control(command='authorize', cookie=cookie, problem='authentication-unavailable')
class SshBridge(Router):
packages: Optional[Packages] = None
ssh_peer: SshPeer
def __init__(self, args: argparse.Namespace):
rule = DefaultRoutingRule(self)
super().__init__([rule])
self.ssh_peer = SshPeer(self, args.destination, args)
rule.peer = self.ssh_peer
def do_send_init(self) -> None:
pass
def do_init(self, message):
logger.debug("SshBridge.do_init: %r", message)
self.ssh_peer.write_control(message)
def setup_session(self) -> None:
self.ssh_peer.add_done_callback(self.close)
async def run(args) -> None:
logger.debug("Hi. How are you today?")
bridge = SshBridge(args)
StdioTransport(asyncio.get_running_loop(), bridge)
try:
message = dict(await bridge.ssh_peer.start())
if bridge.ssh_peer.known_hosts_file.exists():
bridge.write_control(
command='authorize', challenge='x-login-data', cookie='-', login_data={
'known-hosts': bridge.ssh_peer.known_hosts_file.read_text()
}
)
capabilities = message.setdefault('capabilities', {})
if not isinstance(capabilities, dict):
bridge.write_control(command='init', problem='protocol-error', message='capabilities must be a dict')
return
assert isinstance(capabilities, dict)
capabilities['explicit-superuser'] = True
if bridge.packages:
message['packages'] = dict.fromkeys(bridge.packages.packages)
bridge.write_control(message)
bridge.ssh_peer.thaw_endpoint()
except ferny.InteractionError as exc:
error = ferny.ssh_errors.get_exception_for_ssh_stderr(str(exc))
logger.debug("ferny.InteractionError: %s, interpreted as: %r", exc, error)
if isinstance(error, ferny.SshAuthenticationError):
problem = 'authentication-failed'
elif isinstance(error, ferny.SshChangedHostKeyError):
problem = 'invalid-hostkey'
elif isinstance(error, ferny.SshHostKeyError):
problem = 'unknown-hostkey'
elif isinstance(error, OSError):
problem = 'unknown-host'
elif isinstance(error, ferny.SshError):
problem = 'authentication-failed'
else:
problem = 'internal-error'
if bridge.ssh_peer.known_hosts_file.exists():
bridge.write_control(command='init', problem=problem, message=str(error),
known_hosts=bridge.ssh_peer.known_hosts_file.read_text())
else:
bridge.write_control(command='init', problem=problem, message=str(error))
return
except CockpitProblem as exc:
logger.debug("CockpitProblem: %s", exc)
bridge.write_control(exc.attrs, command='init')
return
except asyncio.CancelledError:
logger.debug("Peer bridge got cancelled, exiting")
return
logger.debug('Startup done. Looping until connection closes.')
bridge.setup_session()
try:
await bridge.communicate()
except BrokenPipeError:
pass
def main() -> None:
polyfills.install()
parser = argparse.ArgumentParser(description='cockpit-bridge is run automatically inside of a Cockpit session.')
parser.add_argument('--remote-bridge', choices=['auto', 'never', 'supported', 'always'], default='auto',
help="How to run cockpit-bridge from the remote host: auto: if installed (default), "
"never: always copy the local one; "
"supported: if not installed, copy local one for compatible OSes, fail otherwise; "
"always: fail if not installed")
parser.add_argument('--debug', action='store_true')
parser.add_argument('destination', help="Name of the remote host to connect to, or 'localhost'")
args = parser.parse_args()
setup_logging(debug=args.debug)
asyncio.run(run(args), debug=args.debug)
if __name__ == '__main__':
main()
