🏡 index : quine-global-labs/mirror-cockpit-project-cockpit.git

#!/usr/bin/env python3

#
# Copyright (C) 2018-2021 Red Hat, Inc.
# SPDX-License-Identifier: LGPL-2.1-or-later


import configparser
import ctypes
import logging
import os
import signal
import subprocess
import sys
import tempfile
import textwrap

import gi

gi.require_version("Gtk", "4.0")
from gi.repository import GLib, Gio, Gtk  # ruff: ignore[unsorted-imports, module-import-not-at-top-of-file]

gi.require_version("WebKit", "6.0")
from gi.repository import WebKit  # ruff: ignore[module-import-not-at-top-of-file]

gi.require_version("Adw", "1")
from gi.repository import Adw  # ruff: ignore[module-import-not-at-top-of-file]

libexecdir = os.path.realpath(__file__ + '/..')

sys_prctl = ctypes.CDLL(None).prctl


def prctl(*args):
    if sys_prctl(*args) != 0:
        raise Exception('prctl() failed')


def get_user_state_dir():
    try:
        # GLib ≥ 2.72
        return GLib.get_user_state_dir()
    except AttributeError:
        try:
            return os.environ["XDG_STATE_HOME"]
        except KeyError:
            return os.path.expanduser("~/.local/share")


SET_PDEATHSIG = 1


@Gtk.Template(filename=f'{libexecdir}/cockpit-client.ui')
class CockpitClientWindow(Gtk.ApplicationWindow):
    __gtype_name__ = 'CockpitClientWindow'
    webview_container = Gtk.Template.Child()
    title = Gtk.Template.Child()

    def __init__(self, app, uri):
        super().__init__(application=app)

        self.webview = WebKit.WebView(network_session=app.network_session, vexpand=True, hexpand=True)
        self.webview_container.append(self.webview)

        self.add_action_entries([
            ('reload', self.reload),
            ('reload-bypass-cache', self.reload_bypass_cache),
            ('go-back', self.go_back),
            ('go-forward', self.go_forward),
            ('zoom', self.zoom, 's'),
            ('open-inspector', self.open_inspector),
            ('run-js', self.run_js, 's', '""'),
        ])
        self.lookup_action('run-js').set_enabled(app.enable_run_js)

        self.webview.get_settings().set_enable_developer_extras(enabled=True)
        self.webview.connect('decide-policy', self.decide_policy)
        self.webview.load_uri(uri)

        history = self.webview.get_back_forward_list()
        history.connect('changed', self.history_changed)
        self.history_changed()

        self.webview.connect('notify::title', self.update_subtitle)

        if app.no_ui:
            self.set_titlebar(None)
            self.webview.bind_property('title', self, 'title')

    def history_changed(self, *args):
        self.lookup_action('go-back').set_enabled(self.webview.can_go_back())
        self.lookup_action('go-forward').set_enabled(self.webview.can_go_forward())

    def update_subtitle(self, webview, *args):
        title = webview.get_title() or "Loading..."
        self.title.set_label(title)

    def reload(self, *args):
        self.webview.reload()

    def reload_bypass_cache(self, *args):
        self.webview.reload_bypass_cache()

    def go_back(self, *args):
        self.webview.go_back()

    def go_forward(self, *args):
        self.webview.go_forward()

    def decide_policy(self, _view, decision, decision_type):
        if decision_type == WebKit.PolicyDecisionType.NEW_WINDOW_ACTION:
            uri = decision.get_navigation_action().get_request().get_uri()
            if uri.startswith('http://127'):
                logging.error('warning: no support for pop-ups')
            else:
                launcher = Gtk.UriLauncher.new(uri)
                launcher.launch(self, None, None, None)

            decision.ignore()
            return True

        return False

    def zoom(self, _action, parameter, *_unused):
        current = self.webview.get_zoom_level()
        factors = {'in': current * 1.1, 'default': 1.0, 'out': current * 0.9}
        self.webview.set_zoom_level(factors[parameter.get_string()])

    def open_inspector(self, *_unused):
        self.webview.get_inspector().show()

    def run_js(self, action, parameter, *_unused):
        """Run given JavaScript code in the current page/context

        The JS code has to return a scalar value immediately. To process asynchronous
        actions, this function installs signal handlers to wait until the code calls

            window.webkit.messageHandlers.result.postMessage("some string")

        or a page load happens.

        This is because a page load (as reaction to clicking a link or other action)
        impredictably and immediately stops the current context and execution of the
        given JS code, with no reliable way of returning a result value.

        The postMessage() string is returned via the action state.
        If a page load happens instead, the action state will be "page-load".
        If the JS code throws an error, the action state gets set to the error message.
        """
        def set_result(result):
            action.set_state(GLib.Variant.new_string(result))
            print(f"run-js async result: {result}")
            self.webview.disconnect(on_load_handler)
            uc_manager.disconnect(on_result_handler)
            uc_manager.unregister_script_message_handler("result")

        def run_js_ready(webview, result, _user_data):
            try:
                js_res = webview.evaluate_javascript_finish(result)
                print(f"run-js return value: {js_res.to_json(2)}")
            except GLib.GError as e:
                sys.stderr.write(f"run-js error: {e.message}\n")
                set_result(str(e))

        # zero the current state, to ensure that the result triggers an Actions.Changed signal
        action.set_state(GLib.Variant.new_string(""))

        uc_manager = self.webview.get_user_content_manager()
        on_result_handler = uc_manager.connect(
            "script-message-received::result",
            lambda _mgr, result: set_result(result.to_string()))
        uc_manager.register_script_message_handler("result")

        # wait for loads to complete, to avoid races and ensure that it did not fail
        on_load_handler = self.webview.connect(
            "load-changed",
            lambda webview, event: set_result("page-load") if event == WebKit.LoadEvent.FINISHED else None)

        self.webview.evaluate_javascript(parameter.get_string(), len(parameter.get_string()),
                                         None, None, None, run_js_ready, None)


