Gecko [ www.agenciac4designer.com ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
_vendor	[ DIR ]	drwxr-xr-x
channels	[ DIR ]	drwxr-xr-x
data	[ DIR ]	drwxr-xr-x
__init__.py	113 B	-rw-r--r--
_version.py	22 B	-rw-r--r--
beiboot.py	21.22 KB	-rw-r--r--
beipack.py	2.38 KB	-rw-r--r--
bridge.py	11.67 KB	-rw-r--r--
channel.py	21.71 KB	-rw-r--r--
config.py	2.76 KB	-rw-r--r--
internal_endpoints.py	5.34 KB	-rw-r--r--
jsonutil.py	6.81 KB	-rw-r--r--
osinfo.py	1019 B	-rw-r--r--
packages.py	23.14 KB	-rw-r--r--
peer.py	11.96 KB	-rw-r--r--
polkit.py	8.43 KB	-rw-r--r--
polyfills.py	1.66 KB	-rw-r--r--
protocol.py	8.91 KB	-rw-r--r--
remote.py	8.32 KB	-rw-r--r--
router.py	9.47 KB	-rw-r--r--
samples.py	16.41 KB	-rw-r--r--
superuser.py	12.09 KB	-rw-r--r--
transports.py	17.49 KB	-rw-r--r--

Code Editor : beipack.py

#
# Copyright (C) 2023 Red Hat, Inc.
# SPDX-License-Identifier: GPL-3.0-or-later

import logging
import lzma
from typing import Sequence, Tuple

from cockpit._vendor import ferny
from cockpit._vendor.bei import beipack

from .data import read_cockpit_data_file
from .peer import Peer, PeerError

logger = logging.getLogger(__name__)

def get_bridge_beipack_xz() -> Tuple[str, bytes]:
    try:
        bridge_beipack_xz = read_cockpit_data_file('cockpit-bridge.beipack.xz')
        logger.debug('Got pre-built cockpit-bridge.beipack.xz')
    except FileNotFoundError:
        logger.debug('Pre-built cockpit-bridge.beipack.xz; building our own.')
        # beipack ourselves
        cockpit_contents = beipack.collect_module('cockpit', recursive=True)
        bridge_beipack = beipack.pack(cockpit_contents, entrypoint='cockpit.bridge:main', args='beipack=True')
        bridge_beipack_xz = lzma.compress(bridge_beipack.encode())
        logger.debug('  ... done!')

return 'cockpit/data/cockpit-bridge.beipack.xz', bridge_beipack_xz

class BridgeBeibootHelper(ferny.InteractionHandler):
    # ferny.InteractionHandler ClassVar
    commands = ['beiboot.provide', 'beiboot.exc']

peer: Peer
    payload: bytes
    steps: Sequence[Tuple[str, Sequence[object]]]

def __init__(self, peer: Peer, args: Sequence[str] = ()) -> None:
        filename, payload = get_bridge_beipack_xz()

self.peer = peer
        self.payload = payload
        self.steps = (('boot_xz', (filename, len(payload), tuple(args))),)

async def run_command(self, command: str, args: 'tuple[object, ...]', fds: 'list[int]', stderr: str) -> None:
        logger.debug('Got ferny request %s %s %s %s', command, args, fds, stderr)
        if command == 'beiboot.provide':
            try:
                size, = args
                assert size == len(self.payload)
            except (AssertionError, ValueError) as exc:
                raise PeerError('internal-error', message=f'ferny interaction error {exc!s}') from exc

assert self.peer.transport is not None
            logger.debug('Writing %d bytes of payload', len(self.payload))
            self.peer.transport.write(self.payload)
        elif command == 'beiboot.exc':
            raise PeerError('internal-error', message=f'Remote exception: {args[0]}')
        else:
            raise PeerError('internal-error', message=f'Unexpected ferny interaction command {command}')

${this.title}

Code Editor : beipack.py