1
ˆ
ƒ Mƒ_•I|¥‰ÓM dœv ©ƒ&íK<óo¬ ¿Ã ª
Index of /FSS-BCZ/root/sys/dev/char/1:5/subsystem/null
Index of /FSS-BCZ/root/sys/dev/char/1:5/subsystem/null
$ ˆl–ßi~” jn-jqA
àAq·TÅ£Â
„_iïÁÀÄ .¬ #
# Copyright (C) 2022 Red Hat, Inc.
# SPDX-License-Identifier: GPL-3.0-or-later
import argparse
import asyncio
import contextlib
import grp
import json
import logging
import os
import pwd
import shlex
import socket
import stat
import subprocess
import sys
from typing import Dict, Iterable, List, Optional, Sequence, Tuple, Type
from cockpit._vendor.ferny import interaction_client
from cockpit._vendor.systemd_ctypes import bus, run_async
from . import polyfills
from ._version import __version__
from .channel import ChannelRoutingRule
from .channels import CHANNEL_TYPES
from .config import Config, Environment
from .internal_endpoints import EXPORTS
from .jsonutil import JsonError, JsonObject, JsonValue, get_dict
from .packages import BridgeConfig, Packages, PackagesListener
from .peer import PeersRoutingRule
from .remote import HostRoutingRule
from .router import Router
from .superuser import SuperuserRoutingRule
from .transports import StdioTransport
logger = logging.getLogger(__name__)
class InternalBus:
exportees: List[bus.Slot]
def __init__(self, exports: Iterable[Tuple[str, Type[bus.BaseObject]]]):
client_socket, server_socket = socket.socketpair()
self.client = bus.Bus.new(fd=client_socket.detach())
self.server = bus.Bus.new(fd=server_socket.detach(), server=True)
self.exportees = [self.server.add_object(path, cls()) for path, cls in exports]
def export(self, path: str, obj: bus.BaseObject) -> None:
self.exportees.append(self.server.add_object(path, obj))
class Bridge(Router, PackagesListener):
channels: ChannelRoutingRule
internal_bus: InternalBus
packages: Optional[Packages]
bridge_configs: Sequence[BridgeConfig]
args: argparse.Namespace
def __init__(self, args: argparse.Namespace):
self.internal_bus = InternalBus(EXPORTS)
self.bridge_configs = []
self.args = args
self.superuser_rule = SuperuserRoutingRule(self, privileged=args.privileged)
self.internal_bus.export('/superuser', self.superuser_rule)
self.internal_bus.export('/config', Config())
self.internal_bus.export('/environment', Environment())
self.peers_rule = PeersRoutingRule(self)
if args.beipack:
# Some special stuff for beipack
self.superuser_rule.set_configs((
BridgeConfig({
"privileged": True,
"spawn": ["sudo", "-k", "-A", "python3", "-ic", "# cockpit-bridge", "--privileged"],