class CockpitClient(Gtk.Application):
    def __init__(self):
        super().__init__(application_id='org.cockpit_project.CockpitClient')

        self.add_main_option('no-ui', 0, GLib.OptionFlags.NONE, GLib.OptionArg.NONE,
                             'Show only a window with a webview')
        self.add_main_option('external-ws', 0, GLib.OptionFlags.NONE, GLib.OptionArg.STRING,
                             'Connect to existing cockpit-ws on the given URL')
        self.add_main_option('disable-uniqueness', 0, GLib.OptionFlags.NONE, GLib.OptionArg.NONE,
                             'Disable GApplication single-instance mode')
        self.add_main_option('enable-run-js', 0, GLib.OptionFlags.HIDDEN, GLib.OptionArg.NONE,
                             'Enable run-js action for testing')
        self.add_main_option('wildly-insecure', 0, GLib.OptionFlags.HIDDEN, GLib.OptionArg.NONE,
                             '***DANGEROUS*** Allow anyone to use your ssh credentials')

    def do_startup(self):
        Gtk.Application.do_startup(self)

        Adw.StyleManager.get_default().set_color_scheme(Adw.ColorScheme.PREFER_LIGHT)

        # .add_action_entries() binding is broken for GApplication
        # https://gitlab.gnome.org/GNOME/pygobject/-/issues/426
        Gio.ActionMap.add_action_entries(self, [
            ('new-window', self.new_window),
            ('quit', self.quit_action),
            ('open-path', self.open_path, 's'),
        ])

        self.set_accels_for_action("app.new-window", ["<Ctrl>N"])
        self.set_accels_for_action("win.reload", ["<Primary>r"])
        self.set_accels_for_action("win.reload-bypass-cache", ["<Primary><Shift>r"])
        self.set_accels_for_action("win.go-back", ["<Alt>Left"])
        self.set_accels_for_action("win.go-forward", ["<Alt>Right"])
        self.set_accels_for_action("win.zoom::in", ["<Primary>equal"])
        self.set_accels_for_action("win.zoom::out", ["<Primary>minus"])
        self.set_accels_for_action("win.zoom::default", ["<Primary>0"])
        self.set_accels_for_action("win.open-inspector", ["<Primary><Shift>i", "F12"])

        self.network_session = WebKit.NetworkSession.get_default()
        self.network_session.set_proxy_settings(WebKit.NetworkProxyMode.NO_PROXY)

        cookiesFile = os.path.join(get_user_state_dir(), "cockpit-client", "cookies.txt")
        cookie_manager = self.network_session.get_cookie_manager()
        cookie_manager.set_persistent_storage(cookiesFile, WebKit.CookiePersistentStorage.TEXT)

        self.uri = self.ws.start()

    def new_window(self, *args):
        self.activate()

    def quit_action(self, *args):
        self.quit()

    def open_path(self, action, parameter, *args):
        CockpitClientWindow(self, self.uri + parameter.get_string()).present()

    def do_activate(self):
        CockpitClientWindow(self, self.uri).present()

    def do_shutdown(self):
        self.ws.stop()

        Gtk.Application.do_shutdown(self)

    def do_handle_local_options(self, options):
        self.no_ui = options.lookup_value('no-ui') is not None

        if options.lookup_value('disable-uniqueness'):
            self.set_flags(Gio.ApplicationFlags.NON_UNIQUE)

        self.enable_run_js = bool(options.lookup_value('enable-run-js'))

        if flatpak_id := os.getenv('FLATPAK_ID'):
            self.set_application_id(flatpak_id)

        if external_ws := options.lookup_value('external-ws'):
            self.ws = ExternalCockpitWs(external_ws.get_string())

        elif self.safely_sandboxed() or options.lookup_value('wildly-insecure'):
            self.ws = InternalCockpitWs()

        else:
            logging.error('Unable to detect any sandboxing: refusing to spawn cockpit-ws')
            return 1

        return -1

    def safely_sandboxed(self):
        flatpak_info = configparser.ConfigParser()

        if not flatpak_info.read('/.flatpak-info'):
            return False

        shared = flatpak_info.get('Context', 'Shared', fallback='').lower().split(';')
        if 'network' in shared:
            return False

        return True


