137 lines
4.6 KiB
Python
137 lines
4.6 KiB
Python
# DBus Connection Management
|
|
#
|
|
# Copyright (C) 2022 Eric Callahan <arksine.code@gmail.com>
|
|
#
|
|
# This file may be distributed under the terms of the GNU GPLv3 license.
|
|
from __future__ import annotations
|
|
import os
|
|
import pathlib
|
|
import logging
|
|
import dbus_next
|
|
from dbus_next.aio import MessageBus, ProxyInterface
|
|
from dbus_next.constants import BusType
|
|
|
|
# Annotation imports
|
|
from typing import (
|
|
TYPE_CHECKING,
|
|
List,
|
|
Optional,
|
|
Any,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
from confighelper import ConfigHelper
|
|
|
|
STAT_PATH = "/proc/self/stat"
|
|
DOC_URL = (
|
|
"https://moonraker.readthedocs.io/en/latest/"
|
|
"installation/#policykit-permissions"
|
|
)
|
|
|
|
class DbusManager:
|
|
Variant = dbus_next.Variant
|
|
DbusError = dbus_next.errors.DBusError
|
|
def __init__(self, config: ConfigHelper) -> None:
|
|
self.server = config.get_server()
|
|
self.bus: Optional[MessageBus] = None
|
|
self.polkit: Optional[ProxyInterface] = None
|
|
self.warned: bool = False
|
|
st_path = pathlib.Path(STAT_PATH)
|
|
self.polkit_subject: List[Any] = []
|
|
if not st_path.is_file():
|
|
return
|
|
proc_data = st_path.read_text()
|
|
start_clk_ticks = int(proc_data.split()[21])
|
|
self.polkit_subject = [
|
|
"unix-process",
|
|
{
|
|
"pid": dbus_next.Variant("u", os.getpid()),
|
|
"start-time": dbus_next.Variant("t", start_clk_ticks)
|
|
}
|
|
]
|
|
|
|
def is_connected(self) -> bool:
|
|
return self.bus is not None and self.bus.connected
|
|
|
|
async def component_init(self) -> None:
|
|
try:
|
|
self.bus = MessageBus(bus_type=BusType.SYSTEM)
|
|
await self.bus.connect()
|
|
except Exception:
|
|
logging.info("Unable to Connect to D-Bus")
|
|
return
|
|
# Make sure that all required actions are register
|
|
try:
|
|
self.polkit = await self.get_interface(
|
|
"org.freedesktop.PolicyKit1",
|
|
"/org/freedesktop/PolicyKit1/Authority",
|
|
"org.freedesktop.PolicyKit1.Authority")
|
|
except self.DbusError:
|
|
self.server.add_warning(
|
|
"Unable to find DBus PolKit Interface, this suggests PolKit "
|
|
"is not installed on your OS.")
|
|
|
|
async def check_permission(self,
|
|
action: str,
|
|
err_msg: str = ""
|
|
) -> bool:
|
|
if self.polkit is None:
|
|
return False
|
|
try:
|
|
ret = await self.polkit.call_check_authorization( # type: ignore
|
|
self.polkit_subject, action, {}, 0, "")
|
|
except Exception as e:
|
|
self._check_warned()
|
|
self.server.add_warning(
|
|
f"Error checking authorization for action [{action}]: {e}. "
|
|
"This suggests that a dependency is not installed or "
|
|
f"up to date. {err_msg}.")
|
|
return False
|
|
if not ret[0]:
|
|
self._check_warned()
|
|
self.server.add_warning(
|
|
"Moonraker not authorized for PolicyKit action: "
|
|
f"[{action}], {err_msg}")
|
|
return ret[0]
|
|
|
|
def _check_warned(self):
|
|
if not self.warned:
|
|
self.server.add_warning(
|
|
f"PolKit warnings detected. See {DOC_URL} for instructions "
|
|
"on how to resolve.")
|
|
self.warned = True
|
|
|
|
async def get_interface(self,
|
|
bus_name: str,
|
|
bus_path: str,
|
|
interface_name: str
|
|
) -> ProxyInterface:
|
|
ret = await self.get_interfaces(bus_name, bus_path,
|
|
[interface_name])
|
|
return ret[0]
|
|
|
|
async def get_interfaces(self,
|
|
bus_name: str,
|
|
bus_path: str,
|
|
interface_names: List[str]
|
|
) -> List[ProxyInterface]:
|
|
if self.bus is None:
|
|
raise self.server.error("Bus not avaialable")
|
|
interfaces: List[ProxyInterface] = []
|
|
introspection = await self.bus.introspect(bus_name, bus_path)
|
|
proxy_obj = self.bus.get_proxy_object(bus_name, bus_path,
|
|
introspection)
|
|
for ifname in interface_names:
|
|
intf = proxy_obj.get_interface(ifname)
|
|
interfaces.append(intf)
|
|
return interfaces
|
|
|
|
async def close(self):
|
|
if self.bus is not None and self.bus.connected:
|
|
self.bus.disconnect()
|
|
await self.bus.wait_for_disconnect()
|
|
|
|
|
|
def load_component(config: ConfigHelper) -> DbusManager:
|
|
return DbusManager(config)
|