def _check_permission(self, sender, action):
'''
Verifies if the specified action is permitted, and raises
an AccessDeniedException if not.
The caller should use ObtainAuthorization() to get permission.
'''
try:
if sender:
kit = dbus.SystemBus().get_object('org.freedesktop.PolicyKit1', '/org/freedesktop/PolicyKit1/Authority')
kit = dbus.Interface(kit, 'org.freedesktop.PolicyKit1.Authority')
(granted, _, details) = kit.CheckAuthorization(
('system-bus-name', {'name': sender}),
action, {}, dbus.UInt32(1), '', timeout=600)
if not granted:
raise AccessDeniedException('Session not authorized by PolicyKit')
except AccessDeniedException:
raise
except dbus.DBusException, ex:
raise AccessDeniedException(ex.message)
python类DBusException()的实例源码
def create_dbus_server(cls, session_bus=False):
'''Return a D-BUS server backend instance.
Normally this connects to the system bus. Set session_bus to True to
connect to the session bus (for testing).
'''
backend = Backend()
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
if session_bus:
backend.bus = dbus.SessionBus()
backend.enforce_polkit = False
else:
backend.bus = dbus.SystemBus()
try:
backend.dbus_name = dbus.service.BusName(DBUS_BUS_NAME, backend.bus)
except dbus.exceptions.DBusException as msg:
logging.error("Exception when spawning dbus service")
logging.error(msg)
return None
return backend
#
# Internal methods
#
def backend(self):
'''Return D-BUS backend client interface.
This gets initialized lazily.
'''
if self._dbus_iface is None:
try:
bus = dbus.SystemBus()
self._dbus_iface = dbus.Interface(bus.get_object(DBUS_BUS_NAME,
'/RecoveryMedia'),
DBUS_INTERFACE_NAME)
except dbus.DBusException as msg:
self.dbus_exception_handler(msg)
sys.exit(1)
except Exception as msg:
self.show_alert(Gtk.MessageType.ERROR, "Exception", str(msg),
transient_for=self.tool_widgets.get_object('tool_selector'))
return self._dbus_iface
def __init__(self, services=[], filters=[avahi.LOOKUP_RESULT_LOCAL],
interface=avahi.IF_UNSPEC, protocol=avahi.PROTO_INET):
GObject.GObject.__init__(self)
self.filters = filters
self.services = services
self.interface = interface
self.protocol = protocol
try:
self.system_bus = dbus.SystemBus()
self.system_bus.add_signal_receiver(
self.avahi_dbus_connect_cb, "NameOwnerChanged", "org.freedesktop.DBus", arg0="org.freedesktop.Avahi")
except dbus.DBusException as e:
logger.error("Error Owning name on D-Bus: %s", e)
sys.exit(1)
self.db = ServiceTypeDatabase()
self.service_browsers = {}
self.started = False
def del_service_type(self, interface, protocol, type, domain):
service = (interface, protocol, type, domain)
if service not in self.service_browsers:
return
sb = self.service_browsers[service]
try:
sb.Free()
except dbus.DBusException:
pass
del self.service_browsers[service]
# delete the sub menu of service_type
if type in self.zc_types:
self.service_menu.remove(self.zc_types[type].get_attach_widget())
del self.zc_types[type]
if len(self.zc_types) == 0:
self.add_no_services_menuitem()
def connected(self, service):
if self.handle >= 0:
return True
bus = dbus.SessionBus()
wId = 0
try:
remote_obj = bus.get_object(self.bus_name, self.object_path)
self.iface = dbus.Interface(remote_obj, 'org.kde.KWallet')
self.handle = self.iface.open(
self.iface.networkWallet(), wId, self.appid)
except dbus.DBusException:
self.handle = -1
if self.handle < 0:
return False
self._migrate(service)
return True
def update_xapian_index(self):
self._logger.debug("update_xapian_index")
system_bus = get_dbus_bus()
# axi is optional, so just do nothing if its not installed
try:
axi = dbus.Interface(
system_bus.get_object("org.debian.AptXapianIndex", "/"),
"org.debian.AptXapianIndex")
except dbus.DBusException as e:
self._logger.warning("axi can not be updated '%s'" % e)
return
axi.connect_to_signal("UpdateFinished", self._axi_finished)
# we don't really care for updates at this point
#axi.connect_to_signal("UpdateProgress", progress)
try:
# first arg is force, second update_only
axi.update_async(True, False)
except:
self._logger.warning("could not update axi")
def _on_trans_error(self, error, pkgname=None):
self._logger.warn("_on_trans_error: %s", error)
# re-enable the action button again if anything went wrong
result = TransactionFinishedResult(None, False)
result.pkgname = pkgname
# clean up pending transactions
if pkgname and pkgname in self.pending_transactions:
del self.pending_transactions[pkgname]
self.emit("transaction-stopped", result)
if isinstance(error, dbus.DBusException):
name = error.get_dbus_name()
if name in ["org.freedesktop.PolicyKit.Error.NotAuthorized",
"org.freedesktop.DBus.Error.NoReply"]:
pass
else:
self._logger.exception("_on_trans_error")
#raise error
def update_xapian_index(self):
self._logger.debug("update_xapian_index")
system_bus = get_dbus_bus()
# axi is optional, so just do nothing if its not installed
try:
axi = dbus.Interface(
system_bus.get_object("org.debian.AptXapianIndex", "/"),
"org.debian.AptXapianIndex")
except dbus.DBusException as e:
self._logger.warning("axi can not be updated '%s'" % e)
return
axi.connect_to_signal("UpdateFinished", self._axi_finished)
# we don't really care for updates at this point
#axi.connect_to_signal("UpdateProgress", progress)
try:
# first arg is force, second update_only
axi.update_async(True, False)
except:
self._logger.warning("could not update axi")
def _on_trans_error(self, error, pkgname=None):
self._logger.warn("_on_trans_error: %s", error)
# re-enable the action button again if anything went wrong
result = TransactionFinishedResult(None, False)
result.pkgname = pkgname
# clean up pending transactions
if pkgname and pkgname in self.pending_transactions:
del self.pending_transactions[pkgname]
self.emit("transaction-stopped", result)
if isinstance(error, dbus.DBusException):
name = error.get_dbus_name()
if name in ["org.freedesktop.PolicyKit.Error.NotAuthorized",
"org.freedesktop.DBus.Error.NoReply"]:
pass
else:
self._logger.exception("_on_trans_error")
#raise error
def top_button_clicked(self, widget):
"""Callback for a button pushed in the main UI"""
#Restore Media Button / Driver install button
if widget == self.tool_widgets.get_object('build_os_media_button') or \
widget == self.tool_widgets.get_object('install_drivers_button'):
self.tool_widgets.get_object('tool_selector').set_sensitive(False)
#continue
return True
#Reboot action
else:
tool_selector = self.tool_widgets.get_object('tool_selector')
tool_selector.set_sensitive(False)
#Restore System Button
if widget == self.tool_widgets.get_object('restore_system_button'):
try:
dbus_sync_call_signal_wrapper(self.backend(),
"enable_boot_to_restore",
{},
False)
proc = subprocess.Popen(["gnome-session-quit", "--reboot"])
self.destroy()
except dbus.DBusException as msg:
self.dbus_exception_handler(msg)
tool_selector.set_sensitive(True)
return False
def cleanup_backend(self, widget=None, data=None):
'''cleans up the backend process'''
try:
if self._dbus_iface is not None:
self.backend().request_exit()
except dbus.DBusException as msg:
if hasattr(msg, '_dbus_error_name') and msg.get_dbus_name() == \
'org.freedesktop.DBus.Error.ServiceUnknown':
pass
else:
print("%s when closing DBus service from %s (data: %s)" %
(str(msg), widget, data))
def _register(self):
try:
manager = dbus.Interface(
self._bus.get_object("org.bluez", "/org/bluez"),
"org.bluez.AgentManager1")
manager.RegisterAgent(self.path, self.capability)
manager.RequestDefaultAgent(self.path)
except dbus.exceptions.DBusException as error:
print_error(str(error) + "\n")
return False
return True
def _unregister(self):
try:
manager = dbus.Interface(
self._bus.get_object("org.bluez", "/org/bluez"),
"org.bluez.AgentManager1")
manager.UnregisterAgent(self.path)
except dbus.exceptions.DBusException:
pass
def set_dns_systemd_resolved(lease):
# NOTE: if systemd-resolved is not already running, we might not want to
# run it in case there's specific system configuration for other resolvers
ipr = IPRoute()
index = ipr.link_lookup(ifname=lease.interface)[0]
# Construct the argument to pass to DBUS.
# the equivalent argument for:
# busctl call org.freedesktop.resolve1 /org/freedesktop/resolve1 \
# org.freedesktop.resolve1.Manager SetLinkDNS 'ia(iay)' 2 1 2 4 1 2 3 4
# is SetLinkDNS(2, [(2, [8, 8, 8, 8])]_
iay = [(2, [int(b) for b in ns.split('.')])
for ns in lease.name_server.split()]
# if '.' in ns
# else (10, [ord(x) for x in
# socket.inet_pton(socket.AF_INET6, ns)])
bus = SystemBus()
resolved = bus.get_object('org.freedesktop.resolve1',
'/org/freedesktop/resolve1')
manager = Interface(resolved,
dbus_interface='org.freedesktop.resolve1.Manager')
try:
manager.SetLinkDNS(index, iay)
return True
except DBusException as e:
logger.error(e)
return False
def priority(cls):
if 'dbus' not in globals():
raise RuntimeError('python-dbus not installed')
try:
bus = dbus.SessionBus()
except dbus.DBusException as exc:
raise RuntimeError(exc.get_dbus_message())
try:
bus.get_object(cls.bus_name, cls.object_path)
except dbus.DBusException:
tmpl = 'cannot connect to {bus_name}'
msg = tmpl.format(bus_name=cls.bus_name)
raise RuntimeError(msg)
return 4.9
def setup_dbus_or_bring_other_instance_to_front(self, args):
"""
This sets up a dbus listener
"""
try:
bus = dbus.SessionBus()
except:
LOG.exception("could not initiate dbus")
return
# if there is another Softwarecenter running bring it to front
# and exit, otherwise install the dbus controller
try:
proxy_obj = bus.get_object('com.ubuntu.Softwarecenter',
'/com/ubuntu/Softwarecenter')
iface = dbus.Interface(proxy_obj, 'com.ubuntu.SoftwarecenterIFace')
if args:
res = iface.bringToFront(args)
else:
# None can not be transported over dbus
res = iface.bringToFront('nothing-to-show')
# ensure that the running s-c is working
if res is not True:
LOG.info("found a running software-center on dbus, "
"reconnecting")
sys.exit()
except dbus.DBusException:
bus_name = dbus.service.BusName('com.ubuntu.Softwarecenter', bus)
self.dbusControler = SoftwarecenterDbusController(self, bus_name)
def get_transaction(self, tid):
""" synchroneously return a transaction """
try:
trans = client.get_transaction(tid)
return AptdaemonTransaction(trans)
except dbus.DBusException:
pass
def _on_lowlevel_transactions_changed(self, watcher, current, pending):
# cleanup progress signal (to be sure to not leave dbus
# matchers around)
if self._progress_signal:
GObject.source_remove(self._progress_signal)
self._progress_signal = None
# attach progress-changed signal for current transaction
if current:
try:
trans = client.get_transaction(current)
self._progress_signal = trans.connect("progress-changed",
self._on_progress_changed)
except dbus.DBusException:
pass
# now update pending transactions
self.pending_transactions.clear()
for tid in [current] + pending:
if not tid:
continue
try:
trans = client.get_transaction(tid,
error_handler=lambda x: True)
except dbus.DBusException:
continue
trans_progress = TransactionProgress(trans)
try:
self.pending_transactions[trans_progress.pkgname] = \
trans_progress
except KeyError:
# if its not a transaction from us (sc_pkgname) still
# add it with the tid as key to get accurate results
# (the key of pending_transactions is never directly
# exposed in the UI)
self.pending_transactions[trans.tid] = trans_progress
# emit signal
self.inject_fake_transactions_and_emit_changed_signal()
def get_transaction(self, tid):
""" synchroneously return a transaction """
try:
trans = client.get_transaction(tid)
return AptdaemonTransaction(trans)
except dbus.DBusException:
pass
def _on_lowlevel_transactions_changed(self, watcher, current, pending):
# cleanup progress signal (to be sure to not leave dbus
# matchers around)
if self._progress_signal:
GObject.source_remove(self._progress_signal)
self._progress_signal = None
# attach progress-changed signal for current transaction
if current:
try:
trans = client.get_transaction(current)
self._progress_signal = trans.connect("progress-changed",
self._on_progress_changed)
except dbus.DBusException:
pass
# now update pending transactions
self.pending_transactions.clear()
for tid in [current] + pending:
if not tid:
continue
try:
trans = client.get_transaction(tid,
error_handler=lambda x: True)
except dbus.DBusException:
continue
trans_progress = TransactionProgress(trans)
try:
self.pending_transactions[trans_progress.pkgname] = \
trans_progress
except KeyError:
# if its not a transaction from us (sc_pkgname) still
# add it with the tid as key to get accurate results
# (the key of pending_transactions is never directly
# exposed in the UI)
self.pending_transactions[trans.tid] = trans_progress
# emit signal
self.inject_fake_transactions_and_emit_changed_signal()
def __get_prop(obj, iface, prop):
"""Get object interface properties. Internal use only, dont touch."""
if not dbus:
log.warning("D-Bus module not found or not supported on this OS.")
return # Windows probably.
try:
return obj.Get(iface, prop, dbus_interface=dbus.PROPERTIES_IFACE)
except (dbus.DBusException, dbus.exceptions.DBusException) as err:
print(err)
return
def _check_polkit_privilege(self, sender, conn, privilege):
'''Verify that sender has a given PolicyKit privilege.
sender is the sender's (private) D-BUS name, such as ":1:42"
(sender_keyword in @dbus.service.methods). conn is
the dbus.Connection object (connection_keyword in
@dbus.service.methods). privilege is the PolicyKit privilege string.
This method returns if the caller is privileged, and otherwise throws a
PermissionDeniedByPolicy exception.
'''
if sender is None and conn is None:
# called locally, not through D-BUS
return
if not self.enforce_polkit:
# that happens for testing purposes when running on the session
# bus, and it does not make sense to restrict operations here
return
# get peer PID
if self.dbus_info is None:
self.dbus_info = dbus.Interface(conn.get_object('org.freedesktop.DBus',
'/org/freedesktop/DBus/Bus', False), 'org.freedesktop.DBus')
pid = self.dbus_info.GetConnectionUnixProcessID(sender)
# query PolicyKit
if self.polkit is None:
self.polkit = dbus.Interface(dbus.SystemBus().get_object(
'org.freedesktop.PolicyKit1', '/org/freedesktop/PolicyKit1/Authority', False),
'org.freedesktop.PolicyKit1.Authority')
try:
# we don't need is_challenge return here, since we call with AllowUserInteraction
(is_auth, unused, details) = self.polkit.CheckAuthorization(
('unix-process', {'pid': dbus.UInt32(pid, variant_level=1),
'start-time': dbus.UInt64(0, variant_level=1)}),
privilege, {'': ''}, dbus.UInt32(1), '', timeout=600)
except dbus.DBusException as msg:
if msg.get_dbus_name() == \
'org.freedesktop.DBus.Error.ServiceUnknown':
# polkitd timed out, connect again
self.polkit = None
return self._check_polkit_privilege(sender, conn, privilege)
else:
raise
if not is_auth:
logging.debug('_check_polkit_privilege: sender %s on connection %s pid %i is not authorized for %s: %s',
sender, conn, pid, privilege, str(details))
raise PermissionDeniedByPolicy(privilege)
#
# Internal API for calling from Handlers (not exported through D-BUS)
#
