我们从Python开源项目中，提取了以下23个代码示例，用于说明如何使用dbus.DBusException()。
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)
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 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_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) #