class ExternalCockpitWs:
    def __init__(self, uri):
        self.uri = uri

    def start(self):
        return self.uri

    def stop(self):
        pass


class InternalCockpitWs:
    def start(self):
        self.write_config()
        host = '127.0.0.90'
        port = '9090'
        self.ws = subprocess.Popen(
            [f'{libexecdir}/cockpit-ws', f'--address={host}', f'--port={port}'],
            preexec_fn=self.preexec_fn
        )

        return f'http://{host}:{port}/'

    def preexec_fn(self):
        os.environ['XDG_CONFIG_DIRS'] = self.config_dir.name
        prctl(SET_PDEATHSIG, signal.SIGKILL)

    def stop(self):
        self.ws.kill()
        self.config_dir.cleanup()

    def write_config(self):
        self.config_dir = tempfile.TemporaryDirectory(prefix='cockpit-client-', suffix='-etc')
        os.mkdir(f'{self.config_dir.name}/cockpit')
        with open(f'{self.config_dir.name}/cockpit/cockpit.conf', 'x') as cockpit_conf:
            # avoid putting strings here that ought to be translated
            config = """
                # dynamically generated by cockpit-client
                [WebService]
                X-For-CockpitClient = true
                LoginTo = true

                [Ssh-Login]
                Command = /usr/bin/env python3 -m cockpit.beiboot
                """

            cockpit_conf.write(textwrap.dedent(config))


if __name__ == "__main__":
    app = CockpitClient()
    app.run(sys.argv)